0.2.20 emit status events in consumer

refactor listen method into a handler
master
David Kebler 2019-09-08 19:49:05 -07:00
parent 6862d912b2
commit 5f28baaa74
12 changed files with 113 additions and 2186 deletions

View File

@ -1,518 +0,0 @@
/*--------------------- 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;
}

View File

@ -1,375 +0,0 @@
/*! normalize.css v2.0.1 | MIT License | git.io/normalize */
/* ==========================================================================
HTML5 display definitions
========================================================================== */
/*
* Corrects `block` display not defined in IE 8/9.
*/
article,
aside,
details,
figcaption,
figure,
footer,
header,
hgroup,
nav,
section,
summary {
display: block;
}
/*
* Corrects `inline-block` display not defined in IE 8/9.
*/
audio,
canvas,
video {
display: inline-block;
}
/*
* Prevents modern browsers from displaying `audio` without controls.
* Remove excess height in iOS 5 devices.
*/
audio:not([controls]) {
display: none;
height: 0;
}
/*
* Addresses styling for `hidden` attribute not present in IE 8/9.
*/
[hidden] {
display: none;
}
/* ==========================================================================
Base
========================================================================== */
/*
* 1. Sets default font family to sans-serif.
* 2. Prevents iOS text size adjust after orientation change, without disabling
* user zoom.
*/
html {
font-family: sans-serif; /* 1 */
-webkit-text-size-adjust: 100%; /* 2 */
-ms-text-size-adjust: 100%; /* 2 */
}
/*
* Removes default margin.
*/
body {
margin: 0;
}
/* ==========================================================================
Links
========================================================================== */
/*
* Addresses `outline` inconsistency between Chrome and other browsers.
*/
a:focus {
outline: thin dotted;
}
/*
* Improves readability when focused and also mouse hovered in all browsers.
*/
a:active,
a:hover {
outline: 0;
}
/* ==========================================================================
Typography
========================================================================== */
/*
* Addresses `h1` font sizes within `section` and `article` in Firefox 4+,
* Safari 5, and Chrome.
*/
h1 {
font-size: 2em;
}
/*
* Addresses styling not present in IE 8/9, Safari 5, and Chrome.
*/
abbr[title] {
border-bottom: 1px dotted;
}
/*
* Addresses style set to `bolder` in Firefox 4+, Safari 5, and Chrome.
*/
b,
strong {
font-weight: bold;
}
/*
* Addresses styling not present in Safari 5 and Chrome.
*/
dfn {
font-style: italic;
}
/*
* Addresses styling not present in IE 8/9.
*/
mark {
background: #ff0;
color: #000;
}
/*
* Corrects font family set oddly in Safari 5 and Chrome.
*/
code,
kbd,
pre,
samp {
font-family: monospace, serif;
font-size: 1em;
}
/*
* Improves readability of pre-formatted text in all browsers.
*/
pre {
white-space: pre;
white-space: pre-wrap;
word-wrap: break-word;
}
/*
* Sets consistent quote types.
*/
q {
quotes: "\201C" "\201D" "\2018" "\2019";
}
/*
* Addresses inconsistent and variable font size in all browsers.
*/
small {
font-size: 80%;
}
/*
* Prevents `sub` and `sup` affecting `line-height` in all browsers.
*/
sub,
sup {
font-size: 75%;
line-height: 0;
position: relative;
vertical-align: baseline;
}
sup {
top: -0.5em;
}
sub {
bottom: -0.25em;
}
/* ==========================================================================
Embedded content
========================================================================== */
/*
* Removes border when inside `a` element in IE 8/9.
*/
img {
border: 0;
}
/*
* Corrects overflow displayed oddly in IE 9.
*/
svg:not(:root) {
overflow: hidden;
}
/* ==========================================================================
Figures
========================================================================== */
/*
* Addresses margin not present in IE 8/9 and Safari 5.
*/
figure {
margin: 0;
}
/* ==========================================================================
Forms
========================================================================== */
/*
* Define consistent border, margin, and padding.
*/
fieldset {
border: 1px solid #c0c0c0;
margin: 0 2px;
padding: 0.35em 0.625em 0.75em;
}
/*
* 1. Corrects color not being inherited in IE 8/9.
* 2. Remove padding so people aren't caught out if they zero out fieldsets.
*/
legend {
border: 0; /* 1 */
padding: 0; /* 2 */
}
/*
* 1. Corrects font family not being inherited in all browsers.
* 2. Corrects font size not being inherited in all browsers.
* 3. Addresses margins set differently in Firefox 4+, Safari 5, and Chrome
*/
button,
input,
select,
textarea {
font-family: inherit; /* 1 */
font-size: 100%; /* 2 */
margin: 0; /* 3 */
}
/*
* Addresses Firefox 4+ setting `line-height` on `input` using `!important` in
* the UA stylesheet.
*/
button,
input {
line-height: normal;
}
/*
* 1. Avoid the WebKit bug in Android 4.0.* where (2) destroys native `audio`
* and `video` controls.
* 2. Corrects inability to style clickable `input` types in iOS.
* 3. Improves usability and consistency of cursor style between image-type
* `input` and others.
*/
button,
html input[type="button"], /* 1 */
input[type="reset"],
input[type="submit"] {
-webkit-appearance: button; /* 2 */
cursor: pointer; /* 3 */
}
/*
* Re-set default cursor for disabled elements.
*/
button[disabled],
input[disabled] {
cursor: default;
}
/*
* 1. Addresses box sizing set to `content-box` in IE 8/9.
* 2. Removes excess padding in IE 8/9.
*/
input[type="checkbox"],
input[type="radio"] {
box-sizing: border-box; /* 1 */
padding: 0; /* 2 */
}
/*
* 1. Addresses `appearance` set to `searchfield` in Safari 5 and Chrome.
* 2. Addresses `box-sizing` set to `border-box` in Safari 5 and Chrome
* (include `-moz` to future-proof).
*/
input[type="search"] {
-webkit-appearance: textfield; /* 1 */
-moz-box-sizing: content-box;
-webkit-box-sizing: content-box; /* 2 */
box-sizing: content-box;
}
/*
* Removes inner padding and search cancel button in Safari 5 and Chrome
* on OS X.
*/
input[type="search"]::-webkit-search-cancel-button,
input[type="search"]::-webkit-search-decoration {
-webkit-appearance: none;
}
/*
* Removes inner padding and border in Firefox 4+.
*/
button::-moz-focus-inner,
input::-moz-focus-inner {
border: 0;
padding: 0;
}
/*
* 1. Removes default vertical scrollbar in IE 8/9.
* 2. Improves readability and alignment in all browsers.
*/
textarea {
overflow: auto; /* 1 */
vertical-align: top; /* 2 */
}
/* ==========================================================================
Tables
========================================================================== */
/*
* Remove most spacing between table cells.
*/
table {
border-collapse: collapse;
border-spacing: 0;
}

