Added server push method. Pushes to any/all connected clients with id='pushed' which clients can listen for
Add jsdoc documenation to source and did a few experiments with doc generatorstls
David Kebler
parent
6a8c2b1fca
commit
63beca4199
|
|
@ -0,0 +1,518 @@
|
|||
/*--------------------- Typography ----------------------------*/
|
||||
|
||||
@font-face {
|
||||
font-family: 'aller-light';
|
||||
src: url('public/fonts/aller-light.eot');
|
||||
src: url('public/fonts/aller-light.eot?#iefix') format('embedded-opentype'),
|
||||
url('public/fonts/aller-light.woff') format('woff'),
|
||||
url('public/fonts/aller-light.ttf') format('truetype');
|
||||
font-weight: normal;
|
||||
font-style: normal;
|
||||
}
|
||||
|
||||
@font-face {
|
||||
font-family: 'aller-bold';
|
||||
src: url('public/fonts/aller-bold.eot');
|
||||
src: url('public/fonts/aller-bold.eot?#iefix') format('embedded-opentype'),
|
||||
url('public/fonts/aller-bold.woff') format('woff'),
|
||||
url('public/fonts/aller-bold.ttf') format('truetype');
|
||||
font-weight: normal;
|
||||
font-style: normal;
|
||||
}
|
||||
|
||||
@font-face {
|
||||
font-family: 'roboto-black';
|
||||
src: url('public/fonts/roboto-black.eot');
|
||||
src: url('public/fonts/roboto-black.eot?#iefix') format('embedded-opentype'),
|
||||
url('public/fonts/roboto-black.woff') format('woff'),
|
||||
url('public/fonts/roboto-black.ttf') format('truetype');
|
||||
font-weight: normal;
|
||||
font-style: normal;
|
||||
}
|
||||
|
||||
/*--------------------- Layout ----------------------------*/
|
||||
html { height: 100%; }
|
||||
body {
|
||||
font-family: "aller-light";
|
||||
font-size: 14px;
|
||||
line-height: 18px;
|
||||
color: #30404f;
|
||||
margin: 0; padding: 0;
|
||||
height:100%;
|
||||
}
|
||||
#container { min-height: 100%; }
|
||||
|
||||
a {
|
||||
color: #000;
|
||||
}
|
||||
|
||||
b, strong {
|
||||
font-weight: normal;
|
||||
font-family: "aller-bold";
|
||||
}
|
||||
|
||||
p {
|
||||
margin: 15px 0 0px;
|
||||
}
|
||||
.annotation ul, .annotation ol {
|
||||
margin: 25px 0;
|
||||
}
|
||||
.annotation ul li, .annotation ol li {
|
||||
font-size: 14px;
|
||||
line-height: 18px;
|
||||
margin: 10px 0;
|
||||
}
|
||||
|
||||
h1, h2, h3, h4, h5, h6 {
|
||||
color: #112233;
|
||||
line-height: 1em;
|
||||
font-weight: normal;
|
||||
font-family: "roboto-black";
|
||||
text-transform: uppercase;
|
||||
margin: 30px 0 15px 0;
|
||||
}
|
||||
|
||||
h1 {
|
||||
margin-top: 40px;
|
||||
}
|
||||
h2 {
|
||||
font-size: 1.26em;
|
||||
}
|
||||
|
||||
hr {
|
||||
border: 0;
|
||||
background: 1px #ddd;
|
||||
height: 1px;
|
||||
margin: 20px 0;
|
||||
}
|
||||
|
||||
pre, tt, code {
|
||||
font-size: 12px; line-height: 16px;
|
||||
font-family: Menlo, Monaco, Consolas, "Lucida Console", monospace;
|
||||
margin: 0; padding: 0;
|
||||
}
|
||||
.annotation pre {
|
||||
display: block;
|
||||
margin: 0;
|
||||
padding: 7px 10px;
|
||||
background: #fcfcfc;
|
||||
-moz-box-shadow: inset 0 0 10px rgba(0,0,0,0.1);
|
||||
-webkit-box-shadow: inset 0 0 10px rgba(0,0,0,0.1);
|
||||
box-shadow: inset 0 0 10px rgba(0,0,0,0.1);
|
||||
overflow-x: auto;
|
||||
}
|
||||
.annotation pre code {
|
||||
border: 0;
|
||||
padding: 0;
|
||||
background: transparent;
|
||||
}
|
||||
|
||||
|
||||
blockquote {
|
||||
border-left: 5px solid #ccc;
|
||||
margin: 0;
|
||||
padding: 1px 0 1px 1em;
|
||||
}
|
||||
.sections blockquote p {
|
||||
font-family: Menlo, Consolas, Monaco, monospace;
|
||||
font-size: 12px; line-height: 16px;
|
||||
color: #999;
|
||||
margin: 10px 0 0;
|
||||
white-space: pre-wrap;
|
||||
}
|
||||
|
||||
ul.sections {
|
||||
list-style: none;
|
||||
padding:0 0 5px 0;;
|
||||
margin:0;
|
||||
}
|
||||
|
||||
/*
|
||||
Force border-box so that % widths fit the parent
|
||||
container without overlap because of margin/padding.
|
||||
|
||||
More Info : http://www.quirksmode.org/css/box.html
|
||||
*/
|
||||
ul.sections > li > div {
|
||||
-moz-box-sizing: border-box; /* firefox */
|
||||
-ms-box-sizing: border-box; /* ie */
|
||||
-webkit-box-sizing: border-box; /* webkit */
|
||||
-khtml-box-sizing: border-box; /* konqueror */
|
||||
box-sizing: border-box; /* css3 */
|
||||
}
|
||||
|
||||
|
||||
/*---------------------- Jump Page -----------------------------*/
|
||||
#jump_to, #jump_page {
|
||||
margin: 0;
|
||||
background: white;
|
||||
-webkit-box-shadow: 0 0 25px #777; -moz-box-shadow: 0 0 25px #777;
|
||||
-webkit-border-bottom-left-radius: 5px; -moz-border-radius-bottomleft: 5px;
|
||||
font: 16px Arial;
|
||||
cursor: pointer;
|
||||
text-align: right;
|
||||
list-style: none;
|
||||
}
|
||||
|
||||
#jump_to a {
|
||||
text-decoration: none;
|
||||
}
|
||||
|
||||
#jump_to a.large {
|
||||
display: none;
|
||||
}
|
||||
#jump_to a.small {
|
||||
font-size: 22px;
|
||||
font-weight: bold;
|
||||
color: #676767;
|
||||
}
|
||||
|
||||
#jump_to, #jump_wrapper {
|
||||
position: fixed;
|
||||
right: 0; top: 0;
|
||||
padding: 10px 15px;
|
||||
margin:0;
|
||||
}
|
||||
|
||||
#jump_wrapper {
|
||||
display: none;
|
||||
padding:0;
|
||||
}
|
||||
|
||||
#jump_to:hover #jump_wrapper {
|
||||
display: block;
|
||||
}
|
||||
|
||||
#jump_page_wrapper{
|
||||
position: fixed;
|
||||
right: 0;
|
||||
top: 0;
|
||||
bottom: 0;
|
||||
}
|
||||
|
||||
#jump_page {
|
||||
padding: 5px 0 3px;
|
||||
margin: 0 0 25px 25px;
|
||||
max-height: 100%;
|
||||
overflow: auto;
|
||||
}
|
||||
|
||||
#jump_page .source {
|
||||
display: block;
|
||||
padding: 15px;
|
||||
text-decoration: none;
|
||||
border-top: 1px solid #eee;
|
||||
}
|
||||
|
||||
#jump_page .source:hover {
|
||||
background: #f5f5ff;
|
||||
}
|
||||
|
||||
#jump_page .source:first-child {
|
||||
}
|
||||
|
||||
/*---------------------- Low resolutions (> 320px) ---------------------*/
|
||||
@media only screen and (min-width: 320px) {
|
||||
.pilwrap { display: none; }
|
||||
|
||||
ul.sections > li > div {
|
||||
display: block;
|
||||
padding:5px 10px 0 10px;
|
||||
}
|
||||
|
||||
ul.sections > li > div.annotation ul, ul.sections > li > div.annotation ol {
|
||||
padding-left: 30px;
|
||||
}
|
||||
|
||||
ul.sections > li > div.content {
|
||||
overflow-x:auto;
|
||||
-webkit-box-shadow: inset 0 0 5px #e5e5ee;
|
||||
box-shadow: inset 0 0 5px #e5e5ee;
|
||||
border: 1px solid #dedede;
|
||||
margin:5px 10px 5px 10px;
|
||||
padding-bottom: 5px;
|
||||
}
|
||||
|
||||
ul.sections > li > div.annotation pre {
|
||||
margin: 7px 0 7px;
|
||||
padding-left: 15px;
|
||||
}
|
||||
|
||||
ul.sections > li > div.annotation p tt, .annotation code {
|
||||
background: #f8f8ff;
|
||||
border: 1px solid #dedede;
|
||||
font-size: 12px;
|
||||
padding: 0 0.2em;
|
||||
}
|
||||
}
|
||||
|
||||
/*---------------------- (> 481px) ---------------------*/
|
||||
@media only screen and (min-width: 481px) {
|
||||
#container {
|
||||
position: relative;
|
||||
}
|
||||
body {
|
||||
background-color: #F5F5FF;
|
||||
font-size: 15px;
|
||||
line-height: 21px;
|
||||
}
|
||||
pre, tt, code {
|
||||
line-height: 18px;
|
||||
}
|
||||
p, ul, ol {
|
||||
margin: 0 0 15px;
|
||||
}
|
||||
|
||||
|
||||
#jump_to {
|
||||
padding: 5px 10px;
|
||||
}
|
||||
#jump_wrapper {
|
||||
padding: 0;
|
||||
}
|
||||
#jump_to, #jump_page {
|
||||
font: 10px Arial;
|
||||
text-transform: uppercase;
|
||||
}
|
||||
#jump_page .source {
|
||||
padding: 5px 10px;
|
||||
}
|
||||
#jump_to a.large {
|
||||
display: inline-block;
|
||||
}
|
||||
#jump_to a.small {
|
||||
display: none;
|
||||
}
|
||||
|
||||
|
||||
|
||||
#background {
|
||||
position: absolute;
|
||||
top: 0; bottom: 0;
|
||||
width: 350px;
|
||||
background: #fff;
|
||||
border-right: 1px solid #e5e5ee;
|
||||
z-index: -1;
|
||||
}
|
||||
|
||||
ul.sections > li > div.annotation ul, ul.sections > li > div.annotation ol {
|
||||
padding-left: 40px;
|
||||
}
|
||||
|
||||
ul.sections > li {
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
ul.sections > li > div {
|
||||
display: inline-block;
|
||||
}
|
||||
|
||||
ul.sections > li > div.annotation {
|
||||
max-width: 350px;
|
||||
min-width: 350px;
|
||||
min-height: 5px;
|
||||
padding: 13px;
|
||||
overflow-x: hidden;
|
||||
white-space: normal;
|
||||
vertical-align: top;
|
||||
text-align: left;
|
||||
}
|
||||
ul.sections > li > div.annotation pre {
|
||||
margin: 15px 0 15px;
|
||||
padding-left: 15px;
|
||||
}
|
||||
|
||||
ul.sections > li > div.content {
|
||||
padding: 13px;
|
||||
vertical-align: top;
|
||||
border: none;
|
||||
-webkit-box-shadow: none;
|
||||
box-shadow: none;
|
||||
}
|
||||
|
||||
.pilwrap {
|
||||
position: relative;
|
||||
display: inline;
|
||||
}
|
||||
|
||||
.pilcrow {
|
||||
font: 12px Arial;
|
||||
text-decoration: none;
|
||||
color: #454545;
|
||||
position: absolute;
|
||||
top: 3px; left: -20px;
|
||||
padding: 1px 2px;
|
||||
opacity: 0;
|
||||
-webkit-transition: opacity 0.2s linear;
|
||||
}
|
||||
.for-h1 .pilcrow {
|
||||
top: 47px;
|
||||
}
|
||||
.for-h2 .pilcrow, .for-h3 .pilcrow, .for-h4 .pilcrow {
|
||||
top: 35px;
|
||||
}
|
||||
|
||||
ul.sections > li > div.annotation:hover .pilcrow {
|
||||
opacity: 1;
|
||||
}
|
||||
}
|
||||
|
||||
/*---------------------- (> 1025px) ---------------------*/
|
||||
@media only screen and (min-width: 1025px) {
|
||||
|
||||
body {
|
||||
font-size: 16px;
|
||||
line-height: 24px;
|
||||
}
|
||||
|
||||
#background {
|
||||
width: 525px;
|
||||
}
|
||||
ul.sections > li > div.annotation {
|
||||
max-width: 525px;
|
||||
min-width: 525px;
|
||||
padding: 10px 25px 1px 50px;
|
||||
}
|
||||
ul.sections > li > div.content {
|
||||
padding: 9px 15px 16px 25px;
|
||||
}
|
||||
}
|
||||
|
||||
/*---------------------- Syntax Highlighting -----------------------------*/
|
||||
|
||||
td.linenos { background-color: #f0f0f0; padding-right: 10px; }
|
||||
span.lineno { background-color: #f0f0f0; padding: 0 5px 0 5px; }
|
||||
/*
|
||||
|
||||
github.com style (c) Vasily Polovnyov <vast@whiteants.net>
|
||||
|
||||
*/
|
||||
|
||||
pre code {
|
||||
display: block; padding: 0.5em;
|
||||
color: #000;
|
||||
background: #f8f8ff
|
||||
}
|
||||
|
||||
pre .hljs-comment,
|
||||
pre .hljs-template_comment,
|
||||
pre .hljs-diff .hljs-header,
|
||||
pre .hljs-javadoc {
|
||||
color: #408080;
|
||||
font-style: italic
|
||||
}
|
||||
|
||||
pre .hljs-keyword,
|
||||
pre .hljs-assignment,
|
||||
pre .hljs-literal,
|
||||
pre .hljs-css .hljs-rule .hljs-keyword,
|
||||
pre .hljs-winutils,
|
||||
pre .hljs-javascript .hljs-title,
|
||||
pre .hljs-lisp .hljs-title,
|
||||
pre .hljs-subst {
|
||||
color: #954121;
|
||||
/*font-weight: bold*/
|
||||
}
|
||||
|
||||
pre .hljs-number,
|
||||
pre .hljs-hexcolor {
|
||||
color: #40a070
|
||||
}
|
||||
|
||||
pre .hljs-string,
|
||||
pre .hljs-tag .hljs-value,
|
||||
pre .hljs-phpdoc,
|
||||
pre .hljs-tex .hljs-formula {
|
||||
color: #219161;
|
||||
}
|
||||
|
||||
pre .hljs-title,
|
||||
pre .hljs-id {
|
||||
color: #19469D;
|
||||
}
|
||||
pre .hljs-params {
|
||||
color: #00F;
|
||||
}
|
||||
|
||||
pre .hljs-javascript .hljs-title,
|
||||
pre .hljs-lisp .hljs-title,
|
||||
pre .hljs-subst {
|
||||
font-weight: normal
|
||||
}
|
||||
|
||||
pre .hljs-class .hljs-title,
|
||||
pre .hljs-haskell .hljs-label,
|
||||
pre .hljs-tex .hljs-command {
|
||||
color: #458;
|
||||
font-weight: bold
|
||||
}
|
||||
|
||||
pre .hljs-tag,
|
||||
pre .hljs-tag .hljs-title,
|
||||
pre .hljs-rules .hljs-property,
|
||||
pre .hljs-django .hljs-tag .hljs-keyword {
|
||||
color: #000080;
|
||||
font-weight: normal
|
||||
}
|
||||
|
||||
pre .hljs-attribute,
|
||||
pre .hljs-variable,
|
||||
pre .hljs-instancevar,
|
||||
pre .hljs-lisp .hljs-body {
|
||||
color: #008080
|
||||
}
|
||||
|
||||
pre .hljs-regexp {
|
||||
color: #B68
|
||||
}
|
||||
|
||||
pre .hljs-class {
|
||||
color: #458;
|
||||
font-weight: bold
|
||||
}
|
||||
|
||||
pre .hljs-symbol,
|
||||
pre .hljs-ruby .hljs-symbol .hljs-string,
|
||||
pre .hljs-ruby .hljs-symbol .hljs-keyword,
|
||||
pre .hljs-ruby .hljs-symbol .hljs-keymethods,
|
||||
pre .hljs-lisp .hljs-keyword,
|
||||
pre .hljs-tex .hljs-special,
|
||||
pre .hljs-input_number {
|
||||
color: #990073
|
||||
}
|
||||
|
||||
pre .hljs-builtin,
|
||||
pre .hljs-constructor,
|
||||
pre .hljs-built_in,
|
||||
pre .hljs-lisp .hljs-title {
|
||||
color: #0086b3
|
||||
}
|
||||
|
||||
pre .hljs-preprocessor,
|
||||
pre .hljs-pi,
|
||||
pre .hljs-doctype,
|
||||
pre .hljs-shebang,
|
||||
pre .hljs-cdata {
|
||||
color: #999;
|
||||
font-weight: bold
|
||||
}
|
||||
|
||||
pre .hljs-deletion {
|
||||
background: #fdd
|
||||
}
|
||||
|
||||
pre .hljs-addition {
|
||||
background: #dfd
|
||||
}
|
||||
|
||||
pre .hljs-diff .hljs-change {
|
||||
background: #0086b3
|
||||
}
|
||||
|
||||
pre .hljs-chunk {
|
||||
color: #aaa
|
||||
}
|
||||
|
||||
pre .hljs-tex .hljs-formula {
|
||||
opacity: 0.5;
|
||||
}
|
||||
|
|
@ -0,0 +1,494 @@
|
|||
<!DOCTYPE html>
|
||||
|
||||
<html>
|
||||
<head>
|
||||
<title>consumer.js</title>
|
||||
<meta http-equiv="content-type" content="text/html; charset=UTF-8">
|
||||
<meta name="viewport" content="width=device-width, target-densitydpi=160dpi, initial-scale=1.0; maximum-scale=1.0; user-scalable=0;">
|
||||
<link rel="stylesheet" media="all" href="../docco.css" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="container">
|
||||
<div id="background"></div>
|
||||
|
||||
<ul id="jump_to">
|
||||
<li>
|
||||
<a class="large" href="javascript:void(0);">Jump To …</a>
|
||||
<a class="small" href="javascript:void(0);">+</a>
|
||||
<div id="jump_wrapper">
|
||||
<div id="jump_page_wrapper">
|
||||
<div id="jump_page">
|
||||
|
||||
|
||||
<a class="source" href="consumer.html">
|
||||
./src/consumer.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="index.html">
|
||||
./src/index.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="json-stream.html">
|
||||
./src/json-stream.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="socket.html">
|
||||
./src/socket.js
|
||||
</a>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
</li>
|
||||
</ul>
|
||||
|
||||
<ul class="sections">
|
||||
|
||||
<li id="title">
|
||||
<div class="annotation">
|
||||
<h1>consumer.js</h1>
|
||||
</div>
|
||||
</li>
|
||||
|
||||
|
||||
|
||||
<li id="section-1">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-1">¶</a>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">import</span> { Socket } <span class="hljs-keyword">from</span> <span class="hljs-string">'net'</span>
|
||||
<span class="hljs-keyword">import</span> path <span class="hljs-keyword">from</span> <span class="hljs-string">'path'</span>
|
||||
<span class="hljs-keyword">import</span> btc <span class="hljs-keyword">from</span> <span class="hljs-string">'better-try-catch'</span>
|
||||
<span class="hljs-keyword">import</span> JsonStream <span class="hljs-keyword">from</span> <span class="hljs-string">'./json-stream'</span></pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-2">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-2">¶</a>
|
||||
</div>
|
||||
<p>import logger from ‘../../uci-logger/src/logger’</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">import</span> logger <span class="hljs-keyword">from</span> <span class="hljs-string">'@uci/logger'</span>
|
||||
|
||||
<span class="hljs-keyword">let</span> log = {}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-3">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-3">¶</a>
|
||||
</div>
|
||||
<p>TODO change default pipe dir for windows and mac os</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">const</span> DEFAULT_PIPE_DIR = (process.env.SOCKETS_DIR || <span class="hljs-string">'/tmp/UCI'</span>)
|
||||
<span class="hljs-keyword">const</span> DEFAULT_SOCKET_NAME = <span class="hljs-string">'uci-sock'</span>
|
||||
|
||||
<span class="hljs-keyword">export</span> <span class="hljs-keyword">default</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">Consumer</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">Socket</span> </span>{
|
||||
<span class="hljs-keyword">constructor</span> (opts={}) {
|
||||
<span class="hljs-keyword">super</span>()
|
||||
log = logger({<span class="hljs-attr">file</span>:<span class="hljs-string">'src/consumer.js'</span>,<span class="hljs-attr">class</span>:<span class="hljs-string">'Consumer'</span>,<span class="hljs-attr">name</span>:<span class="hljs-string">'socket'</span>,<span class="hljs-attr">id</span>:<span class="hljs-keyword">this</span>.id})
|
||||
<span class="hljs-keyword">this</span>.id = opts.id || opts.name || <span class="hljs-string">'socket:'</span>+ <span class="hljs-keyword">new</span> <span class="hljs-built_in">Date</span>().getTime()
|
||||
<span class="hljs-keyword">if</span> (!opts.path) {
|
||||
log.warn({<span class="hljs-attr">opts</span>:opts},<span class="hljs-string">'no host supplied using localhost...use named piped instead'</span>)
|
||||
opts.host = opts.host || <span class="hljs-string">'127.0.0.1'</span>
|
||||
opts.port = opts.port || <span class="hljs-number">8080</span>
|
||||
} <span class="hljs-keyword">else</span> {
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">typeof</span> opts.path === <span class="hljs-string">'boolean'</span>) opts.path = path.join(DEFAULT_PIPE_DIR,DEFAULT_SOCKET_NAME )
|
||||
<span class="hljs-keyword">if</span> (path.dirname(opts.path)===<span class="hljs-string">'.'</span>) opts.path = path.join(DEFAULT_PIPE_DIR,opts.path )
|
||||
}
|
||||
<span class="hljs-keyword">this</span>.opts=opts
|
||||
<span class="hljs-keyword">this</span>.keepAlive = <span class="hljs-string">'keepAlive'</span> <span class="hljs-keyword">in</span> opts ? opts.keepAlive : <span class="hljs-literal">true</span>
|
||||
<span class="hljs-keyword">this</span>._ready = <span class="hljs-literal">false</span>
|
||||
<span class="hljs-keyword">this</span>.timeout = opts.timeout || <span class="hljs-number">30</span>
|
||||
<span class="hljs-keyword">this</span>.wait = opts.wait || <span class="hljs-number">1</span>
|
||||
<span class="hljs-keyword">this</span>.stream = <span class="hljs-keyword">new</span> JsonStream()</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-4">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-4">¶</a>
|
||||
</div>
|
||||
<p>bind to class for other class functions</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">this</span>.connect = <span class="hljs-keyword">this</span>.connect.bind(<span class="hljs-keyword">this</span>)
|
||||
<span class="hljs-keyword">this</span>.__ready = <span class="hljs-keyword">this</span>.__ready.bind(<span class="hljs-keyword">this</span>)</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-5">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-5">¶</a>
|
||||
</div>
|
||||
<p>this.<em>write = this.</em>write.bind(this)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> }
|
||||
|
||||
<span class="hljs-keyword">async</span> connect () {
|
||||
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>( <span class="hljs-function">(<span class="hljs-params">resolve,reject</span>) =></span> {
|
||||
|
||||
<span class="hljs-keyword">const</span> connect = <span class="hljs-function"><span class="hljs-params">()</span> =></span> {
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>.opts.host ===<span class="hljs-string">'127.0.0.1'</span>) log.warn(<span class="hljs-string">'tcp consumer on same machine as host, use named Pipe(Unix) Socket Instead'</span>)
|
||||
log.info({<span class="hljs-attr">opts</span>:<span class="hljs-keyword">this</span>.opts},<span class="hljs-string">`attempting to connect <span class="hljs-subst">${<span class="hljs-keyword">this</span>.id}</span> to socket`</span>)
|
||||
<span class="hljs-keyword">super</span>.connect(<span class="hljs-keyword">this</span>.opts)
|
||||
}
|
||||
|
||||
<span class="hljs-keyword">let</span> reconnect = {}
|
||||
<span class="hljs-keyword">const</span> timeout = setTimeout(<span class="hljs-function"><span class="hljs-params">()</span> =></span>{
|
||||
clearTimeout(reconnect)
|
||||
log.fatal({<span class="hljs-attr">opts</span>:<span class="hljs-keyword">this</span>.opts},<span class="hljs-string">`unable to connect in <span class="hljs-subst">${<span class="hljs-keyword">this</span>.timeout}</span>s`</span>)
|
||||
reject({<span class="hljs-attr">opts</span>:<span class="hljs-keyword">this</span>.opts},<span class="hljs-string">`unable to connect to socket server in <span class="hljs-subst">${<span class="hljs-keyword">this</span>.timeout}</span>secs`</span>)
|
||||
}
|
||||
,<span class="hljs-keyword">this</span>.timeout*<span class="hljs-number">1000</span>)
|
||||
|
||||
<span class="hljs-keyword">this</span>.once(<span class="hljs-string">'connect'</span>, <span class="hljs-keyword">async</span> () => {
|
||||
clearTimeout(timeout)
|
||||
<span class="hljs-keyword">this</span>._listen()
|
||||
log.info({<span class="hljs-attr">opts</span>:<span class="hljs-keyword">this</span>.opts},<span class="hljs-string">'connected waiting for socket ready handshake'</span>)
|
||||
<span class="hljs-keyword">this</span>.setKeepAlive(<span class="hljs-keyword">this</span>.keepAlive,<span class="hljs-number">100</span>)
|
||||
<span class="hljs-keyword">let</span> [err, res] = <span class="hljs-keyword">await</span> btc(isReady).bind(<span class="hljs-keyword">this</span>)(<span class="hljs-keyword">this</span>.__ready, <span class="hljs-keyword">this</span>.wait, <span class="hljs-keyword">this</span>.timeout)
|
||||
<span class="hljs-keyword">if</span> (err) reject(err)
|
||||
log.info(<span class="hljs-string">'handshake done, authenticating'</span>)</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-6">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-6">¶</a>
|
||||
</div>
|
||||
<p>TODO authenticate here by encrypting a payload with private key and sending that.
|
||||
await btc(authenticate)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> resolve(res)
|
||||
})
|
||||
|
||||
<span class="hljs-keyword">this</span>.on(<span class="hljs-string">'error'</span>, <span class="hljs-keyword">async</span> (err) => {
|
||||
log.warn({<span class="hljs-attr">error</span>:err.code},<span class="hljs-string">`connect error <span class="hljs-subst">${err.code}</span>`</span>)
|
||||
<span class="hljs-keyword">if</span> (err.code === <span class="hljs-string">'EISCONN'</span>) {
|
||||
<span class="hljs-keyword">return</span> resolve(<span class="hljs-string">'ready'</span>)
|
||||
}
|
||||
|
||||
reconnect = setTimeout( <span class="hljs-function"><span class="hljs-params">()</span> =></span>{ connect()
|
||||
},<span class="hljs-keyword">this</span>.wait*<span class="hljs-number">1000</span> )
|
||||
|
||||
})
|
||||
|
||||
<span class="hljs-keyword">this</span>.on(<span class="hljs-string">'end'</span>, <span class="hljs-keyword">async</span> () => {
|
||||
log.warn(<span class="hljs-string">'socket (server) terminated unexpectantly'</span>)
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>.keepAlive) {
|
||||
log.info(<span class="hljs-string">'keep alive was set, so waiting on server to come online for reconnect'</span>)
|
||||
<span class="hljs-keyword">this</span>.destroy()
|
||||
<span class="hljs-keyword">this</span>.emit(<span class="hljs-string">'error'</span>, {<span class="hljs-attr">code</span>:<span class="hljs-string">'DISCONNECTED'</span>})
|
||||
}
|
||||
})
|
||||
|
||||
connect() <span class="hljs-comment">// initial connect request</span>
|
||||
|
||||
}) <span class="hljs-comment">//end promise</span>
|
||||
|
||||
}
|
||||
|
||||
|
||||
<span class="hljs-keyword">async</span> send(ipacket) {
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>( <span class="hljs-keyword">async</span> (resolve) => {</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-7">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-7">¶</a>
|
||||
</div>
|
||||
<p>need this for when multiple sends for different consumers use same packet instance</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">let</span> packet = <span class="hljs-built_in">Object</span>.assign({},ipacket)
|
||||
setTimeout(<span class="hljs-function"><span class="hljs-params">()</span> =></span> {resolve({<span class="hljs-attr">error</span>:<span class="hljs-string">'no response from socket in 10sec'</span>})},<span class="hljs-number">10000</span>)
|
||||
packet._header =
|
||||
{ <span class="hljs-attr">id</span>:<span class="hljs-built_in">Math</span>.random().toString().slice(<span class="hljs-number">2</span>), <span class="hljs-comment">// need this for when multiple sends for different consumers use same packet instanceack</span>
|
||||
sender:{ <span class="hljs-attr">name</span>:<span class="hljs-keyword">this</span>.name, <span class="hljs-attr">instanceID</span>:<span class="hljs-keyword">this</span>.id },
|
||||
<span class="hljs-attr">path</span>: <span class="hljs-keyword">this</span>.opts.path,
|
||||
<span class="hljs-attr">port</span>: <span class="hljs-keyword">this</span>.opts.port,
|
||||
<span class="hljs-attr">host</span>: <span class="hljs-keyword">this</span>.opts.host
|
||||
}
|
||||
<span class="hljs-keyword">let</span> [err, res] = <span class="hljs-keyword">await</span> btc(<span class="hljs-keyword">this</span>.stream.serialize)(packet)
|
||||
<span class="hljs-keyword">if</span> (err) resolve({<span class="hljs-attr">error</span>:<span class="hljs-string">'unable to serialize packet for sending'</span>, <span class="hljs-attr">packet</span>:packet})
|
||||
<span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>.__write(res)
|
||||
<span class="hljs-keyword">this</span>.once(packet._header.id,<span class="hljs-keyword">async</span> <span class="hljs-function"><span class="hljs-keyword">function</span>(<span class="hljs-params">reply</span>)</span>{
|
||||
<span class="hljs-keyword">let</span> res = <span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._packetProcess(reply)
|
||||
<span class="hljs-keyword">if</span> (!res) { <span class="hljs-comment">// if process was not promise returning like just logged to console</span>
|
||||
res = reply</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-8">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-8">¶</a>
|
||||
</div>
|
||||
<p>log.warn(‘consumer function was not promise returning further processing may be out of sequence’)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> }
|
||||
resolve(res)
|
||||
}) <span class="hljs-comment">//end listener</span>
|
||||
})
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-9">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-9">¶</a>
|
||||
</div>
|
||||
<p>TODO register alt stream processor (emit ‘message’ with JSON, serialize function, onData method for raw socket chucks)
|
||||
TODO register authenciation function (set up default)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre>
|
||||
registerPacketProcessor (func) {
|
||||
<span class="hljs-keyword">this</span>._packetProcess = func
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-10">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-10">¶</a>
|
||||
</div>
|
||||
<p>PRIVATE METHODS</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre>
|
||||
<span class="hljs-keyword">async</span> __write(packet) {</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-11">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-11">¶</a>
|
||||
</div>
|
||||
<p>timeout already set if sockect can’t be drained in 10 secs</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>(<span class="hljs-function"><span class="hljs-params">resolve</span> =></span> {
|
||||
<span class="hljs-keyword">const</span> cb = <span class="hljs-function"><span class="hljs-params">()</span> =></span> resolve(<span class="hljs-string">'packet written to consumer side socket stream '</span>)
|
||||
<span class="hljs-keyword">if</span> (!<span class="hljs-keyword">super</span>.write(packet)) {
|
||||
<span class="hljs-keyword">this</span>.once(<span class="hljs-string">'drain'</span>,cb )
|
||||
} <span class="hljs-keyword">else</span> {
|
||||
process.nextTick(cb)
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
|
||||
__ready() {<span class="hljs-keyword">return</span> <span class="hljs-keyword">this</span>._ready}
|
||||
|
||||
<span class="hljs-keyword">async</span> _listen () {
|
||||
log.info(<span class="hljs-string">'listening for incoming packets from socket'</span>)</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-12">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-12">¶</a>
|
||||
</div>
|
||||
<p>listen for pushed packets</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">this</span>.on(<span class="hljs-string">'pushed'</span>,<span class="hljs-keyword">async</span> <span class="hljs-function"><span class="hljs-keyword">function</span>(<span class="hljs-params">packet</span>)</span>{</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-13">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-13">¶</a>
|
||||
</div>
|
||||
<p>TODO do some extra security here?</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">let</span> res = <span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._packetProcess(packet)
|
||||
<span class="hljs-keyword">if</span> (!res) { <span class="hljs-comment">// if process was not promise returning like just logged to console</span></pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-14">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-14">¶</a>
|
||||
</div>
|
||||
<p>log.warn(‘consumer function was not promise returning’)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> }
|
||||
})</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-15">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-15">¶</a>
|
||||
</div>
|
||||
<p>listen on socket stream</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">this</span>.on(<span class="hljs-string">'data'</span>, <span class="hljs-keyword">this</span>.stream.onData)
|
||||
<span class="hljs-keyword">this</span>.stream.on(<span class="hljs-string">'message'</span>, messageProcess.bind(<span class="hljs-keyword">this</span>))
|
||||
|
||||
<span class="hljs-keyword">async</span> <span class="hljs-function"><span class="hljs-keyword">function</span> <span class="hljs-title">messageProcess</span> (<span class="hljs-params">packet</span>) </span>{</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-16">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-16">¶</a>
|
||||
</div>
|
||||
<p>console.log(‘incoming packet from socket’,packet)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">if</span> (packet._handshake) {
|
||||
<span class="hljs-keyword">this</span>._ready = <span class="hljs-literal">true</span>
|
||||
<span class="hljs-keyword">return</span> }</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-17">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-17">¶</a>
|
||||
</div>
|
||||
<p>TODO send back ack with consumer ID and authorization and wait
|
||||
when authorized drop through here to emit</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">this</span>.emit(packet._header.id, packet)
|
||||
}
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-18">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-18">¶</a>
|
||||
</div>
|
||||
<p>default packet process just a simple console logger. ignores any cmd: prop</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> _packetProcess (packet) {
|
||||
<span class="hljs-built_in">console</span>.log(<span class="hljs-string">'default consumer processor -- log packet from socket to console'</span>)
|
||||
<span class="hljs-built_in">console</span>.dir(packet)
|
||||
}
|
||||
|
||||
} <span class="hljs-comment">// end class</span></pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-19">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-19">¶</a>
|
||||
</div>
|
||||
<p>HELP FUNCTIONS
|
||||
wait until a passed ready function returns true</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-function"><span class="hljs-keyword">function</span> <span class="hljs-title">isReady</span>(<span class="hljs-params">ready, wait=<span class="hljs-number">30</span>, timeout=<span class="hljs-number">1000</span></span>) </span>{
|
||||
<span class="hljs-keyword">let</span> time = <span class="hljs-number">0</span>
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>(<span class="hljs-function">(<span class="hljs-params">resolve, reject</span>) =></span> {
|
||||
(<span class="hljs-function"><span class="hljs-keyword">function</span> <span class="hljs-title">waitReady</span>(<span class="hljs-params"></span>)</span>{
|
||||
<span class="hljs-keyword">if</span> (time > timeout) <span class="hljs-keyword">return</span> reject(<span class="hljs-string">`timeout waiting for socket ready handshake - <span class="hljs-subst">${timeout}</span>ms`</span>)
|
||||
<span class="hljs-keyword">if</span> (ready()) <span class="hljs-keyword">return</span> resolve(<span class="hljs-string">'ready'</span>)
|
||||
log.info(<span class="hljs-string">`waiting <span class="hljs-subst">${wait}</span>ms for handshake`</span>)
|
||||
time += wait
|
||||
setTimeout(waitReady, wait)
|
||||
})()
|
||||
})
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
</ul>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
<!DOCTYPE html>
|
||||
|
||||
<html>
|
||||
<head>
|
||||
<title>index.js</title>
|
||||
<meta http-equiv="content-type" content="text/html; charset=UTF-8">
|
||||
<meta name="viewport" content="width=device-width, target-densitydpi=160dpi, initial-scale=1.0; maximum-scale=1.0; user-scalable=0;">
|
||||
<link rel="stylesheet" media="all" href="../docco.css" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="container">
|
||||
<div id="background"></div>
|
||||
|
||||
<ul id="jump_to">
|
||||
<li>
|
||||
<a class="large" href="javascript:void(0);">Jump To …</a>
|
||||
<a class="small" href="javascript:void(0);">+</a>
|
||||
<div id="jump_wrapper">
|
||||
<div id="jump_page_wrapper">
|
||||
<div id="jump_page">
|
||||
|
||||
|
||||
<a class="source" href="consumer.html">
|
||||
./src/consumer.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="index.html">
|
||||
./src/index.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="json-stream.html">
|
||||
./src/json-stream.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="socket.html">
|
||||
./src/socket.js
|
||||
</a>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
</li>
|
||||
</ul>
|
||||
|
||||
<ul class="sections">
|
||||
|
||||
<li id="title">
|
||||
<div class="annotation">
|
||||
<h1>index.js</h1>
|
||||
</div>
|
||||
</li>
|
||||
|
||||
|
||||
|
||||
<li id="section-1">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-1">¶</a>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">import</span> Socket <span class="hljs-keyword">from</span> <span class="hljs-string">'./socket'</span>
|
||||
<span class="hljs-keyword">import</span> Consumer <span class="hljs-keyword">from</span> <span class="hljs-string">'./consumer'</span>
|
||||
|
||||
<span class="hljs-keyword">export</span> { Socket <span class="hljs-keyword">as</span> Socket }
|
||||
<span class="hljs-keyword">export</span> { Consumer <span class="hljs-keyword">as</span> Consumer }
|
||||
<span class="hljs-keyword">export</span> <span class="hljs-keyword">default</span> { Socket, Consumer }</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
</ul>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
|
|
@ -0,0 +1,199 @@
|
|||
<!DOCTYPE html>
|
||||
|
||||
<html>
|
||||
<head>
|
||||
<title>json-stream.js</title>
|
||||
<meta http-equiv="content-type" content="text/html; charset=UTF-8">
|
||||
<meta name="viewport" content="width=device-width, target-densitydpi=160dpi, initial-scale=1.0; maximum-scale=1.0; user-scalable=0;">
|
||||
<link rel="stylesheet" media="all" href="../docco.css" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="container">
|
||||
<div id="background"></div>
|
||||
|
||||
<ul id="jump_to">
|
||||
<li>
|
||||
<a class="large" href="javascript:void(0);">Jump To …</a>
|
||||
<a class="small" href="javascript:void(0);">+</a>
|
||||
<div id="jump_wrapper">
|
||||
<div id="jump_page_wrapper">
|
||||
<div id="jump_page">
|
||||
|
||||
|
||||
<a class="source" href="consumer.html">
|
||||
./src/consumer.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="index.html">
|
||||
./src/index.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="json-stream.html">
|
||||
./src/json-stream.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="socket.html">
|
||||
./src/socket.js
|
||||
</a>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
</li>
|
||||
</ul>
|
||||
|
||||
<ul class="sections">
|
||||
|
||||
<li id="title">
|
||||
<div class="annotation">
|
||||
<h1>json-stream.js</h1>
|
||||
</div>
|
||||
</li>
|
||||
|
||||
|
||||
|
||||
<li id="section-1">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-1">¶</a>
|
||||
</div>
|
||||
<p>adpated from <a href="https://github.com/sebastianseilund/node-json-socket">https://github.com/sebastianseilund/node-json-socket</a></p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre>
|
||||
<span class="hljs-keyword">import</span> {StringDecoder} <span class="hljs-keyword">from</span> <span class="hljs-string">'string_decoder'</span>
|
||||
<span class="hljs-keyword">import</span> EventEmitter <span class="hljs-keyword">from</span> <span class="hljs-string">'events'</span>
|
||||
<span class="hljs-keyword">import</span> btc <span class="hljs-keyword">from</span> <span class="hljs-string">'better-try-catch'</span>
|
||||
|
||||
<span class="hljs-keyword">const</span> decoder = <span class="hljs-keyword">new</span> StringDecoder()
|
||||
|
||||
<span class="hljs-keyword">export</span> <span class="hljs-keyword">default</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">JsonStream</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">EventEmitter</span></span>{
|
||||
<span class="hljs-keyword">constructor</span>(opts={}){
|
||||
<span class="hljs-keyword">super</span>()
|
||||
<span class="hljs-keyword">this</span>._contentLength = <span class="hljs-literal">null</span>
|
||||
<span class="hljs-keyword">this</span>._buffer = <span class="hljs-string">''</span>
|
||||
<span class="hljs-keyword">this</span>._delimeter = opts.delimiter || <span class="hljs-string">'#'</span>
|
||||
<span class="hljs-keyword">this</span>.onData = <span class="hljs-keyword">this</span>.onData.bind(<span class="hljs-keyword">this</span>)
|
||||
<span class="hljs-keyword">this</span>.serialize = <span class="hljs-keyword">this</span>.serialize.bind(<span class="hljs-keyword">this</span>)
|
||||
}
|
||||
|
||||
|
||||
onData (data) {</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-2">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-2">¶</a>
|
||||
</div>
|
||||
<p>console.log(‘a chunk arrived’, data)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> data = decoder.write(data)
|
||||
<span class="hljs-keyword">try</span> {
|
||||
<span class="hljs-keyword">this</span>._handleData(data)
|
||||
} <span class="hljs-keyword">catch</span> (e) {
|
||||
<span class="hljs-keyword">this</span>.emit(<span class="hljs-string">'error'</span>, { <span class="hljs-attr">error</span>: e })
|
||||
}
|
||||
}
|
||||
|
||||
<span class="hljs-keyword">async</span> serialize(message) {
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>( <span class="hljs-function">(<span class="hljs-params">resolve,reject</span>) =></span> {
|
||||
<span class="hljs-keyword">let</span> [err,messageData] = btc(<span class="hljs-built_in">JSON</span>.stringify)(message)
|
||||
<span class="hljs-keyword">if</span> (err) reject(err)
|
||||
<span class="hljs-keyword">let</span> [err2,length] = btc(Buffer.byteLength)(messageData, <span class="hljs-string">'utf8'</span>)
|
||||
<span class="hljs-keyword">if</span> (err2) reject(err2)
|
||||
<span class="hljs-keyword">let</span> data = length + <span class="hljs-keyword">this</span>._delimeter + messageData</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-3">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-3">¶</a>
|
||||
</div>
|
||||
<p>console.log(‘serialized’,data)</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> resolve(data)
|
||||
})
|
||||
}
|
||||
|
||||
_handleData (data) {
|
||||
<span class="hljs-keyword">this</span>._buffer += data
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>._contentLength == <span class="hljs-literal">null</span>) {
|
||||
<span class="hljs-keyword">var</span> i = <span class="hljs-keyword">this</span>._buffer.indexOf(<span class="hljs-keyword">this</span>._delimeter)</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-4">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-4">¶</a>
|
||||
</div>
|
||||
<p>Check if the buffer has a this._opts.delimeter or “#”, if not, the end of the buffer string might be in the middle of a content length string</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">if</span> (i !== <span class="hljs-number">-1</span>) {
|
||||
<span class="hljs-keyword">var</span> rawContentLength = <span class="hljs-keyword">this</span>._buffer.substring(<span class="hljs-number">0</span>, i)
|
||||
<span class="hljs-keyword">this</span>._contentLength = <span class="hljs-built_in">parseInt</span>(rawContentLength)
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-built_in">isNaN</span>(<span class="hljs-keyword">this</span>._contentLength)) {
|
||||
<span class="hljs-keyword">this</span>._contentLength = <span class="hljs-literal">null</span>
|
||||
<span class="hljs-keyword">this</span>._buffer = <span class="hljs-string">''</span>
|
||||
<span class="hljs-keyword">var</span> err = <span class="hljs-keyword">new</span> <span class="hljs-built_in">Error</span>(<span class="hljs-string">'Invalid content length supplied ('</span>+rawContentLength+<span class="hljs-string">') in: '</span>+<span class="hljs-keyword">this</span>._buffer)
|
||||
err.code = <span class="hljs-string">'E_INVALID_CONTENT_LENGTH'</span>
|
||||
<span class="hljs-keyword">throw</span> err
|
||||
}
|
||||
<span class="hljs-keyword">this</span>._buffer = <span class="hljs-keyword">this</span>._buffer.substring(i+<span class="hljs-number">1</span>)
|
||||
}
|
||||
}
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>._contentLength != <span class="hljs-literal">null</span>) {
|
||||
<span class="hljs-keyword">var</span> length = Buffer.byteLength(<span class="hljs-keyword">this</span>._buffer, <span class="hljs-string">'utf8'</span>)
|
||||
<span class="hljs-keyword">if</span> (length == <span class="hljs-keyword">this</span>._contentLength) {
|
||||
<span class="hljs-keyword">this</span>._handleMessage(<span class="hljs-keyword">this</span>._buffer)
|
||||
} <span class="hljs-keyword">else</span> <span class="hljs-keyword">if</span> (length > <span class="hljs-keyword">this</span>._contentLength) {
|
||||
<span class="hljs-keyword">var</span> message = <span class="hljs-keyword">this</span>._buffer.substring(<span class="hljs-number">0</span>, <span class="hljs-keyword">this</span>._contentLength)
|
||||
<span class="hljs-keyword">var</span> rest = <span class="hljs-keyword">this</span>._buffer.substring(<span class="hljs-keyword">this</span>._contentLength)
|
||||
<span class="hljs-keyword">this</span>._handleMessage(message)
|
||||
<span class="hljs-keyword">this</span>.onData(rest)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_handleMessage (data) {
|
||||
<span class="hljs-keyword">this</span>._contentLength = <span class="hljs-literal">null</span>
|
||||
<span class="hljs-keyword">this</span>._buffer = <span class="hljs-string">''</span>
|
||||
<span class="hljs-keyword">var</span> message
|
||||
<span class="hljs-keyword">try</span> {
|
||||
message = <span class="hljs-built_in">JSON</span>.parse(data)
|
||||
} <span class="hljs-keyword">catch</span> (e) {
|
||||
<span class="hljs-keyword">var</span> err = <span class="hljs-keyword">new</span> <span class="hljs-built_in">Error</span>(<span class="hljs-string">'Could not parse JSON: '</span>+e.message+<span class="hljs-string">'\nRequest data: '</span>+data)
|
||||
err.code = <span class="hljs-string">'E_INVALID_JSON'</span>
|
||||
<span class="hljs-keyword">throw</span> err
|
||||
}
|
||||
message = message || {}
|
||||
<span class="hljs-keyword">this</span>.emit(<span class="hljs-string">'message'</span>, message)
|
||||
}
|
||||
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
</ul>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
|
|
@ -0,0 +1,403 @@
|
|||
<!DOCTYPE html>
|
||||
|
||||
<html>
|
||||
<head>
|
||||
<title>socket.js</title>
|
||||
<meta http-equiv="content-type" content="text/html; charset=UTF-8">
|
||||
<meta name="viewport" content="width=device-width, target-densitydpi=160dpi, initial-scale=1.0; maximum-scale=1.0; user-scalable=0;">
|
||||
<link rel="stylesheet" media="all" href="../docco.css" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="container">
|
||||
<div id="background"></div>
|
||||
|
||||
<ul id="jump_to">
|
||||
<li>
|
||||
<a class="large" href="javascript:void(0);">Jump To …</a>
|
||||
<a class="small" href="javascript:void(0);">+</a>
|
||||
<div id="jump_wrapper">
|
||||
<div id="jump_page_wrapper">
|
||||
<div id="jump_page">
|
||||
|
||||
|
||||
<a class="source" href="consumer.html">
|
||||
./src/consumer.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="index.html">
|
||||
./src/index.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="json-stream.html">
|
||||
./src/json-stream.js
|
||||
</a>
|
||||
|
||||
|
||||
<a class="source" href="socket.html">
|
||||
./src/socket.js
|
||||
</a>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
</li>
|
||||
</ul>
|
||||
|
||||
<ul class="sections">
|
||||
|
||||
<li id="title">
|
||||
<div class="annotation">
|
||||
<h1>socket.js</h1>
|
||||
</div>
|
||||
</li>
|
||||
|
||||
|
||||
|
||||
<li id="section-1">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-1">¶</a>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">import</span> { Server } <span class="hljs-keyword">from</span> <span class="hljs-string">'net'</span>
|
||||
<span class="hljs-keyword">import</span> { unlink <span class="hljs-keyword">as</span> fileDelete } <span class="hljs-keyword">from</span> <span class="hljs-string">'fs'</span>
|
||||
<span class="hljs-keyword">import</span> { promisify } <span class="hljs-keyword">from</span> <span class="hljs-string">'util'</span>
|
||||
<span class="hljs-keyword">import</span> path <span class="hljs-keyword">from</span> <span class="hljs-string">'path'</span>
|
||||
<span class="hljs-keyword">import</span> mkdir <span class="hljs-keyword">from</span> <span class="hljs-string">'make-dir'</span>
|
||||
<span class="hljs-keyword">import</span> btc <span class="hljs-keyword">from</span> <span class="hljs-string">'better-try-catch'</span>
|
||||
<span class="hljs-keyword">import</span> _ON_DEATH <span class="hljs-keyword">from</span> <span class="hljs-string">'death'</span> <span class="hljs-comment">//this is intentionally ugly</span>
|
||||
<span class="hljs-keyword">import</span> JSONStream <span class="hljs-keyword">from</span> <span class="hljs-string">'./json-stream'</span>
|
||||
<span class="hljs-keyword">import</span> clone <span class="hljs-keyword">from</span> <span class="hljs-string">'clone'</span></pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-2">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-2">¶</a>
|
||||
</div>
|
||||
<p>import logger from ‘../../uci-logger/src/logger’</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">import</span> logger <span class="hljs-keyword">from</span> <span class="hljs-string">'@uci/logger'</span>
|
||||
<span class="hljs-keyword">let</span> log = {}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-3">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-3">¶</a>
|
||||
</div>
|
||||
<p>TODO change default pipe dir for windows and mac os</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre><span class="hljs-keyword">const</span> DEFAULT_PIPE_DIR = (process.env.SOCKETS_DIR || <span class="hljs-string">'/tmp/UCI'</span>)
|
||||
<span class="hljs-keyword">const</span> DEFAULT_SOCKET_NAME = <span class="hljs-string">'uci-sock'</span>
|
||||
|
||||
<span class="hljs-keyword">export</span> <span class="hljs-keyword">default</span> <span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">Socket</span> <span class="hljs-keyword">extends</span> <span class="hljs-title">Server</span> </span>{
|
||||
<span class="hljs-keyword">constructor</span> (opts={}) {
|
||||
<span class="hljs-keyword">super</span>()
|
||||
<span class="hljs-keyword">this</span>.id = opts.id || opts.name || <span class="hljs-string">'socket:'</span>+ <span class="hljs-keyword">new</span> <span class="hljs-built_in">Date</span>().getTime()
|
||||
<span class="hljs-keyword">if</span> (!opts.path) {
|
||||
opts.host = opts.host || <span class="hljs-string">'0.0.0.0'</span>
|
||||
opts.port = opts.port || <span class="hljs-number">8080</span>
|
||||
} <span class="hljs-keyword">else</span> {
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">typeof</span> opts.path === <span class="hljs-string">'boolean'</span>) opts.path = path.join(DEFAULT_PIPE_DIR,DEFAULT_SOCKET_NAME )
|
||||
<span class="hljs-keyword">if</span> (path.dirname(opts.path)===<span class="hljs-string">'.'</span>) opts.path = path.join(DEFAULT_PIPE_DIR,opts.path )
|
||||
}
|
||||
<span class="hljs-keyword">this</span>.clientTracking = opts.clientTracking || <span class="hljs-literal">true</span>
|
||||
<span class="hljs-keyword">this</span>.clients = [] <span class="hljs-comment">// track consumers (i.e. clients)</span>
|
||||
<span class="hljs-keyword">this</span>.opts = opts <span class="hljs-comment">// for use to recover from selected errors</span></pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-4">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-4">¶</a>
|
||||
</div>
|
||||
<p>self bindings</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">this</span>._listen = <span class="hljs-keyword">this</span>._listen.bind(<span class="hljs-keyword">this</span>)
|
||||
<span class="hljs-keyword">this</span>.create = <span class="hljs-keyword">this</span>.create.bind(<span class="hljs-keyword">this</span>)
|
||||
log = logger({<span class="hljs-attr">file</span>:<span class="hljs-string">'src/socket.js'</span>,<span class="hljs-attr">class</span>:<span class="hljs-string">'Socket'</span>,<span class="hljs-attr">name</span>:<span class="hljs-string">'socket'</span>,<span class="hljs-attr">id</span>:<span class="hljs-keyword">this</span>.id})
|
||||
} <span class="hljs-comment">// end constructor</span>
|
||||
|
||||
<span class="hljs-keyword">async</span> create () {
|
||||
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>( <span class="hljs-keyword">async</span> (resolve,reject) => {</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-5">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-5">¶</a>
|
||||
</div>
|
||||
<p>couple ways to kill socket process when needed</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> _ON_DEATH( <span class="hljs-keyword">async</span> () => {
|
||||
log.info(<span class="hljs-string">'\nhe\'s dead jim'</span>)
|
||||
<span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._destroy()
|
||||
})
|
||||
process.once(<span class="hljs-string">'SIGUSR2'</span>, <span class="hljs-keyword">async</span> () => {
|
||||
<span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._destroy
|
||||
process.kill(process.pid, <span class="hljs-string">'SIGUSR2'</span>)
|
||||
})
|
||||
|
||||
<span class="hljs-keyword">this</span>.once(<span class="hljs-string">'error'</span>, <span class="hljs-keyword">async</span> (err) => {</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-6">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-6">¶</a>
|
||||
</div>
|
||||
<p>recover from socket file that was not removed</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">if</span> (err.code === <span class="hljs-string">'EADDRINUSE'</span>) {
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>.opts.path) { <span class="hljs-comment">// if TCP socket should already be dead</span>
|
||||
<span class="hljs-keyword">let</span> [err, res] = <span class="hljs-keyword">await</span> btc(promisify(fileDelete))(<span class="hljs-keyword">this</span>.opts.path)
|
||||
<span class="hljs-keyword">if</span>(!err) {
|
||||
log.info({<span class="hljs-attr">res</span>:res, <span class="hljs-attr">socket</span>: <span class="hljs-keyword">this</span>.opts.path}, <span class="hljs-string">'socket already exists.....deleted'</span>)
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._listen(<span class="hljs-keyword">this</span>.opts)
|
||||
}
|
||||
log.fatal({<span class="hljs-attr">err</span>:err},<span class="hljs-string">'error deleting socket. Can not establish a socket'</span>)
|
||||
<span class="hljs-keyword">return</span> err
|
||||
}
|
||||
}
|
||||
<span class="hljs-keyword">if</span> (err.code ===<span class="hljs-string">'EACCES'</span>){
|
||||
<span class="hljs-built_in">console</span>.log({<span class="hljs-attr">socket</span>: <span class="hljs-keyword">this</span>.opts.path}, <span class="hljs-string">'directory does not exist...creating'</span>)
|
||||
<span class="hljs-keyword">await</span> mkdir(path.dirname(<span class="hljs-keyword">this</span>.opts.path))
|
||||
<span class="hljs-built_in">console</span>.log({<span class="hljs-attr">socket</span>: <span class="hljs-keyword">this</span>.opts.path}, <span class="hljs-string">'created'</span>)
|
||||
log.warn({<span class="hljs-attr">socket</span>: <span class="hljs-keyword">this</span>.opts.path}, <span class="hljs-string">'directory does not exist...creating'</span>)
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._listen(<span class="hljs-keyword">this</span>.opts)
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-7">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-7">¶</a>
|
||||
</div>
|
||||
<p>otherwise fatally exit</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> log.info(err, <span class="hljs-string">'creating socket'</span>)
|
||||
reject(err)
|
||||
})
|
||||
|
||||
<span class="hljs-keyword">let</span> [err, res] = <span class="hljs-keyword">await</span> btc(<span class="hljs-keyword">this</span>._listen)(<span class="hljs-keyword">this</span>.opts)
|
||||
<span class="hljs-keyword">if</span> (err) reject(err)
|
||||
resolve(res)
|
||||
|
||||
}) <span class="hljs-comment">// end creeate promise</span>
|
||||
} <span class="hljs-comment">// end create</span>
|
||||
|
||||
registerPacketProcessor (func) {
|
||||
<span class="hljs-keyword">this</span>._packetProcess = func
|
||||
}
|
||||
|
||||
<span class="hljs-keyword">async</span> _listen (opts) {
|
||||
<span class="hljs-keyword">super</span>.listen(opts, <span class="hljs-keyword">async</span> (err, res) => {
|
||||
<span class="hljs-keyword">if</span> (err) <span class="hljs-keyword">return</span> err</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-8">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-8">¶</a>
|
||||
</div>
|
||||
<p>this gets called for each client connection and is unique to each</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">this</span>.on(<span class="hljs-string">'connection'</span>, <span class="hljs-keyword">async</span> (socket) => {
|
||||
<span class="hljs-keyword">const</span> stream = <span class="hljs-keyword">new</span> JSONStream()
|
||||
socket.stream = stream <span class="hljs-comment">// need this to track clients</span>
|
||||
<span class="hljs-keyword">let</span> send = <span class="hljs-keyword">this</span>._send.bind(socket)
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>.clientTracking) <span class="hljs-keyword">this</span>.clients.push(socket)</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-9">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-9">¶</a>
|
||||
</div>
|
||||
<p>TODO add ‘close’ listener to socket to remove from this.clients</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> log.info(<span class="hljs-string">'new consumer connecting'</span>)
|
||||
log.info(<span class="hljs-keyword">await</span> send(<span class="hljs-keyword">await</span> stream.serialize({<span class="hljs-string">'_handshake'</span>:<span class="hljs-literal">true</span>})))
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>.opts.conPacket) {
|
||||
<span class="hljs-keyword">this</span>.opts.conPacket._header = { <span class="hljs-attr">id</span>:<span class="hljs-string">'pushed'</span>}
|
||||
log.info({<span class="hljs-attr">conPacket</span>:<span class="hljs-keyword">this</span>.opts.conPacket},<span class="hljs-string">'pushing a preset command to just connected consumer'</span>)
|
||||
send(<span class="hljs-keyword">await</span> stream.serialize(<span class="hljs-keyword">this</span>.opts.conPacket)) <span class="hljs-comment">// send a packet command on to consumer on connection</span>
|
||||
}
|
||||
socket.on(<span class="hljs-string">'data'</span>, stream.onData)</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-10">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-10">¶</a>
|
||||
</div>
|
||||
<p>TODO need to start error listener for stream so errors can be processed</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> stream.on(<span class="hljs-string">'message'</span>, messageProcess.bind(<span class="hljs-keyword">this</span>,socket))
|
||||
|
||||
<span class="hljs-keyword">async</span> <span class="hljs-function"><span class="hljs-keyword">function</span> <span class="hljs-title">messageProcess</span> (<span class="hljs-params">client, packet</span>) </span>{
|
||||
log.info({<span class="hljs-attr">packet</span>:packet},<span class="hljs-string">'incoming packet on socket side'</span>)
|
||||
<span class="hljs-keyword">let</span> res = {}
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-keyword">this</span>.clientTracking && packet.clientID) {
|
||||
client.ID = packet.clientID
|
||||
res.cmd=<span class="hljs-string">'ackID'</span>
|
||||
}
|
||||
<span class="hljs-keyword">else</span> {
|
||||
res = <span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._packetProcess(clone(packet)) || {}
|
||||
<span class="hljs-keyword">if</span> (<span class="hljs-built_in">Object</span>.keys(res).length === <span class="hljs-number">0</span>) res = { <span class="hljs-attr">error</span>: <span class="hljs-string">'socket packet command function likely did not return a promise'</span>, <span class="hljs-attr">packet</span>:packet}
|
||||
}
|
||||
<span class="hljs-keyword">if</span> (packet) {
|
||||
res._header = clone(packet._header,<span class="hljs-literal">false</span>) || {} <span class="hljs-comment">//make sure return packet has header with id in case it was removed in processing</span>
|
||||
<span class="hljs-keyword">delete</span> packet._header <span class="hljs-comment">// remove before adding to response header as request</span>
|
||||
} <span class="hljs-keyword">else</span> res._header = {}
|
||||
res._header.request = clone(packet,<span class="hljs-literal">false</span>)
|
||||
res._header.responder = {<span class="hljs-attr">name</span>:<span class="hljs-keyword">this</span>.name,<span class="hljs-attr">instanceID</span>:<span class="hljs-keyword">this</span>.id}
|
||||
res._header.socket = <span class="hljs-keyword">this</span>.address()
|
||||
<span class="hljs-keyword">if</span> (!res.cmd) res.cmd = <span class="hljs-string">'reply'</span> <span class="hljs-comment">// by default return command is 'reply'</span>
|
||||
<span class="hljs-keyword">let</span> [err, ser] = <span class="hljs-keyword">await</span> btc(stream.serialize)(res)
|
||||
<span class="hljs-keyword">if</span> (err) ser = <span class="hljs-keyword">await</span> stream.serialize({ <span class="hljs-attr">error</span>: <span class="hljs-string">'was not able to serialze the res packet'</span>, <span class="hljs-attr">err</span>:err, <span class="hljs-attr">_header</span>:{<span class="hljs-attr">id</span>:res._header.id}})
|
||||
log.info(<span class="hljs-keyword">await</span> send(ser))
|
||||
} <span class="hljs-comment">// end process message</span>
|
||||
|
||||
}) <span class="hljs-comment">// end connecttion consumer</span>
|
||||
log.info({<span class="hljs-attr">opts</span>: <span class="hljs-keyword">this</span>.opts},<span class="hljs-string">'socket created'</span>)
|
||||
<span class="hljs-keyword">return</span> res
|
||||
}) <span class="hljs-comment">// end super listen callback</span>
|
||||
|
||||
} <span class="hljs-comment">// end listen</span>
|
||||
|
||||
<span class="hljs-keyword">async</span> _destroy () {
|
||||
log.info(<span class="hljs-string">'closing down socket'</span>)
|
||||
<span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>.close()
|
||||
log.info(<span class="hljs-string">'all connections closed....exiting'</span>)
|
||||
process.exit()
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-11">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-11">¶</a>
|
||||
</div>
|
||||
<p>default packet process, just a simple echo</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">async</span> _packetProcess (packet) {
|
||||
<span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>(<span class="hljs-function"><span class="hljs-params">resolve</span> =></span> {
|
||||
resolve(packet)
|
||||
})
|
||||
}</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-12">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-12">¶</a>
|
||||
</div>
|
||||
<p>must have a consumer socket bound to use</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">async</span> _send(packet) {</pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
|
||||
<li id="section-13">
|
||||
<div class="annotation">
|
||||
|
||||
<div class="pilwrap ">
|
||||
<a class="pilcrow" href="#section-13">¶</a>
|
||||
</div>
|
||||
<p>timeout already set if sockect can’t be drained in 10 secs</p>
|
||||
|
||||
</div>
|
||||
|
||||
<div class="content"><div class='highlight'><pre> <span class="hljs-keyword">return</span> <span class="hljs-keyword">new</span> <span class="hljs-built_in">Promise</span>(<span class="hljs-function"><span class="hljs-params">resolve</span> =></span> {
|
||||
<span class="hljs-keyword">const</span> cb = <span class="hljs-function"><span class="hljs-params">()</span> =></span> resolve(<span class="hljs-string">'packet written to socket stream'</span>)
|
||||
<span class="hljs-keyword">if</span> (!<span class="hljs-keyword">this</span>.write(packet)) {
|
||||
<span class="hljs-keyword">this</span>.once(<span class="hljs-string">'drain'</span>,cb )
|
||||
} <span class="hljs-keyword">else</span> {
|
||||
process.nextTick(cb)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
<span class="hljs-keyword">async</span> push (packet,id) {
|
||||
packet._header = { <span class="hljs-attr">id</span>:<span class="hljs-string">'pushed'</span>}
|
||||
log.info({<span class="hljs-attr">opts</span>:<span class="hljs-keyword">this</span>.opts,<span class="hljs-attr">packet</span>:packet},<span class="hljs-string">'pushing a packet to all connected consumers'</span>)
|
||||
<span class="hljs-keyword">this</span>.clients.forEach(<span class="hljs-keyword">async</span> (client) => {
|
||||
<span class="hljs-keyword">if</span> (client.writable) {
|
||||
<span class="hljs-keyword">let</span> [err, ser] = <span class="hljs-keyword">await</span> btc(client.stream.serialize)(packet)
|
||||
<span class="hljs-keyword">if</span> (err) ser = <span class="hljs-keyword">await</span> client.stream.serialize({ <span class="hljs-attr">error</span>: <span class="hljs-string">'was not able to serialze the res packet'</span>, <span class="hljs-attr">err</span>:err, <span class="hljs-attr">_header</span>:{<span class="hljs-attr">id</span>:packet._header.id}})
|
||||
<span class="hljs-keyword">if</span> (!id || id ===client.ID ) <span class="hljs-keyword">await</span> <span class="hljs-keyword">this</span>._send.bind(client)(ser)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
} <span class="hljs-comment">// end class</span></pre></div></div>
|
||||
|
||||
</li>
|
||||
|
||||
</ul>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
14
package.json
14
package.json
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"name": "@uci/socket",
|
||||
"version": "0.1.9",
|
||||
"version": "0.2.1",
|
||||
"description": "JSON packet intra(named)/inter(TCP) host communication over socket",
|
||||
"main": "src",
|
||||
"scripts": {
|
||||
|
|
@ -38,18 +38,18 @@
|
|||
},
|
||||
"homepage": "https://github.com/uCOMmandIt/message#readme",
|
||||
"devDependencies": {
|
||||
"chai": "^4.1.2",
|
||||
"chai": "^4.2.0",
|
||||
"chai-as-promised": "^7.1.1",
|
||||
"codecov": "^3.0.2",
|
||||
"esm": "^3.0.37",
|
||||
"codecov": "^3.1.0",
|
||||
"esm": "^3.0.84",
|
||||
"istanbul": "^0.4.5",
|
||||
"mocha": "^5.2.0",
|
||||
"nodemon": "^1.17.5"
|
||||
"nodemon": "^1.18.6"
|
||||
},
|
||||
"dependencies": {
|
||||
"@uci/logger": "0.0.3",
|
||||
"@uci/logger": "0.0.6",
|
||||
"better-try-catch": "^0.6.2",
|
||||
"clone": "^2.1.1",
|
||||
"clone": "^2.1.2",
|
||||
"death": "^1.1.0",
|
||||
"make-dir": "^1.3.0"
|
||||
}
|
||||
|
|
|
|||
169
src/consumer.js
169
src/consumer.js
|
|
@ -9,23 +9,44 @@ import logger from '@uci/logger'
|
|||
let log = {}
|
||||
|
||||
// TODO change default pipe dir for windows and mac os
|
||||
const DEFAULT_PIPE_DIR = (process.env.SOCKETS_DIR || '/tmp/UCI')
|
||||
const DEFAULT_PIPE_DIR = process.env.SOCKETS_DIR || '/tmp/UCI'
|
||||
const DEFAULT_SOCKET_NAME = 'uci-sock'
|
||||
|
||||
export default class Consumer extends Socket {
|
||||
constructor (opts={}) {
|
||||
/**
|
||||
* Socket Consumer - connects to UCI TCP or Named Pipe Sockets and coummunicates with uci packet.<br/>
|
||||
* Extends {@link https://nodejs.org/api/net.html#net_class_net_socket | nodejs net.Socket}
|
||||
* @extends Socket
|
||||
*/
|
||||
class SocketConsumer extends Socket {
|
||||
/**
|
||||
* constructor - Description
|
||||
*
|
||||
* @param {object} [opts={}] test
|
||||
*/
|
||||
|
||||
constructor(opts = {}) {
|
||||
super()
|
||||
log = logger({file:'src/consumer.js',class:'Consumer',name:'socket',id:this.id})
|
||||
this.id = opts.id || opts.name || 'socket:'+ new Date().getTime()
|
||||
log = logger({
|
||||
file: 'src/consumer.js',
|
||||
class: 'Consumer',
|
||||
name: 'socket',
|
||||
id: this.id
|
||||
})
|
||||
this.id = opts.id || opts.name || 'socket:' + new Date().getTime()
|
||||
if (!opts.path) {
|
||||
log.warn({opts:opts},'no host supplied using localhost...use named piped instead')
|
||||
log.warn(
|
||||
{ opts: opts },
|
||||
'no host supplied using localhost...use named piped instead'
|
||||
)
|
||||
opts.host = opts.host || '127.0.0.1'
|
||||
opts.port = opts.port || 8080
|
||||
} else {
|
||||
if (typeof opts.path === 'boolean') opts.path = path.join(DEFAULT_PIPE_DIR,DEFAULT_SOCKET_NAME )
|
||||
if (path.dirname(opts.path)==='.') opts.path = path.join(DEFAULT_PIPE_DIR,opts.path )
|
||||
if (typeof opts.path === 'boolean')
|
||||
opts.path = path.join(DEFAULT_PIPE_DIR, DEFAULT_SOCKET_NAME)
|
||||
if (path.dirname(opts.path) === '.')
|
||||
opts.path = path.join(DEFAULT_PIPE_DIR, opts.path)
|
||||
}
|
||||
this.opts=opts
|
||||
this.opts = opts
|
||||
this.keepAlive = 'keepAlive' in opts ? opts.keepAlive : true
|
||||
this._ready = false
|
||||
this.timeout = opts.timeout || 30
|
||||
|
|
@ -37,30 +58,43 @@ export default class Consumer extends Socket {
|
|||
// this._write = this._write.bind(this)
|
||||
}
|
||||
|
||||
async connect () {
|
||||
|
||||
return new Promise( (resolve,reject) => {
|
||||
|
||||
async connect() {
|
||||
return new Promise((resolve, reject) => {
|
||||
const connect = () => {
|
||||
if (this.opts.host ==='127.0.0.1') log.warn('tcp consumer on same machine as host, use named Pipe(Unix) Socket Instead')
|
||||
log.info({opts:this.opts},`attempting to connect ${this.id} to socket`)
|
||||
if (this.opts.host === '127.0.0.1')
|
||||
log.warn(
|
||||
'tcp consumer on same machine as host, use named Pipe(Unix) Socket Instead'
|
||||
)
|
||||
log.info(
|
||||
{ opts: this.opts },
|
||||
`attempting to connect ${this.id} to socket`
|
||||
)
|
||||
super.connect(this.opts)
|
||||
}
|
||||
|
||||
let reconnect = {}
|
||||
const timeout = setTimeout(() =>{
|
||||
const timeout = setTimeout(() => {
|
||||
clearTimeout(reconnect)
|
||||
log.fatal({opts:this.opts},`unable to connect in ${this.timeout}s`)
|
||||
reject({opts:this.opts},`unable to connect to socket server in ${this.timeout}secs`)
|
||||
}
|
||||
,this.timeout*1000)
|
||||
log.fatal({ opts: this.opts }, `unable to connect in ${this.timeout}s`)
|
||||
reject(
|
||||
{ opts: this.opts },
|
||||
`unable to connect to socket server in ${this.timeout}secs`
|
||||
)
|
||||
}, this.timeout * 1000)
|
||||
|
||||
this.once('connect', async () => {
|
||||
clearTimeout(timeout)
|
||||
this._listen()
|
||||
log.info({opts:this.opts},'connected waiting for socket ready handshake')
|
||||
this.setKeepAlive(this.keepAlive,100)
|
||||
let [err, res] = await btc(isReady).bind(this)(this.__ready, this.wait, this.timeout)
|
||||
log.info(
|
||||
{ opts: this.opts },
|
||||
'connected waiting for socket ready handshake'
|
||||
)
|
||||
this.setKeepAlive(this.keepAlive, 100)
|
||||
let [err, res] = await btc(isReady).bind(this)(
|
||||
this.__ready,
|
||||
this.wait,
|
||||
this.timeout
|
||||
)
|
||||
if (err) reject(err)
|
||||
log.info('handshake done, authenticating')
|
||||
// TODO authenticate here by encrypting a payload with private key and sending that.
|
||||
|
|
@ -68,51 +102,59 @@ export default class Consumer extends Socket {
|
|||
resolve(res)
|
||||
})
|
||||
|
||||
this.on('error', async (err) => {
|
||||
log.warn({error:err.code},`connect error ${err.code}`)
|
||||
this.on('error', async err => {
|
||||
log.warn({ error: err.code }, `connect error ${err.code}`)
|
||||
if (err.code === 'EISCONN') {
|
||||
return resolve('ready')
|
||||
}
|
||||
|
||||
reconnect = setTimeout( () =>{ connect()
|
||||
},this.wait*1000 )
|
||||
|
||||
reconnect = setTimeout(() => {
|
||||
connect()
|
||||
}, this.wait * 1000)
|
||||
})
|
||||
|
||||
this.on('end', async () => {
|
||||
log.warn('socket (server) terminated unexpectantly')
|
||||
if (this.keepAlive) {
|
||||
log.info('keep alive was set, so waiting on server to come online for reconnect')
|
||||
log.info(
|
||||
'keep alive was set, so waiting on server to come online for reconnect'
|
||||
)
|
||||
this.destroy()
|
||||
this.emit('error', {code:'DISCONNECTED'})
|
||||
this.emit('error', { code: 'DISCONNECTED' })
|
||||
}
|
||||
})
|
||||
|
||||
connect() // initial connect request
|
||||
|
||||
connect() // initial connect request
|
||||
}) //end promise
|
||||
|
||||
}
|
||||
|
||||
|
||||
async send(ipacket) {
|
||||
return new Promise( async (resolve) => {
|
||||
return new Promise(async resolve => {
|
||||
// need this for when multiple sends for different consumers use same packet instance
|
||||
let packet = Object.assign({},ipacket)
|
||||
setTimeout(() => {resolve({error:'no response from socket in 10sec'})},10000)
|
||||
packet._header =
|
||||
{ id:Math.random().toString().slice(2), // need this for when multiple sends for different consumers use same packet instanceack
|
||||
sender:{ name:this.name, instanceID:this.id },
|
||||
let packet = Object.assign({}, ipacket)
|
||||
setTimeout(() => {
|
||||
resolve({ error: 'no response from socket in 10sec' })
|
||||
}, 10000)
|
||||
packet._header = {
|
||||
id: Math.random()
|
||||
.toString()
|
||||
.slice(2), // need this for when multiple sends for different consumers use same packet instanceack
|
||||
sender: { name: this.name, instanceID: this.id },
|
||||
path: this.opts.path,
|
||||
port: this.opts.port,
|
||||
host: this.opts.host
|
||||
}
|
||||
let [err, res] = await btc(this.stream.serialize)(packet)
|
||||
if (err) resolve({error:'unable to serialize packet for sending', packet:packet})
|
||||
if (err)
|
||||
resolve({
|
||||
error: 'unable to serialize packet for sending',
|
||||
packet: packet
|
||||
})
|
||||
await this.__write(res)
|
||||
this.once(packet._header.id,async function(reply){
|
||||
this.once(packet._header.id, async function(reply) {
|
||||
let res = await this._packetProcess(reply)
|
||||
if (!res) { // if process was not promise returning like just logged to console
|
||||
if (!res) {
|
||||
// if process was not promise returning like just logged to console
|
||||
res = reply
|
||||
// log.warn('consumer function was not promise returning further processing may be out of sequence')
|
||||
}
|
||||
|
|
@ -124,7 +166,7 @@ export default class Consumer extends Socket {
|
|||
// TODO register alt stream processor (emit 'message' with JSON, serialize function, onData method for raw socket chucks)
|
||||
// TODO register authenciation function (set up default)
|
||||
|
||||
registerPacketProcessor (func) {
|
||||
registerPacketProcessor(func) {
|
||||
this._packetProcess = func
|
||||
}
|
||||
|
||||
|
|
@ -135,23 +177,25 @@ export default class Consumer extends Socket {
|
|||
return new Promise(resolve => {
|
||||
const cb = () => resolve('packet written to consumer side socket stream ')
|
||||
if (!super.write(packet)) {
|
||||
this.once('drain',cb )
|
||||
this.once('drain', cb)
|
||||
} else {
|
||||
process.nextTick(cb)
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
|
||||
__ready() {return this._ready}
|
||||
__ready() {
|
||||
return this._ready
|
||||
}
|
||||
|
||||
async _listen () {
|
||||
async _listen() {
|
||||
log.info('listening for incoming packets from socket')
|
||||
// listen for pushed packets
|
||||
this.on('pushed',async function(packet){
|
||||
this.on('pushed', async function(packet) {
|
||||
// TODO do some extra security here?
|
||||
let res = await this._packetProcess(packet)
|
||||
if (!res) { // if process was not promise returning like just logged to console
|
||||
if (!res) {
|
||||
// if process was not promise returning like just logged to console
|
||||
// log.warn('consumer function was not promise returning')
|
||||
}
|
||||
})
|
||||
|
|
@ -159,11 +203,12 @@ export default class Consumer extends Socket {
|
|||
this.on('data', this.stream.onData)
|
||||
this.stream.on('message', messageProcess.bind(this))
|
||||
|
||||
async function messageProcess (packet) {
|
||||
async function messageProcess(packet) {
|
||||
// console.log('incoming packet from socket',packet)
|
||||
if (packet._handshake) {
|
||||
this._ready = true
|
||||
return }
|
||||
return
|
||||
}
|
||||
// TODO send back ack with consumer ID and authorization and wait
|
||||
// when authorized drop through here to emit
|
||||
this.emit(packet._header.id, packet)
|
||||
|
|
@ -171,20 +216,26 @@ export default class Consumer extends Socket {
|
|||
}
|
||||
|
||||
// default packet process just a simple console logger. ignores any cmd: prop
|
||||
_packetProcess (packet) {
|
||||
console.log('default consumer processor -- log packet from socket to console')
|
||||
_packetProcess(packet) {
|
||||
console.log(
|
||||
'default consumer processor -- log packet from socket to console'
|
||||
)
|
||||
console.dir(packet)
|
||||
}
|
||||
} // end class
|
||||
|
||||
} // end class
|
||||
export default SocketConsumer
|
||||
|
||||
// HELP FUNCTIONS
|
||||
// Helper Functions
|
||||
// wait until a passed ready function returns true
|
||||
function isReady(ready, wait=30, timeout=1000) {
|
||||
function isReady(ready, wait = 30, timeout = 1000) {
|
||||
let time = 0
|
||||
return new Promise((resolve, reject) => {
|
||||
(function waitReady(){
|
||||
if (time > timeout) return reject(`timeout waiting for socket ready handshake - ${timeout}ms`)
|
||||
(function waitReady() {
|
||||
if (time > timeout)
|
||||
return reject(
|
||||
`timeout waiting for socket ready handshake - ${timeout}ms`
|
||||
)
|
||||
if (ready()) return resolve('ready')
|
||||
log.info(`waiting ${wait}ms for handshake`)
|
||||
time += wait
|
||||
|
|
|
|||
|
|
@ -1,13 +1,17 @@
|
|||
// adpated from https://github.com/sebastianseilund/node-json-socket
|
||||
|
||||
import {StringDecoder} from 'string_decoder'
|
||||
import { StringDecoder } from 'string_decoder'
|
||||
import EventEmitter from 'events'
|
||||
import btc from 'better-try-catch'
|
||||
|
||||
const decoder = new StringDecoder()
|
||||
|
||||
export default class JsonStream extends EventEmitter{
|
||||
constructor(opts={}){
|
||||
/**
|
||||
* JsonStream - Description
|
||||
* @extends EventEmitter
|
||||
*/
|
||||
class JsonStream extends EventEmitter {
|
||||
constructor(opts = {}) {
|
||||
super()
|
||||
this._contentLength = null
|
||||
this._buffer = ''
|
||||
|
|
@ -16,8 +20,7 @@ export default class JsonStream extends EventEmitter{
|
|||
this.serialize = this.serialize.bind(this)
|
||||
}
|
||||
|
||||
|
||||
onData (data) {
|
||||
onData(data) {
|
||||
// console.log('a chunk arrived', data)
|
||||
data = decoder.write(data)
|
||||
try {
|
||||
|
|
@ -28,10 +31,10 @@ export default class JsonStream extends EventEmitter{
|
|||
}
|
||||
|
||||
async serialize(message) {
|
||||
return new Promise( (resolve,reject) => {
|
||||
let [err,messageData] = btc(JSON.stringify)(message)
|
||||
return new Promise((resolve, reject) => {
|
||||
let [err, messageData] = btc(JSON.stringify)(message)
|
||||
if (err) reject(err)
|
||||
let [err2,length] = btc(Buffer.byteLength)(messageData, 'utf8')
|
||||
let [err2, length] = btc(Buffer.byteLength)(messageData, 'utf8')
|
||||
if (err2) reject(err2)
|
||||
let data = length + this._delimeter + messageData
|
||||
// console.log('serialized',data)
|
||||
|
|
@ -39,7 +42,7 @@ export default class JsonStream extends EventEmitter{
|
|||
})
|
||||
}
|
||||
|
||||
_handleData (data) {
|
||||
_handleData(data) {
|
||||
this._buffer += data
|
||||
if (this._contentLength == null) {
|
||||
var i = this._buffer.indexOf(this._delimeter)
|
||||
|
|
@ -50,11 +53,16 @@ export default class JsonStream extends EventEmitter{
|
|||
if (isNaN(this._contentLength)) {
|
||||
this._contentLength = null
|
||||
this._buffer = ''
|
||||
var err = new Error('Invalid content length supplied ('+rawContentLength+') in: '+this._buffer)
|
||||
var err = new Error(
|
||||
'Invalid content length supplied (' +
|
||||
rawContentLength +
|
||||
') in: ' +
|
||||
this._buffer
|
||||
)
|
||||
err.code = 'E_INVALID_CONTENT_LENGTH'
|
||||
throw err
|
||||
}
|
||||
this._buffer = this._buffer.substring(i+1)
|
||||
this._buffer = this._buffer.substring(i + 1)
|
||||
}
|
||||
}
|
||||
if (this._contentLength != null) {
|
||||
|
|
@ -70,19 +78,22 @@ export default class JsonStream extends EventEmitter{
|
|||
}
|
||||
}
|
||||
|
||||
_handleMessage (data) {
|
||||
_handleMessage(data) {
|
||||
this._contentLength = null
|
||||
this._buffer = ''
|
||||
var message
|
||||
try {
|
||||
message = JSON.parse(data)
|
||||
} catch (e) {
|
||||
var err = new Error('Could not parse JSON: '+e.message+'\nRequest data: '+data)
|
||||
var err = new Error(
|
||||
'Could not parse JSON: ' + e.message + '\nRequest data: ' + data
|
||||
)
|
||||
err.code = 'E_INVALID_JSON'
|
||||
throw err
|
||||
}
|
||||
message = message || {}
|
||||
this.emit('message', message)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
export default JsonStream
|
||||
|
|
|
|||
213
src/socket.js
213
src/socket.js
|
|
@ -1,46 +1,81 @@
|
|||
// node modules
|
||||
import { Server } from 'net'
|
||||
import { unlink as fileDelete } from 'fs'
|
||||
import { promisify } from 'util'
|
||||
import path from 'path'
|
||||
// npmjs modules
|
||||
import mkdir from 'make-dir'
|
||||
import btc from 'better-try-catch'
|
||||
import _ON_DEATH from 'death' //this is intentionally ugly
|
||||
import JSONStream from './json-stream'
|
||||
import clone from 'clone'
|
||||
|
||||
// import logger from '../../uci-logger/src/logger'
|
||||
// uci modules
|
||||
import logger from '@uci/logger'
|
||||
let log = {}
|
||||
let log = {} // must declare here and set later for module wide access
|
||||
|
||||
// TODO change default pipe dir for windows and mac os
|
||||
const DEFAULT_PIPE_DIR = (process.env.SOCKETS_DIR || '/tmp/UCI')
|
||||
// TODO change default pipe dir depending on OS linux,windows,mac
|
||||
/** @constant {String} DEFAULT_PIPE_DIR
|
||||
* @description SOCKETS_DIR environment variable or '/tmp/UCI'
|
||||
*/
|
||||
const DEFAULT_PIPE_DIR = process.env.SOCKETS_DIR || '/tmp/UCI'
|
||||
/** @constant {String} DEFAULT_SOCKET_NAME
|
||||
* @description for named pipe 'uci-sock' if not set in options */
|
||||
const DEFAULT_SOCKET_NAME = 'uci-sock'
|
||||
|
||||
export default class Socket extends Server {
|
||||
constructor (opts={}) {
|
||||
/**
|
||||
* UCI Socket - class used to create a socket (server) that supports passing json packets
|
||||
* supports both named pipes and tcp sockets
|
||||
* also supports push of packets to all connected consumers (clients)
|
||||
* is extended from {@link https://nodejs.org/api/net.html#net_class_net_server | nodejs net.Server }
|
||||
* @extends Server
|
||||
*/
|
||||
|
||||
class Socket extends Server {
|
||||
/**
|
||||
* UCI Socket class constructor
|
||||
* @param {Object} opts hash of options
|
||||
* @param {String} options.host a tcp host name nornally not used as 0.0.0.0 is set by default
|
||||
* @param {String} options.port a tcp
|
||||
* @param {String | Boolean} options.path xeither full path to where socket should be created or if just 'true' then use default
|
||||
* @param {Boolean} options.clientTracking track connected clients for push notifications - default: true
|
||||
* @param {Object} options.conPacket A json operson's property
|
||||
*
|
||||
*/
|
||||
constructor(opts = {}) {
|
||||
super()
|
||||
this.id = opts.id || opts.name || 'socket:'+ new Date().getTime()
|
||||
this.id = opts.id || opts.name || 'socket:' + new Date().getTime()
|
||||
if (!opts.path) {
|
||||
opts.host = opts.host || '0.0.0.0'
|
||||
opts.port = opts.port || 8080
|
||||
} else {
|
||||
if (typeof opts.path === 'boolean') opts.path = path.join(DEFAULT_PIPE_DIR,DEFAULT_SOCKET_NAME )
|
||||
if (path.dirname(opts.path)==='.') opts.path = path.join(DEFAULT_PIPE_DIR,opts.path )
|
||||
if (typeof opts.path === 'boolean')
|
||||
opts.path = path.join(DEFAULT_PIPE_DIR, DEFAULT_SOCKET_NAME)
|
||||
if (path.dirname(opts.path) === '.')
|
||||
opts.path = path.join(DEFAULT_PIPE_DIR, opts.path)
|
||||
}
|
||||
this.clientTracking = opts.clientTracking || true
|
||||
this.clients = [] // track consumers (i.e. clients)
|
||||
this.opts = opts // for use to recover from selected errors
|
||||
this.clients = [] // track consumers (i.e. clients)
|
||||
this.opts = opts // for use to recover from selected errors
|
||||
//self bindings
|
||||
this._listen = this._listen.bind(this)
|
||||
this.create = this.create.bind(this)
|
||||
log = logger({file:'src/socket.js',class:'Socket',name:'socket',id:this.id})
|
||||
log = logger({
|
||||
file: 'src/socket.js',
|
||||
class: 'Socket',
|
||||
name: 'socket',
|
||||
id: this.id
|
||||
})
|
||||
} // end constructor
|
||||
|
||||
async create () {
|
||||
|
||||
return new Promise( async (resolve,reject) => {
|
||||
// couple ways to kill socket process when needed
|
||||
_ON_DEATH( async () => {
|
||||
/**
|
||||
* create - Description
|
||||
*
|
||||
* @returns {type} Description
|
||||
*/
|
||||
async create() {
|
||||
return new Promise(async (resolve, reject) => {
|
||||
// set up a couple ways to gracefully destroy socket process is killed/aborted
|
||||
_ON_DEATH(async () => {
|
||||
log.info('\nhe\'s dead jim')
|
||||
await this._destroy()
|
||||
})
|
||||
|
|
@ -49,24 +84,37 @@ export default class Socket extends Server {
|
|||
process.kill(process.pid, 'SIGUSR2')
|
||||
})
|
||||
|
||||
this.once('error', async (err) => {
|
||||
this.once('error', async err => {
|
||||
// recover from socket file that was not removed
|
||||
if (err.code === 'EADDRINUSE') {
|
||||
if (this.opts.path) { // if TCP socket should already be dead
|
||||
if (this.opts.path) {
|
||||
// if TCP socket should already be dead
|
||||
let [err, res] = await btc(promisify(fileDelete))(this.opts.path)
|
||||
if(!err) {
|
||||
log.info({res:res, socket: this.opts.path}, 'socket already exists.....deleted')
|
||||
if (!err) {
|
||||
log.info(
|
||||
{ res: res, socket: this.opts.path },
|
||||
'socket already exists.....deleted'
|
||||
)
|
||||
return await this._listen(this.opts)
|
||||
}
|
||||
log.fatal({err:err},'error deleting socket. Can not establish a socket')
|
||||
log.fatal(
|
||||
{ err: err },
|
||||
'error deleting socket. Can not establish a socket'
|
||||
)
|
||||
return err
|
||||
}
|
||||
}
|
||||
if (err.code ==='EACCES'){
|
||||
console.log({socket: this.opts.path}, 'directory does not exist...creating')
|
||||
if (err.code === 'EACCES') {
|
||||
console.log(
|
||||
{ socket: this.opts.path },
|
||||
'directory does not exist...creating'
|
||||
)
|
||||
await mkdir(path.dirname(this.opts.path))
|
||||
console.log({socket: this.opts.path}, 'created')
|
||||
log.warn({socket: this.opts.path}, 'directory does not exist...creating')
|
||||
console.log({ socket: this.opts.path }, 'created')
|
||||
log.warn(
|
||||
{ socket: this.opts.path },
|
||||
'directory does not exist...creating'
|
||||
)
|
||||
return await this._listen(this.opts)
|
||||
}
|
||||
// otherwise fatally exit
|
||||
|
|
@ -77,67 +125,109 @@ export default class Socket extends Server {
|
|||
let [err, res] = await btc(this._listen)(this.opts)
|
||||
if (err) reject(err)
|
||||
resolve(res)
|
||||
|
||||
}) // end creeate promise
|
||||
} // end create
|
||||
|
||||
registerPacketProcessor (func) {
|
||||
/**
|
||||
* registerPacketProcessor - Description
|
||||
* @public
|
||||
* @param {func} Description
|
||||
*
|
||||
*/
|
||||
registerPacketProcessor(func) {
|
||||
this._packetProcess = func
|
||||
}
|
||||
|
||||
async _listen (opts) {
|
||||
/**
|
||||
* push - pushes a supplied UCI object packet to all connected clients
|
||||
*
|
||||
* @param {object} packet Description
|
||||
* @param {string} id the header id string of the pushed packet, default: 'pushed'
|
||||
*
|
||||
*/
|
||||
async push(packet, id) {
|
||||
packet._header = { id: id || 'pushed' }
|
||||
log.info(
|
||||
{ opts: this.opts, packet: packet },
|
||||
'pushing a packet to all connected consumers'
|
||||
)
|
||||
this.clients.forEach(async client => {
|
||||
if (client.writable) {
|
||||
let [err, ser] = await btc(client.stream.serialize)(packet)
|
||||
if (err)
|
||||
ser = await client.stream.serialize({
|
||||
error: 'was not able to serialze the res packet',
|
||||
err: err,
|
||||
_header: { id: packet._header.id }
|
||||
})
|
||||
if (!id || id === client.ID) await this._send.bind(client)(ser)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async _listen(opts) {
|
||||
super.listen(opts, async (err, res) => {
|
||||
if (err) return err
|
||||
// this gets called for each client connection and is unique to each
|
||||
this.on('connection', async (socket) => {
|
||||
this.on('connection', async socket => {
|
||||
const stream = new JSONStream()
|
||||
socket.stream = stream // need this to track clients
|
||||
let send = this._send.bind(socket)
|
||||
if (this.clientTracking) this.clients.push(socket)
|
||||
// TODO add 'close' listener to socket to remove from this.clients
|
||||
log.info('new consumer connecting')
|
||||
log.info(await send(await stream.serialize({'_handshake':true})))
|
||||
log.info(await send(await stream.serialize({ _handshake: true })))
|
||||
if (this.opts.conPacket) {
|
||||
this.opts.conPacket._header = { id:'pushed'}
|
||||
log.info({conPacket:this.opts.conPacket},'pushing a preset command to just connected consumer')
|
||||
this.opts.conPacket._header = { id: 'pushed' }
|
||||
log.info(
|
||||
{ conPacket: this.opts.conPacket },
|
||||
'pushing a preset command to just connected consumer'
|
||||
)
|
||||
send(await stream.serialize(this.opts.conPacket)) // send a packet command on to consumer on connection
|
||||
}
|
||||
socket.on('data', stream.onData)
|
||||
// TODO need to start error listener for stream so errors can be processed
|
||||
stream.on('message', messageProcess.bind(this,socket))
|
||||
stream.on('message', messageProcess.bind(this, socket))
|
||||
|
||||
async function messageProcess (client, packet) {
|
||||
log.info({packet:packet},'incoming packet on socket side')
|
||||
async function messageProcess(client, packet) {
|
||||
log.info({ packet: packet }, 'incoming packet on socket side')
|
||||
let res = {}
|
||||
if (this.clientTracking && packet.clientID) {
|
||||
client.ID = packet.clientID
|
||||
res.cmd='ackID'
|
||||
}
|
||||
else {
|
||||
res = await this._packetProcess(clone(packet)) || {}
|
||||
if (Object.keys(res).length === 0) res = { error: 'socket packet command function likely did not return a promise', packet:packet}
|
||||
res.cmd = 'ackID'
|
||||
} else {
|
||||
res = (await this._packetProcess(clone(packet))) || {}
|
||||
if (Object.keys(res).length === 0)
|
||||
res = {
|
||||
error:
|
||||
'socket packet command function likely did not return a promise',
|
||||
packet: packet
|
||||
}
|
||||
}
|
||||
if (packet) {
|
||||
res._header = clone(packet._header,false) || {} //make sure return packet has header with id in case it was removed in processing
|
||||
delete packet._header // remove before adding to response header as request
|
||||
res._header = clone(packet._header, false) || {} //make sure return packet has header with id in case it was removed in processing
|
||||
delete packet._header // remove before adding to response header as request
|
||||
} else res._header = {}
|
||||
res._header.request = clone(packet,false)
|
||||
res._header.responder = {name:this.name,instanceID:this.id}
|
||||
res._header.request = clone(packet, false)
|
||||
res._header.responder = { name: this.name, instanceID: this.id }
|
||||
res._header.socket = this.address()
|
||||
if (!res.cmd) res.cmd = 'reply' // by default return command is 'reply'
|
||||
if (!res.cmd) res.cmd = 'reply' // by default return command is 'reply'
|
||||
let [err, ser] = await btc(stream.serialize)(res)
|
||||
if (err) ser = await stream.serialize({ error: 'was not able to serialze the res packet', err:err, _header:{id:res._header.id}})
|
||||
if (err)
|
||||
ser = await stream.serialize({
|
||||
error: 'was not able to serialze the res packet',
|
||||
err: err,
|
||||
_header: { id: res._header.id }
|
||||
})
|
||||
log.info(await send(ser))
|
||||
} // end process message
|
||||
|
||||
}) // end connecttion consumer
|
||||
log.info({opts: this.opts},'socket created')
|
||||
log.info({ opts: this.opts }, 'socket created')
|
||||
return res
|
||||
}) // end super listen callback
|
||||
|
||||
} // end listen
|
||||
|
||||
async _destroy () {
|
||||
async _destroy() {
|
||||
log.info('closing down socket')
|
||||
await this.close()
|
||||
log.info('all connections closed....exiting')
|
||||
|
|
@ -145,7 +235,7 @@ export default class Socket extends Server {
|
|||
}
|
||||
|
||||
// default packet process, just a simple echo
|
||||
async _packetProcess (packet) {
|
||||
async _packetProcess(packet) {
|
||||
return new Promise(resolve => {
|
||||
resolve(packet)
|
||||
})
|
||||
|
|
@ -157,23 +247,12 @@ export default class Socket extends Server {
|
|||
return new Promise(resolve => {
|
||||
const cb = () => resolve('packet written to socket stream')
|
||||
if (!this.write(packet)) {
|
||||
this.once('drain',cb )
|
||||
this.once('drain', cb)
|
||||
} else {
|
||||
process.nextTick(cb)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async push (packet,id) {
|
||||
packet._header = { id:'pushed'}
|
||||
log.info({opts:this.opts,packet:packet},'pushing a packet to all connected consumers')
|
||||
this.clients.forEach(async (client) => {
|
||||
if (client.writable) {
|
||||
let [err, ser] = await btc(client.stream.serialize)(packet)
|
||||
if (err) ser = await client.stream.serialize({ error: 'was not able to serialze the res packet', err:err, _header:{id:packet._header.id}})
|
||||
if (!id || id ===client.ID ) await this._send.bind(client)(ser)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
} // end class
|
||||
|
||||
export default Socket
|
||||
|
|
|
|||
Loading…
Reference in New Issue