'use strict' ; const debug = require ( 'debug' ) ( 'pandora:messenger:client' ) ; const is = require ( 'is-type-of' ) ; import * as net from 'net' ; import MessengerBase from './base' ; import eventName from './eventName' ; const defaultOptions = { noDelay : true , responseTimeout : 20000 , reConnectTimes : 0 , reConnectInterval : 1000 , } ; const MAX_PACKET_ID = Math . pow ( 2 , 30 ) ; export default class Client extends MessengerBase { private _reConnectTimes : number ; private _socket : net . Socket ; private _header : Buffer ; private _bodyLength : number ; private _queue : Array < any > ; private _packetId : number ; private _unref : boolean ; param { Object } options constructor constructor ( options ) { options = Object . assign ( { } , defaultOptions , options ) ; super ( options ) ; this . _unref = ! ! options . unref ; this . _reConnectTimes = this . options . reConnectTimes ; this . _socket = this . options . socket ; this . _header = null ; this . _bodyLength = null ; this . _queue = [ ] ; this . _packetId = 0 ; if ( ! this . _socket ) { this . _connect ( ) ; } else { this . _bind ( ) ; } } createPacketId ( ) { this . _packetId += 1 ; if ( this . _packetId >= MAX_PACKET_ID ) { this . _packetId = 1 ; } return this . _packetId ; } return { Buffer } getHeader ( ) { return this . read ( 9 ) ; } param { Buffer } header return { Number } getBodyLength ( header ) { return header . readUInt32BE ( 5 ) ; } param { Buffer } buf return { Object } decode ( buf , header ) { const first = header . readUInt8 ( 0 ) ; const id = header . readUInt32BE ( 1 ) ; let data ; if ( buf ) { data = JSON . parse ( buf ) ; } return { oneway : ! ! ( first & 0x80 ) , isResponse : ! ( first & 0x40 ) , id , data , } ; } param { Buffer } buf return { Object } encode ( message ) { let first = 0 ; if ( message . oneway ) { first = first | 0x80 ; } if ( message . isResponse === false ) { first = first | 0x40 ; } const header = new Buffer ( 9 ) ; const data = JSON . stringify ( message . data , replaceErrors ) ; const body = new Buffer ( data ) ; header . fill ( 0 ) ; header . writeUInt8 ( first , 0 ) ; header . writeUInt32BE ( message . id , 1 ) ; header . writeUInt32BE ( Buffer . byteLength ( data ) , 5 ) ; return Buffer . concat ( [ header , body ] ) ; } property get _writable ( ) { return this . isOK ; } property get isOK ( ) { return this . _socket && this . _socket . writable ; } get shouldUnref ( ) { return this . _unref && this . _ready ; } param { Number } n return { Buffer } read ( n ) { return this . _socket . read ( n ) ; } send ( action , data , callback ? , timeout ? ) { return this . _send ( { timeout , data : { action , data , } , } , callback ) ; } param { Object } packet param { Function } callback return { void } _send ( packet , callback ? ) { if ( ! this . _writable ) { this . _queue . push ( [ packet , callback ] ) ; if ( ! this . _socket && ! this . _reConnectTimes ) { this . _cleanQueue ( ) ; } return ; } packet . id = this . createPacketId ( ) ; if ( callback ) { const timeout = packet . timeout || this . options . responseTimeout ; const callbackEvent = `response_callback_${ packet . id }` ; const timer = setTimeout ( ( ) => { this . removeAllListeners ( callbackEvent ) ; const err = new Error ( `target no response in ${ timeout }ms` ) ; err . name = 'MessengerRequestTimeoutError' ; callback ( err ) ; } , timeout ) ; this . once ( callbackEvent , ( message ) => { clearTimeout ( timer ) ; callback ( null , message ) ; } ) ; } this . _socket . write ( this . encode ( packet ) ) ; this . _resume ( ) ; } _cleanQueue ( ) { this . _queue = [ ] ; } _resume ( ) { let args = this . _queue . shift ( ) ; if ( args ) { this . _send ( args [ 0 ] , args [ 1 ] ) ; } } _readPacket ( ) { if ( is . nullOrUndefined ( this . _bodyLength ) ) { this . _header = this . getHeader ( ) ; if ( ! this . _header ) { return false ; } this . _bodyLength = this . getBodyLength ( this . _header ) ; } let body ; if ( this . _bodyLength > 0 ) { body = this . read ( this . _bodyLength ) ; if ( ! body ) { return false ; } } this . _bodyLength = null ; let entity = this . decode ( body , this . _header ) ; setImmediate ( ( ) => { this . emit ( eventName , entity . data , ( res ) => { const id = entity . id ; this . send ( `response_callback_${ id }` , res ) ; } ) ; } ) ; return true ; } return { void } close ( ) { this . _reConnectTimes = 0 ; this . _close ( ) ; } param { Error } err return { void } _close ( err ? ) { if ( ! this . _socket ) { return ; } this . _socket . destroy ( ) ; this . _handleClose ( err ) ; } _handleClose ( err ? ) { if ( ! this . _socket ) { return ; } this . _socket . removeAllListeners ( ) ; this . _socket = null ; if ( err ) { this . _throwError ( err ) ; } if ( this . _reConnectTimes && ( this . _ready || this . options . reConnectAtFirstTime ) ) { const timer = setTimeout ( ( ) => { this . _reConnectTimes -- ; this . _connect ( ( ) => { this . _reConnectTimes = this . options . reConnectTimes ; debug ( '[client] reconnected to the server, process pid is %s' , process . pid ) ; } ) ; } , this . options . reConnectInterval ) ; if ( this . shouldUnref ) { ( < any > timer ) . unref ( ) ; } return ; } this . _cleanQueue ( ) ; this . emit ( 'close' ) ; this . removeAllListeners ( ) ; } _connect ( done ? ) { this . _socket = net . connect ( this . sockPath ) ; this . _socket . once ( 'connect' , ( ) => { this . ready ( true ) ; done && done ( ) ; if ( this . shouldUnref ) { this . _socket . unref ( ) ; } this . _resume ( ) ; this . emit ( 'connect' ) ; } ) ; this . _bind ( ) ; } _bind ( ) { const socket = this . _socket ; socket . setNoDelay ( this . options . noDelay ) ; socket . on ( 'readable' , ( ) => { try { let remaining = false ; do { remaining = this . _readPacket ( ) ; } while ( remaining ) ; } catch ( err ) { err . name = 'PacketParsedError' ; this . _close ( err ) ; } } ) ; socket . once ( 'close' , ( ) => this . _handleClose ( ) ) ; socket . once ( 'error' , err => { this . _close ( err ) ; } ) ; } } function replaceErrors ( key , value ) { if ( value instanceof Error ) { let error = { } ; Object . getOwnPropertyNames ( value ) . forEach ( function ( key ) { error [ key ] = value [ key ] ; } ) ; return error ; } return value ; }