View File

@ -1,494 +0,0 @@
<!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 &hellip;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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>) =&gt;</span> {
<span class="hljs-keyword">const</span> connect = <span class="hljs-function"><span class="hljs-params">()</span> =&gt;</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> =&gt;</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> () =&gt; {
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">&#182;</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) =&gt; {
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> =&gt;</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> () =&gt; {
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) =&gt; {</pre></div></div>
</li>
<li id="section-7">
<div class="annotation">
<div class="pilwrap ">
<a class="pilcrow" href="#section-7">&#182;</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> =&gt;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</a>
</div>
<p>timeout already set if sockect cant 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> =&gt;</span> {
<span class="hljs-keyword">const</span> cb = <span class="hljs-function"><span class="hljs-params">()</span> =&gt;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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>) =&gt;</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 &gt; 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>

View File

@ -1,78 +0,0 @@
<!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 &hellip;</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">&#182;</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>

View File

@ -1,199 +0,0 @@
<!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 &hellip;</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">&#182;</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">&#182;</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>) =&gt;</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">&#182;</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">&#182;</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 &gt; <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>

View File

@ -1,403 +0,0 @@
<!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 &hellip;</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">&#182;</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">&#182;</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">&#182;</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">&#182;</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) =&gt; {</pre></div></div>
</li>
<li id="section-5">
<div class="annotation">
<div class="pilwrap ">
<a class="pilcrow" href="#section-5">&#182;</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> () =&gt; {
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> () =&gt; {
<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) =&gt; {</pre></div></div>
</li>
<li id="section-6">
<div class="annotation">
<div class="pilwrap ">
<a class="pilcrow" href="#section-6">&#182;</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">&#182;</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) =&gt; {
<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">&#182;</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) =&gt; {
<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">&#182;</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">&#182;</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 &amp;&amp; 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">&#182;</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> =&gt;</span> {
resolve(packet)
})
}</pre></div></div>
</li>
<li id="section-12">
<div class="annotation">
<div class="pilwrap ">
<a class="pilcrow" href="#section-12">&#182;</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">&#182;</a>
</div>
<p>timeout already set if sockect cant 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> =&gt;</span> {
<span class="hljs-keyword">const</span> cb = <span class="hljs-function"><span class="hljs-params">()</span> =&gt;</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) =&gt; {
<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>

View File

@ -1,8 +1,9 @@
import Consumer from '../src/consumer'
import btc from 'better-try-catch'
// const client1= new Consumer({name:'example-consumer1' })
const client= new Consumer({path:true, name:'example-consumer', initTimeout:30 })
// const PATH ='/opt/bogus/socket'
const PATH=true
const client= new Consumer({path:PATH, name:'example-consumer', initTimeout:30 })
// This is your client handler object waiting on a message to do something
async function processor (packet) {
@ -44,8 +45,9 @@ client.pushed = (packet) => {
client.on('connection', event => {
console.log('============ connection update ============')
client.on('status', event => {
console.log('============ socket status ============')
console.log('status level',event.level)
console.log(event.id)
console.log(event.msg)
console.log(`Consumer is ${event.connected ? 'connected' : 'disconnected'}`)

0
examples/opts.yaml Normal file
View File

View File

@ -36,11 +36,11 @@ class Test extends Socket {
// // This is necessary only if the client uses a self-signed certificate.
// // ca: [ fs.readFileSync('client-cert.pem') ]
// }
let options = {path:true}
// const PATH = '/opt/bogus/socket'
const PATH = true
// options.conPacket = {cmd:'onconnect', data:'this is a packet data sent consumer after handshake/authentification'}
options.tokens = ['cheetos']
let test = new Test(options)
const TOKENS = ['cheetos']
let test = new Test({path:PATH, tokens:TOKENS})
async function processor (packet) {
// console.log('packet being processed at socket', packet)
@ -70,25 +70,20 @@ test.registerPacketProcessor(processor)
await test.create()
let count = 0
// const push = setInterval( () => {
// count++
// test.push({cmd:'pushed', count:count, status:`some pushed data ${count}`})
// if (count >3) {
// clearInterval(push)
// test.push({cmd:'pushed',status:'now will simulate server going offline by stopping to send pingfor 10 seconds'})
// test.disablePing()
// setTimeout( () => {
// test.enablePing()
// },10000)
//
// }
// },3000)
const push = setInterval( () => {
count++
test.push({cmd:'pushed', count:count, status:`some pushed data ${count}`})
if (count >3) {
clearInterval(push)
test.push({cmd:'pushed',status:'now will simulate server going offline by stopping to send ping for 10 seconds'})
test.disablePing()
setTimeout( () => {
test.enablePing()
},10000)
}
},3000)
// setTimeout( () => {
// console.log('closing server')
// test._destroy()
// },20000)
})().catch(err => {
console.error('FATAL: UNABLE TO START SYSTEM!\n',err)

View File

@ -1,6 +1,6 @@
{
"name": "@uci/socket",
"version": "0.2.19",
"version": "0.2.20",
"description": "JSON packet intra(named)/inter(TCP) host communication over socket",
"main": "src",
"scripts": {

View File

@ -73,12 +73,12 @@ class SocketConsumer extends Socket {
log.warn({method:'connect', line:107, msg:'tcp consumer on same machine as host, use named Pipe(Unix) Socket Instead'})
log.debug('first connnect attempt for', this.opts.name)
this.emit('connection',{msg:'attempting initial connection', id:this.id, opts:this.opts, ready:false})
this.emit('status',{level:'info', msg:'attempting initial connection', id:this.id, opts:this.opts, ready:false})
let initTimeout = {}
if (this.initTimeout > 499) {
initTimeout = setTimeout(() => {
this.emit('connection',{msg:'initial connection timed out', id:this.id, timeout:this.initTimeout, wait:this.retryWait, opts:this.opts, ready:false})
this.emit('status',{level:'info', msg:'initial connection timed out', id:this.id, timeout:this.initTimeout, wait:this.retryWait, opts:this.opts, ready:false})
this.removeAllListeners()
log.fatal({method:'connect', line:69, opts: this.opts, msg:`unable to initially connect to ${this.opts.name} in ${this.initTimeout/1000} secs no more attempts!`})
this.stream.removeAllListeners()
@ -98,7 +98,7 @@ class SocketConsumer extends Socket {
authPacket.clientName = this.id
let res = (await this._authenticateSend(authPacket)) || {}
if (!res.authenticated) {
this.emit('connection',{msg:`authentication failed: ${res.reason}`, id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'info', msg:`authentication failed: ${res.reason}`, id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
reject('unable to authenticate')
}
else {
@ -106,7 +106,7 @@ class SocketConsumer extends Socket {
this.removeListener('error',initialErrorHandler)
this._listen() // setup for active connection
log.info({method:'connect', line:87, msg:'initial connect/authentication complete ready for communication'})
this.emit('connection',{msg:'authentication succesfull', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'info', msg:'authentication succesfull', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
resolve('initial connection successful')
}
}
@ -137,7 +137,6 @@ class SocketConsumer extends Socket {
return new Promise(async resolve => {
if (!this._connected) {
resolve({ error: 'socket consumer not connected, aborting send' })
return
}
let packet = Object.assign({}, ipacket) // need to avoid mutuation for different consumers using same packet instance
setTimeout(() => {resolve({ error: 'no response from socket in 10sec' })}, 10000)
@ -206,7 +205,7 @@ class SocketConsumer extends Socket {
const reconnectHandler = () => {
this.stream.once('message', handshake.bind(this))
log.debug({method:'connect', line:113, msg:'connected waiting for socket ready handshake'})
this.emit('connection',{msg:'attemping reconnect', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'info', msg:'attemping reconnect', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
}
const handshake = async (packet) => {
@ -217,19 +216,19 @@ class SocketConsumer extends Socket {
authPacket.clientName = this.id
let res = (await this._authenticateSend(authPacket)) || {}
if (!res.authenticated) {
this.emit('connection',{msg:`authentication failed: ${res.reason}`, id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'error', msg:`authentication failed: ${res.reason}`, id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('error',{code:'authentification failed'})
}
else {
this._authenticated = res.authenticated
log.info({method:'connect', line:87, msg:'initial connect/authentication complete ready for communication'})
this.emit('connection',{msg:'authentication successful', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'info', msg:'authentication successful', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
if (this.keepAlive) { // only attempt reconnect if keepAlive is set which it is by default
this.on('ping',pingHandler)
this.setKeepAlive(this.keepAlive,3000) // keep connection alive unless disabled
}
this.stream.on('message', messageHandler.bind(this)) // reset default message handler
this.emit('connection',{msg:'reconnected', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'info', msg:'reconnected', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
}
}
}
@ -238,7 +237,7 @@ class SocketConsumer extends Socket {
log.debug({msg:'connection error emitted ', error:err})
this._connected = false
this._authenticated = false
this.emit('connection',{msg:'connection(socket) error', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
this.emit('status',{level:'error', msg:'connection error', id:this.id, opts:this.opts, authenticated:this._authenticated, connected:this._connected})
log.debug({ method:'connect', line:130, error: err.code, msg:`connect error ${err.code}, attempting reconnect after ${this.retryWait/1000} secs`})
await pause(this.retryWait)
this.stream.removeAllListeners('message') // remove regular message handler in prep for reconnect
@ -304,8 +303,6 @@ class SocketConsumer 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')

View File

@ -65,8 +65,8 @@ export default function socketClass(Server) {
this.clients = [] // track consumers (i.e. clients)
this.nextClientID = 0 // incrementer for default initial client ID
this.opts = opts // for use to recover from selected errors
this.errorCount = 0
//self bindings
this._listen = this._listen.bind(this)
this.create = this.create.bind(this)
this.authenticateClient = this.authenticateClient.bind(this)
this._authenticate = this._authenticate.bind(this)
@ -86,7 +86,6 @@ export default function socketClass(Server) {
*/
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.error({method:'create', line:84, msg:'\nhe\'s dead jim'})
await this._destroy()
@ -103,28 +102,41 @@ export default function socketClass(Server) {
// if TCP socket should already be dead
let [err, res] = await btc(promisify(fileDelete))(this.opts.path)
if (!err) {
log.debug({method:'create', line:99, res: res, socket: this.opts.path, msg:'socket already exists.....deleted'})
return await this._listen(this.opts)
log.info({method:'create', line:99, res: res, socket: this.opts.path, msg:'socket already exists.....deleted'})
// try again
this.removeAllListeners('listening')
return await this.create()
}
log.error({method:'create', line:102, err: err, msg:'error deleting socket. Can not establish a socket'})
return err
}
}
if (err.code === 'EACCES') {
log.debug({method:'create', line:107, socket: this.opts.path, msg:'directory does not exist...creating'})
await mkdir(path.dirname(this.opts.path))
log.debug({method:'create', line:109, socket: this.opts.path, msg:'directory created'})
return await this._listen(this.opts)
this.removeAllListeners('listening')
return await this.create()
}
// otherwise fatally exit
log.error({method:'create', line:113, err:err, msg:'error creating socket'})
log.error({method:'create', line:113, err:err, opts:this.opts, msg:`error creating socket server ${this.name}`})
reject(err)
})
let [err, res] = await btc(this._listen)(this.opts)
if (err) reject(err)
this.once('listening', () => {
this.on('error', err => {
this.errorCount +=1 // log errors here
this.errors.push(err)
if(this.errorCount>2) this.emit('warn', {msg:'something bad maybe going on, 3 errors', errors:this.errors})
if(this.errorCount>5) this.emit('fatal', {msg:'something fatal is going on, 6 errors', errors:this.errors})
})
log.info({method:'create', line:54, msg:'socket server created and listening at', address:this.address()})
this.on('connection', this._connectionHandler.bind(this))
resolve(`socket ready and listening at ${this.address().address}:${this.address().port}`)
})
super.listen(this.opts)
this.enablePing()
resolve(res)
}) // end creeate promise
} // end create
@ -268,88 +280,76 @@ export default function socketClass(Server) {
return true
}
async _listen(opts) {
return super.listen(opts, async (err, res) => {
if (err) return Promise.reject(err)
// this gets called for each client connection and is unique to each
this.on('connection', async socket => {
log.debug({method:'_listen', line:167, msg:'new consumer connecting'})
socket.id = ++this.nextClientID // server assigned ID
socket.authenticated = false
this.clients.push(socket) // add client to list
const stream = new JSONStream()
socket.stream = stream
socket.setKeepAlive(this.keepAlive,3000)
// add listeners
const clientCloseHandler = (id) => {
log.warn({msg:'client connection closed during listen,',id:id})
this.removeClient(id)
}
async _connectionHandler(socket) { // this gets called for each client connection and is unique to each
log.debug({method:'_listen', line:167, msg:'new consumer connecting'})
socket.id = ++this.nextClientID // server assigned ID
socket.authenticated = false
this.clients.push(socket) // add client to list
const stream = new JSONStream()
socket.stream = stream
socket.setKeepAlive(this.keepAlive,3000)
socket.on('close', clientCloseHandler.bind(this,socket.id) )
// add listeners
const clientCloseHandler = (id) => {
log.warn({msg:'client connection closed during listen,',id:id})
this.removeClient(id)
}
socket.on('error', (err) => {
log.error({msg:'client connection error during listen',error:err})
// TODO do more handling than just logging
})
socket.on('close', clientCloseHandler.bind(this,socket.id) )
socket.on('data', stream.onData) // send data to
socket.on('error', (err) => {
log.error({msg:'client connection error during listen',error:err})
// TODO do more handling than just logging
})
stream.on('error', (err) => {
log.error({msg:'client-socket stream error during listen',error:err})
// TODO do more handling than just logging
})
socket.on('data', stream.onData) // send data to
let [err] = await btc(this.authenticateClient)(socket)
if (!this.allowAnonymous) {
if (err) {
socket.end()// abort new connection socket, cleanup, remove listeners
this.removeClient(socket.id)
return
}
}
stream.on('error', (err) => {
log.error({msg:'client-socket stream error during listen',error:err})
// TODO do more handling than just logging
})
// all's set main message processor
stream.on('message', messageProcess.bind(this, socket))
let [err] = await btc(this.authenticateClient)(socket)
if (!this.allowAnonymous) {
if (err) {
socket.end()// abort new connection socket, cleanup, remove listeners
this.removeClient(socket.id)
return
}
}
if (this.opts.conPacket) {
this.opts.conPacket._header = { id: 'pushed' }
log.debug({method:'_listen', line:171, conPacket: this.opts.conPacket, msg:'pushing a preset command to just connected consumer'})
this._send(socket,this.opts.conPacket) // send a packet command on to consumer on connection
}
// all's set main message processor
stream.on('message', messageProcess.bind(this, socket))
// that's it. Connection is active
if (this.opts.conPacket) {
this.opts.conPacket._header = { id: 'pushed' }
log.debug({method:'_listen', line:171, conPacket: this.opts.conPacket, msg:'pushing a preset command to just connected consumer'})
this._send(socket,this.opts.conPacket) // send a packet command on to consumer on connection
}
async function messageProcess(client, packet) {
log.debug({method:'_listen', line:179, packet: packet, client:client.name, msg:'incoming packet on socket side'})
let res = (await this._packetProcess(clone(packet))) || {}
if (Object.keys(res).length === 0)
res = {
error:
// that's it. Connection is active
async function messageProcess(client, packet) {
log.debug({method:'_listen', line:179, packet: packet, client:client.name, msg:'incoming packet on socket side'})
let 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
} else res._header = {}
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'
let [err] = await btc(this._send)(client,res)
if (err) log.error({msg:err, error:err})
} // end message process
}) // end connecttion handler
log.info({method:'_listen', line:255, opts: this.opt, msg:'socket server created and listening'})
return res
}) // end super listen callback
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
} else res._header = {}
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'
let [err] = await btc(this._send)(client,res)
if (err) log.error({msg:err, error:err})
} // end message process
} // end listen