import { EventEmitter } from 'events' ; import createLogger from 'ilp-logger' ; import { DataAndMoneyStream } from './stream' ; import * as IlpPacket from 'ilp-packet' ; import * as cryptoHelper from './crypto' ; import { Packet , Frame , StreamMoneyFrame , StreamCloseFrame , StreamDataFrame , StreamMaxMoneyFrame , FrameType , IlpPacketType , ConnectionNewAddressFrame , ConnectionAssetDetailsFrame , ErrorCode , ConnectionCloseFrame , ConnectionStreamIdBlockedFrame , ConnectionMaxStreamIdFrame , StreamMaxDataFrame , StreamDataBlockedFrame , StreamReceiptFrame , ConnectionMaxDataFrame , ConnectionDataBlockedFrame , StreamMoneyBlockedFrame } from './packet' ; import { Reader , Writer } from 'oer-utils' ; import { CongestionController } from './util/congestion' ; import { Plugin } from './util/plugin-interface' ; import { maxLong , minLong , minLongs , countDigits , checkedAdd , checkedSubtract , multiplyDivideFloor } from './util/long' ; import * as Long from 'long' ; import Rational from './util/rational' ; import { createReceipt , RECEIPT_VERSION } from './util/receipt' ; import { StoppableTimeout } from './util/stoppable-timeout' ; import { v4 as uuid } from 'uuid' ; const RETRY_DELAY_START = 100 ; const RETRY_DELAY_MAX = 43200000 ; const RETRY_DELAY_INCREASE_FACTOR = 1.5 ; const DEFAULT_PACKET_TIMEOUT = 30000 ; const DEFAULT_IDLE_TIMEOUT = 60000 ; const MAX_DATA_SIZE = 32767 ; const DEFAULT_MAX_REMOTE_STREAMS = 10 ; const DEFAULT_MINIMUM_EXCHANGE_RATE_PRECISION = 3 ; const TEST_PACKET_MAX_ATTEMPTS = 15 ; export interface ConnectionOpts { /** Token in the ILP address uniquely identifying this connection */ connectionId ? : string ; /** Ledger plugin (V2) */ plugin : Plugin ; /** ILP Address of the remote entity */ destinationAccount ? : string ; /** ILP Address of the plugin */ sourceAccount ? : string ; /** Specifies how much worse than the initial test packet that the exchange rate is allowed to get before packets are rejected */ slippage ? : number ; /** Pad packets to the maximum size (data field of 32767 bytes). False by default */ enablePadding ? : boolean ; /** User-specified connection identifier that was passed into [`generateAddressAndSecret`] Server connectionTag ? : string ; /** User-specified receipt nonce that was passed into [`generateAddressAndSecret`] Server receiptNonce ? : Buffer ; /** User-specified receipt secret that was passed into [`generateAddressAndSecret`] Server receiptSecret ? : Buffer ; /** Maximum number of streams the other entity can have open at once. Defaults to 10 */ maxRemoteStreams ? : number ; /** Number of bytes each connection can have in the buffer. Defaults to 65534 */ connectionBufferSize ? : number ; /** Minimum Precision to use when determining the exchange rate */ minExchangeRatePrecision ? : number ; /** Inactivity timeout (milliseconds) */ idleTimeout ? : number ; /**
     * Fixed maximum packet amount. When set, the connection's maximum packet amount
     * is the minimum of this setting and the path's discovered maximum amount.
     */ maximumPacketAmount ? : string ; /**
     * Fixed exchange rate. When set, the connection skips the packet volley step.
     * Note that the minimum acceptable exchange rate is usually slightly lower than
     * this (depending on the connection's slippage).
     *
     * This option should usually be used in concert with `ConnectionOpts.maximumPacketAmount`.
     */ exchangeRate ? : number ; /**
     * Returns the expiry to use for an ILP Prepare to the specified destination.
     * When omitted, use a timeout of 30 seconds.
     */ getExpiry ? : ( destination : string ) => Date ; /**
     * Callback for the consumer to perform accounting and choose to fulfill an incoming ILP Prepare,
     * given the amount received, a unique identifier for the packet, and `connectionTag`.
     *
     * If the returned Promise is resolved, the ILP Prepare will be fulfilled; if it is rejected,
     * the ILP Prepare will be rejected. The ILP Fulfill will be immediately sent back after
     * the Promise is resolved.
     *
     * Note: a misbehaving sender can trigger duplicate packetIds, which should be ignored and rejected.
     */ shouldFulfill ? : ( packetAmount : Long , packetId : Buffer , connectionTag ? : string ) => Promise < void > ; } export interface BuildConnectionOpts extends ConnectionOpts { sourceAccount : string ; assetCode : string ; assetScale : number ; isServer : boolean ; /** Shared secret generated by the server with [`generateAddressAndSecret`] Server sharedSecret : Buffer ; } interface NewConnectionOpts extends BuildConnectionOpts { pskKey : Buffer ; fulfillmentKey : Buffer ; } export class ConnectionError extends Error { streamErrorCode : ErrorCode ; constructor ( message , streamErrorCode ? ) { super ( message ) ; this . streamErrorCode = streamErrorCode || ErrorCode . InternalError ; } } enum RemoteState { Init , Connected , Closed } function defaultGetExpiry ( ) { return new Date ( Date . now ( ) + DEFAULT_PACKET_TIMEOUT ) ; } /**
 * Class representing the connection between a [`Client`] createConnection and a [`Server`] Server .
 * A single connection can be used to send or receive on [Streams] DataAndMoneyStream .
 *
 * Streams are created using the [`createStream`] createStream method.
 * The `'stream'` event will be emitted whenever a new incoming stream is opened by the other party.
  export class Connection extends EventEmitter { /** Application identifier for a certain connection */ readonly connectionTag ? : string ; protected readonly _receiptNonce ? : Buffer ; protected readonly _receiptSecret ? : Buffer ; protected connectionId : string ; protected plugin : Plugin ; protected _sourceAccount : string ; protected _sourceAssetCode : string ; protected _sourceAssetScale : number ; protected _destinationAccount ? : string ; protected _destinationAssetCode ? : string ; protected _destinationAssetScale ? : number ; protected sharedSecret : Buffer ; protected _pskKey : Buffer ; protected _fulfillmentKey : Buffer ; protected isServer : boolean ; protected slippage : Rational ; protected allowableReceiveExtra : Rational ; protected enablePadding : boolean ; protected maxBufferedData : number ; protected idleTimeout : number ; protected lastActive : Date ; protected idleTimer : NodeJS . Timer ; protected rateRetryTimer : StoppableTimeout = new StoppableTimeout ( ) ; protected nextPacketSequence : number ; protected streams : Map < number , DataAndMoneyStream > ; protected closedStreams : Set < number > ; protected nextStreamId : number ; protected maxStreamId : number ; protected log : any ; protected sending : boolean ; protected looping : boolean = false ; protected congestion : CongestionController ; protected minExchangeRatePrecision : number ; protected connected : boolean ; protected closed : boolean ; protected done : boolean = false ; protected exchangeRate ? : Rational ; protected retryDelay : number ; protected queuedFrames : Frame [ ] ; protected remoteState : RemoteState = RemoteState . Init ; protected remoteMaxStreamId : number ; protected remoteKnowsOurAccount : boolean ; protected remoteMaxOffset : number ; protected _incomingHold : Long ; protected _totalReceived : Long ; protected _totalSent : Long ; protected _totalDelivered : Long ; protected _lastPacketExchangeRate : Rational ; protected getExpiry : ( destination : string ) => Date ; protected shouldFulfill ? : ( packetAmount : Long , packetId : Buffer , connectionTag ? : string ) => Promise < void > ; constructor ( opts ) { super ( ) ; const lastAddressSegment = opts . destinationAccount ? opts . destinationAccount . split ( '.' ) . slice ( - 1 ) [ 0 ] : undefined ; this . connectionId = ( opts . connectionId || lastAddressSegment || uuid ( ) ) . replace ( /[-_]/g , '' ) . slice ( 0 , 8 ) ; this . plugin = opts . plugin ; this . _sourceAccount = opts . sourceAccount ; this . _sourceAssetCode = opts . assetCode ; this . _sourceAssetScale = opts . assetScale ; this . _destinationAccount = opts . destinationAccount ; this . sharedSecret = opts . sharedSecret ; this . isServer = opts . isServer ; this . _pskKey = opts . pskKey ; this . _fulfillmentKey = opts . fulfillmentKey ; this . slippage = Rational . fromNumber ( opts . slippage === undefined ? 0.01 : opts . slippage , true ) ; if ( this . slippage . greaterThanOne ( ) ) { throw new Error ( 'slippage must be less than one' ) ; } this . allowableReceiveExtra = Rational . fromNumber ( 1.01 , true ) ; this . enablePadding = ! ! opts . enablePadding ; this . connectionTag = opts . connectionTag ; if ( ! opts . receiptNonce !== ! opts . receiptSecret ) { throw new Error ( 'receiptNonce and receiptSecret must accompany each other' ) ; } this . _receiptNonce = opts . receiptNonce ; this . _receiptSecret = opts . receiptSecret ; this . maxStreamId = 2 * ( opts . maxRemoteStreams || DEFAULT_MAX_REMOTE_STREAMS ) ; this . maxBufferedData = opts . connectionBufferSize || MAX_DATA_SIZE * 2 ; this . minExchangeRatePrecision = opts . minExchangeRatePrecision || DEFAULT_MINIMUM_EXCHANGE_RATE_PRECISION ; this . exchangeRate = opts . exchangeRate === undefined ? undefined : Rational . fromNumber ( opts . exchangeRate , true ) ; this . getExpiry = opts . getExpiry || defaultGetExpiry ; this . shouldFulfill = opts . shouldFulfill ; this . idleTimeout = opts . idleTimeout || DEFAULT_IDLE_TIMEOUT ; this . lastActive = new Date ( ) ; this . nextPacketSequence = 1 ; this . streams = new Map ( ) ; this . closedStreams = new Set ( ) ; this . nextStreamId = ( this . isServer ? 2 : 1 ) ; this . log = createLogger ( `ilp-protocol-stream:${ this . isServer ? 'Server' : 'Client' }:Connection:${ this . connectionId }` ) ; this . sending = false ; this . connected = false ; this . closed = true ; this . queuedFrames = [ ] ; this . congestion = new CongestionController ( { maximumPacketAmount : opts . maximumPacketAmount === undefined ? undefined : Long . fromString ( opts . maximumPacketAmount , true ) } ) ; this . retryDelay = RETRY_DELAY_START ; this . remoteKnowsOurAccount = this . isServer ; this . remoteMaxStreamId = DEFAULT_MAX_REMOTE_STREAMS * 2 ; this . remoteMaxOffset = this . maxBufferedData ; this . _incomingHold = Long . UZERO ; this . _totalReceived = Long . UZERO ; this . _totalSent = Long . UZERO ; this . _totalDelivered = Long . UZERO ; this . _lastPacketExchangeRate = Rational . UZERO ; this . once ( 'connect' , ( ) => this . startIdleTimer ( ) ) ; } private static async build ( opts ) { const pskKey = await cryptoHelper . generatePskEncryptionKey ( opts . sharedSecret ) ; const fulfillmentKey = await cryptoHelper . generateFulfillmentKey ( opts . sharedSecret ) ; return new Connection ( Object . assign ( { pskKey , fulfillmentKey } , opts ) ) ; } event type { DataAndMoneyStream } fires async connect ( ) { if ( ! this . closed ) { return Promise . resolve ( ) ; } this . startSendLoop ( ) ; await new Promise ( ( resolve , reject ) => { const connectHandler = ( ) => { cleanup ( ) ; resolve ( ) ; } ; const closeHandler = ( ) => { cleanup ( ) ; this . stopTimers ( ) ; reject ( new Error ( 'Connection was closed before it was connected' ) ) ; } ; const errorHandler = ( error ? ) => { cleanup ( ) ; this . stopTimers ( ) ; reject ( new Error ( `Error connecting${ error ? ': ' + error . message : '' }` ) ) ; } ; this . once ( 'connect' , connectHandler ) ; this . once ( 'error' , errorHandler ) ; this . once ( 'close' , closeHandler ) ; this . once ( 'end' , closeHandler ) ; const self = this ; function cleanup ( ) { self . removeListener ( 'connect' , connectHandler ) ; self . removeListener ( 'error' , errorHandler ) ; self . removeListener ( 'close' , closeHandler ) ; self . removeListener ( 'end' , closeHandler ) ; } } ) ; this . closed = false ; } /**
     * Close the connection when all streams have finished sending their money and data
     */ async end ( ) { this . log . info ( 'closing connection' ) ; let streamEndPromises = [ ] ; for ( let [ _ , stream ] of this . streams ) { if ( stream . isOpen ( ) ) { streamEndPromises . push ( new Promise ( ( resolve , reject ) => { stream . on ( 'end' , resolve ) ; } ) ) ; stream . end ( ) ; } } await new Promise ( ( resolve , reject ) => { this . once ( '_send_loop_finished' , resolve ) ; this . once ( 'error' , reject ) ; this . startSendLoop ( ) ; } ) ; await Promise . all ( streamEndPromises ) ; this . closed = true ; await this . sendConnectionClose ( ) ; this . safeEmit ( 'end' ) ; this . safeEmit ( 'close' ) ; this . stopTimers ( ) ; } /**
     * Immediately close the connection and all streams
     */ async destroy ( err ? ) { this . log . error ( 'destroying connection with error:' , err ) ; if ( this . done ) return ; if ( err ) { this . safeEmit ( 'error' , err ) ; } let streamClosePromises = [ ] ; for ( let [ _ , stream ] of this . streams ) { streamClosePromises . push ( new Promise ( ( resolve , reject ) => { stream . on ( 'close' , resolve ) ; } ) ) ; stream . destroy ( ) ; } await this . sendConnectionClose ( err || new ConnectionError ( 'Connection destroyed' , ErrorCode . ApplicationError ) ) ; await Promise . all ( streamClosePromises ) ; this . safeEmit ( 'close' ) ; this . stopTimers ( ) ; } /**
     * Returns a new bidirectional [`DataAndMoneyStream`] DataAndMoneyStream createStream ( ) { if ( this . remoteMaxStreamId < this . nextStreamId ) { this . log . debug ( 'cannot create another stream. nextStreamId: %d, remote maxStreamId: %d' , this . nextStreamId , this . remoteMaxStreamId ) ; this . queuedFrames . push ( new ConnectionStreamIdBlockedFrame ( this . nextStreamId ) ) ; throw new Error ( `Creating another stream would exceed the remote connection's maximum number of open streams` ) ; } const stream = new DataAndMoneyStream ( { id : this . nextStreamId , isServer : this . isServer , connectionId : this . connectionId } ) ; this . streams . set ( this . nextStreamId , stream ) ; this . log . debug ( 'created stream: %d' , this . nextStreamId ) ; this . nextStreamId += 2 ; stream . on ( '_maybe_start_send_loop' , this . startSendLoop . bind ( this ) ) ; stream . once ( 'close' , ( ) => this . removeStreamRecord ( stream ) ) ; return stream ; } /**
     * ILP Address of the remote party to this connection.
     */ get destinationAccount ( ) : string | undefined { return this . _destinationAccount ; } /**
     * Scale of the asset used by the remote party to this connection
     */ get destinationAssetScale ( ) : number | undefined { return this . _destinationAssetScale ; } /**
     * Code of the asset used by the remote party to this connection
     */ get destinationAssetCode ( ) : string | undefined { return this . _destinationAssetCode ; } /**
     * ILP Address of the plugin passed to this connection.
     */ get sourceAccount ( ) : string { return this . _sourceAccount ; } /**
     * Scale of the asset used by the plugin passed to this connection
     */ get sourceAssetScale ( ) : number { return this . _sourceAssetScale ; } /**
     * Code of the asset used by the plugin passed to this connection
     */ get sourceAssetCode ( ) : string { return this . _sourceAssetCode ; } /**
     * Connections minimum exchange rate with slippage included, if not set '0' is returned.
     */ get minimumAcceptableExchangeRate ( ) : string { if ( this . exchangeRate ) { const minimumExchangeWithSlippage = this . exchangeRate . multiplyByRational ( this . slippage . complement ( ) ) ; return minimumExchangeWithSlippage . toString ( ) ; } return '0' ; } /**
     * Calculates the last exchange rate based on last packet successfully sent.
     */ get lastPacketExchangeRate ( ) : string { return this . _lastPacketExchangeRate . toString ( ) ; } /**
     * Total delivered so far, denominated in the connection plugin's units.
     */ get totalDelivered ( ) : string { return this . _totalDelivered . toString ( ) ; } /**
     * Total sent so far, denominated in the connection plugin's units.
     */ get totalSent ( ) : string { return this . _totalSent . toString ( ) ; } /**
     * Total received so far by the local side, denominated in the connection plugin's units.
     */ get totalReceived ( ) : string { return this . _totalReceived . toString ( ) ; } private async handlePrepare ( prepare ) { let requestPacket ; try { requestPacket = await Packet . decryptAndDeserialize ( this . _pskKey , prepare . data ) ; } catch ( err ) { this . log . error ( 'error parsing frames:' , err ) ; throw new IlpPacket . Errors . UnexpectedPaymentError ( '' ) ; } this . log . trace ( 'handling packet:' , JSON . stringify ( requestPacket ) ) ; if ( requestPacket . ilpPacketType . valueOf ( ) !== IlpPacket . Type . TYPE_ILP_PREPARE ) { this . log . error ( 'prepare packet contains a frame that says it should be something other than a prepare: %d' , requestPacket . ilpPacketType ) ; throw new IlpPacket . Errors . UnexpectedPaymentError ( '' ) ; } this . bumpIdle ( ) ; let responseFrames = [ ] ; responseFrames . push ( new ConnectionMaxDataFrame ( this . getIncomingOffsets ( ) . maxAcceptable ) ) ; const throwFinalApplicationError = async ( ) => { responseFrames = responseFrames . concat ( this . queuedFrames ) ; this . queuedFrames = [ ] ; const responsePacket = new Packet ( requestPacket . sequence , IlpPacketType . Reject , prepare . amount , responseFrames ) ; this . log . trace ( 'rejecting packet %s: %j' , requestPacket . sequence , responsePacket ) ; throw new IlpPacket . Errors . FinalApplicationError ( '' , await responsePacket . serializeAndEncrypt ( this . _pskKey , ( this . enablePadding ? MAX_DATA_SIZE : undefined ) ) ) ; } ; for ( let frame of requestPacket . frames ) { if ( frame . type === FrameType . StreamMoney || frame . type === FrameType . StreamData || frame . type === FrameType . StreamMaxMoney || frame . type === FrameType . StreamMaxData ) { const streamId = frame . streamId . toNumber ( ) ; if ( this . closedStreams . has ( streamId ) ) { this . log . trace ( 'got packet with frame for stream %d, which was already closed' , streamId ) ; if ( frame . type !== FrameType . StreamMoney && frame . type !== FrameType . StreamData ) { continue ; } const testStreamClose = ( frame ) : boolean => { return frame . type === FrameType . StreamClose && frame . streamId . equals ( streamId ) ; } ; const includesStreamClose = responseFrames . find ( testStreamClose ) || this . queuedFrames . find ( testStreamClose ) ; if ( ! includesStreamClose ) { responseFrames . push ( new StreamCloseFrame ( streamId , ErrorCode . StreamStateError , 'Stream is already closed' ) ) ; } await throwFinalApplicationError ( ) ; } try { this . handleNewStream ( frame . streamId . toNumber ( ) ) ; } catch ( err ) { this . log . debug ( 'error handling new stream %s: %s' , frame . streamId , err && err . message ) ; await throwFinalApplicationError ( ) ; } } } try { this . handleControlFrames ( requestPacket . frames ) ; } catch ( err ) { this . log . debug ( 'error handling frames:' , err && err . message ) ; await throwFinalApplicationError ( ) ; } const incomingOffsets = this . getIncomingOffsets ( ) ; if ( incomingOffsets . max > incomingOffsets . maxAcceptable ) { this . destroy ( new ConnectionError ( `Exceeded flow control limits. Max connection byte offset: ${ incomingOffsets . maxAcceptable }, received: ${ incomingOffsets . max }` , ErrorCode . FlowControlError ) ) ; await throwFinalApplicationError ( ) ; } const incomingAmount = Long . fromString ( prepare . amount , true ) ; if ( requestPacket . prepareAmount . greaterThan ( incomingAmount ) ) { this . log . debug ( 'received less than minimum destination amount. actual: %s, expected: %s' , prepare . amount , requestPacket . prepareAmount ) ; await throwFinalApplicationError ( ) ; } const fulfillment = await cryptoHelper . generateFulfillment ( this . _fulfillmentKey , prepare . data ) ; const generatedCondition = await cryptoHelper . hash ( fulfillment ) ; if ( ! generatedCondition . equals ( prepare . executionCondition ) ) { this . log . debug ( 'got unfulfillable prepare for amount: %s. generated condition: %h, prepare condition: %h' , prepare . amount , generatedCondition , prepare . executionCondition ) ; await throwFinalApplicationError ( ) ; } const amountsToReceive = [ ] ; const totalMoneyShares = requestPacket . frames . reduce ( ( sum , frame ) => { if ( frame instanceof StreamMoneyFrame ) { const result = checkedAdd ( sum , frame . shares ) ; if ( result . overflow ) throw new Error ( 'Total shares exceeded MaxUint64' ) ; return result . sum ; } return sum ; } , Long . UZERO ) ; for ( let frame of requestPacket . frames ) { if ( ! ( frame instanceof StreamMoneyFrame ) ) { continue ; } const streamId = frame . streamId . toNumber ( ) ; const streamAmount = multiplyDivideFloor ( incomingAmount , frame . shares , totalMoneyShares ) ; const stream = this . streams . get ( streamId ) ! ; amountsToReceive . push ( { stream , amount : streamAmount } ) ; const maxStreamCanReceive = this . allowableReceiveExtra . multiplyByLongCeil ( stream . _getAmountStreamCanReceive ( ) ) ; if ( maxStreamCanReceive . lessThan ( streamAmount ) ) { this . log . debug ( 'peer sent too much for stream: %d. got: %s, max receivable: %s' , streamId , streamAmount , maxStreamCanReceive ) ; responseFrames . push ( new StreamMaxMoneyFrame ( streamId , stream . receiveMax , stream . totalReceived ) ) ; await throwFinalApplicationError ( ) ; } if ( ! stream . isOpen ( ) ) { this . log . debug ( 'peer sent money for stream that was already closed: %d' , streamId ) ; responseFrames . push ( new StreamCloseFrame ( streamId , ErrorCode . StreamStateError , 'Stream is already closed' ) ) ; await throwFinalApplicationError ( ) ; } } this . addIncomingHold ( incomingAmount ) ; if ( this . shouldFulfill && incomingAmount . greaterThan ( 0 ) ) { const packetId = await cryptoHelper . generateIncomingPacketId ( this . sharedSecret , requestPacket . sequence ) ; await this . shouldFulfill ( incomingAmount , packetId , this . connectionTag ) . catch ( async ( err ) => { this . removeIncomingHold ( incomingAmount ) ; this . log . debug ( 'application declined to fulfill packet %s:' , requestPacket . sequence , err ) ; await throwFinalApplicationError ( ) ; } ) ; } const totalsReceived = new Map ( ) ; for ( let { stream , amount } of amountsToReceive ) { stream . _addToIncoming ( amount , prepare ) ; totalsReceived . set ( stream . id , stream . totalReceived ) ; } if ( ! this . closed && this . remoteState !== RemoteState . Closed ) { for ( let [ _ , stream ] of this . streams ) { if ( ! stream . isOpen ( ) && ! stream . _remoteClosed ) { this . log . trace ( 'telling other side that stream %d is closed' , stream . id ) ; if ( stream . _errorMessage ) { responseFrames . push ( new StreamCloseFrame ( stream . id , ErrorCode . ApplicationError , stream . _errorMessage ) ) ; } else { responseFrames . push ( new StreamCloseFrame ( stream . id , ErrorCode . NoError , '' ) ) ; } stream . _remoteClosed = true ; } else { this . log . trace ( 'telling other side that stream %d can receive %s' , stream . id , stream . receiveMax ) ; responseFrames . push ( new StreamMaxMoneyFrame ( stream . id , stream . receiveMax , stream . totalReceived ) ) ; responseFrames . push ( new StreamMaxDataFrame ( stream . id , stream . _getIncomingOffsets ( ) . maxAcceptable ) ) ; } } } if ( this . _receiptNonce && this . _receiptSecret ) { for ( let [ streamId , totalReceived ] of totalsReceived ) { responseFrames . push ( new StreamReceiptFrame ( streamId , createReceipt ( { nonce : this . _receiptNonce , streamId , totalReceived , secret : this . _receiptSecret } ) ) ) ; } } responseFrames = responseFrames . concat ( this . queuedFrames ) ; this . queuedFrames = [ ] ; const responsePacket = new Packet ( requestPacket . sequence , IlpPacketType . Fulfill , incomingAmount , responseFrames ) ; this . removeIncomingHold ( incomingAmount ) ; this . addTotalReceived ( incomingAmount ) ; this . log . trace ( 'fulfilling prepare with fulfillment: %h and response packet: %j' , fulfillment , responsePacket ) ; return { fulfillment , data : await responsePacket . serializeAndEncrypt ( this . _pskKey , ( this . enablePadding ? MAX_DATA_SIZE : undefined ) ) } ; } /**
     * Parse the frames from the incoming packet and apply all effects
     * except for passing money to the streams
     */ protected handleControlFrames ( frames ) { for ( let frame of frames ) { let stream ; switch ( frame . type ) { case FrameType . ConnectionNewAddress : this . log . trace ( 'peer notified us of their account: %s' , frame . sourceAccount ) ; this . queuedFrames . push ( new ConnectionMaxStreamIdFrame ( this . maxStreamId ) , new ConnectionAssetDetailsFrame ( this . sourceAssetCode , this . sourceAssetScale ) ) ; const firstConnection = this . _destinationAccount === undefined ; this . _destinationAccount = frame . sourceAccount ; if ( firstConnection ) { this . closed = false ; this . log . info ( 'connected' ) ; this . safeEmit ( 'connect' ) ; } break ; case FrameType . ConnectionAssetDetails : this . log . trace ( 'peer notified us of their asset details: code=%s, scale=%d' , frame . sourceAssetCode , frame . sourceAssetScale ) ; this . _destinationAssetCode = frame . sourceAssetCode ; this . _destinationAssetScale = frame . sourceAssetScale ; break ; case FrameType . ConnectionClose : this . sending = false ; this . closed = true ; this . remoteState = RemoteState . Closed ; if ( frame . errorCode === ErrorCode . NoError ) { this . log . info ( 'remote closed connection' ) ; this . end ( ) ; } else { this . log . error ( 'remote connection error. code: %s, message: %s' , ErrorCode [ frame . errorCode ] , frame . errorMessage ) ; this . destroy ( new Error ( `Remote connection error. Code: ${ ErrorCode [ frame . errorCode ] }, message: ${ frame . errorMessage }` ) ) ; } break ; case FrameType . ConnectionMaxData : const outgoingOffsets = this . getOutgoingOffsets ( ) ; this . log . trace ( 'remote connection max byte offset is: %s, we\'ve sent: %d, we want to send up to: %d' , frame . maxOffset , outgoingOffsets . currentOffset , outgoingOffsets . maxOffset ) ; if ( frame . maxOffset . notEquals ( this . maxBufferedData ) ) { this . remoteMaxOffset = Math . max ( frame . maxOffset . toNumber ( ) , this . remoteMaxOffset ) ; } else { this . remoteMaxOffset = frame . maxOffset . toNumber ( ) ; } break ; case FrameType . ConnectionDataBlocked : this . log . trace ( 'remote wants to send more data but we are blocking them. current max incoming offset: %d, remote max offset: %s' , this . getIncomingOffsets ( ) , frame . maxOffset ) ; break ; case FrameType . ConnectionMaxStreamId : this . log . trace ( 'remote set max stream id to %s' , frame . maxStreamId ) ; this . remoteMaxStreamId = frame . maxStreamId . toNumber ( ) ; break ; case FrameType . ConnectionStreamIdBlocked : this . log . trace ( 'remote wants to open more streams but we are blocking them' ) ; break ; case FrameType . StreamClose : this . handleStreamClose ( frame ) ; break ; case FrameType . StreamMaxMoney : this . log . trace ( 'peer told us that stream %s can receive up to: %s and has received: %s so far' , frame . streamId , frame . receiveMax , frame . totalReceived ) ; stream = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( ! stream ) { break ; } stream . _remoteReceived = maxLong ( stream . _remoteReceived , frame . totalReceived ) ; if ( stream . _remoteReceiveMax . notEquals ( Long . MAX_UNSIGNED_VALUE ) ) { stream . _remoteReceiveMax = maxLong ( stream . _remoteReceiveMax , frame . receiveMax ) ; } else { stream . _remoteReceiveMax = frame . receiveMax ; } if ( stream . _remoteReceiveMax . greaterThan ( stream . _remoteReceived ) && stream . _getAmountAvailableToSend ( ) . greaterThan ( 0 ) ) { this . startSendLoop ( ) ; } break ; case FrameType . StreamMoneyBlocked : this . log . debug ( 'peer told us that they want to send more money on stream %s but we are blocking them. they have sent: %s so far and want to send: %s' , frame . streamId , frame . totalSent , frame . sendMax ) ; break ; case FrameType . StreamData : this . log . trace ( 'got data for stream %s' , frame . streamId ) ; stream = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( ! stream ) { break ; } stream . _pushIncomingData ( frame . data , frame . offset . toNumber ( ) ) ; const incomingOffsets = stream . _getIncomingOffsets ( ) ; if ( incomingOffsets . max > incomingOffsets . maxAcceptable ) { this . destroy ( new ConnectionError ( `Exceeded flow control limits. Stream ${ stream . id } can accept up to offset: ${ incomingOffsets . maxAcceptable } but got bytes up to offset: ${ incomingOffsets . max }` , ErrorCode . FlowControlError ) ) ; } break ; case FrameType . StreamMaxData : stream = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( ! stream ) { break ; } const oldOffset = stream . _remoteMaxOffset ; const newOffset = frame . maxOffset . toNumber ( ) ; if ( newOffset > oldOffset ) { this . log . trace ( 'peer told us that stream %s can receive up to byte offset: %s (we\'ve sent up to offset: %d)' , frame . streamId , frame . maxOffset , stream . _getOutgoingOffsets ( ) . current ) ; stream . _remoteMaxOffset = newOffset ; this . startSendLoop ( ) ; } else { this . log . trace ( 'peer told us that stream %s can receive up to byte offset: %d; ignoring new offset: %d' , frame . streamId , oldOffset , newOffset ) ; } break ; case FrameType . StreamDataBlocked : stream = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( ! stream ) { break ; } this . log . debug ( 'peer told us that stream %s is blocked. they want to send up to offset: %s, but we are only allowing up to: %d' , frame . streamId , frame . maxOffset , stream . _getIncomingOffsets ( ) . maxAcceptable ) ; break ; default : continue ; } } } /**
     * Ensure that the new stream is valid and does not exceed our limits
     * and if it looks good, emit the 'stream' event
     */ protected handleNewStream ( streamId ) { if ( this . streams . has ( streamId ) || this . closedStreams . has ( streamId ) ) { return ; } if ( this . isServer && streamId % 2 === 0 ) { this . log . error ( 'got invalid stream ID %d from peer (should be odd)' , streamId ) ; this . queuedFrames . push ( new ConnectionCloseFrame ( ErrorCode . ProtocolViolation , `Invalid Stream ID: ${ streamId }. Client-initiated streams must have odd-numbered IDs` ) ) ; const err = new Error ( `Invalid Stream ID: ${ streamId }. Client-initiated streams must have odd-numbered IDs` ) ; this . safeEmit ( 'error' , err ) ; throw err ; } else if ( ! this . isServer && streamId % 2 === 1 ) { this . log . error ( 'got invalid stream ID %d from peer (should be even)' , streamId ) ; this . queuedFrames . push ( new ConnectionCloseFrame ( ErrorCode . ProtocolViolation , `Invalid Stream ID: ${ streamId }. Server-initiated streams must have even-numbered IDs` ) ) ; const err = new Error ( `Invalid Stream ID: ${ streamId }. Server-initiated streams must have even-numbered IDs` ) ; this . safeEmit ( 'error' , err ) ; throw err ; } if ( streamId > this . maxStreamId ) { this . log . debug ( 'peer opened too many streams. got stream: %d, but max stream id is: %d. closing connection' , streamId , this . maxStreamId ) ; this . queuedFrames . push ( new ConnectionCloseFrame ( ErrorCode . StreamIdError , `Maximum number of open streams exceeded. Got stream: ${ streamId }, current max stream ID: ${ this . maxStreamId }` ) ) ; const err = new Error ( `Maximum number of open streams exceeded. Got stream: ${ streamId }, current max stream ID: ${ this . maxStreamId }` ) ; this . safeEmit ( 'error' , err ) ; throw err ; } if ( this . maxStreamId * .75 < streamId ) { this . log . trace ( 'informing peer that our max stream id is: %d' , this . maxStreamId ) ; this . queuedFrames . push ( new ConnectionMaxStreamIdFrame ( this . maxStreamId ) ) ; } this . log . info ( 'got new stream: %d' , streamId ) ; const stream = new DataAndMoneyStream ( { id : streamId , isServer : this . isServer , connectionId : this . connectionId } ) ; this . streams . set ( streamId , stream ) ; stream . on ( '_maybe_start_send_loop' , ( ) => this . startSendLoop ( ) ) ; stream . once ( 'close' , ( ) => this . removeStreamRecord ( stream ) ) ; this . safeEmit ( 'stream' , stream ) ; } /**
     * Mark the stream as closed
     */ protected handleStreamClose ( frame ) { const streamId = frame . streamId . toNumber ( ) ; const stream = this . streams . get ( streamId ) ; if ( ! stream ) { this . log . error ( 'remote error on stream %d, but we don\'t have a record of that stream' , streamId ) ; return ; } if ( ! stream . isOpen ( ) || stream . _remoteSentEnd ) { return ; } this . log . error ( 'peer closed stream %d with error code: %s and message: %s' , stream . id , ErrorCode [ frame . errorCode ] , frame . errorMessage ) ; stream . _sentEnd = true ; let err ; if ( frame . errorMessage ) { err = new Error ( frame . errorMessage ) ; err . name = ErrorCode [ frame . errorCode ] ; } stream . _remoteEnded ( err ) ; this . maxStreamId += 2 ; this . log . trace ( 'raising maxStreamId to %d' , this . maxStreamId ) ; this . queuedFrames . push ( new ConnectionMaxStreamIdFrame ( this . maxStreamId ) ) ; this . startSendLoop ( ) ; } private protected async startSendLoop ( ) { if ( this . looping ) { this . sending = true ; return ; } if ( this . remoteState === RemoteState . Closed ) { this . log . debug ( 'remote connection is already closed, not starting another loop' ) ; this . safeEmit ( '_send_loop_finished' ) ; return ; } if ( ! this . _destinationAccount ) { this . log . debug ( 'not sending because we do not know the client\'s address' ) ; this . safeEmit ( '_send_loop_finished' ) ; return ; } this . looping = true ; this . sending = true ; this . log . debug ( 'starting send loop' ) ; try { while ( this . sending ) { if ( ! this . connected ) { await this . setupExchangeRate ( ) ; this . connected = true ; } else { await this . loadAndSendPacket ( ) ; } } this . looping = false ; } catch ( err ) { this . looping = false ; return this . destroy ( err ) ; } this . log . debug ( 'finished sending' ) ; this . safeEmit ( '_send_loop_finished' ) ; for ( let [ _ , stream ] of this . streams ) { stream . emit ( '_send_loop_finished' ) ; } } private protected async loadAndSendPacket ( ) { await new Promise ( ( resolve , reject ) => setImmediate ( resolve ) ) ; this . log . trace ( 'loadAndSendPacket' ) ; let amountToSend = Long . UZERO ; const requestPacket = new Packet ( this . getNextPacketSequence ( ) , IlpPacketType . Prepare , undefined , this . queuedFrames ) ; this . queuedFrames = [ ] ; this . maybePushAccountFrames ( requestPacket ) ; for ( let [ _ , stream ] of this . streams ) { if ( stream . isOpen ( ) ) { requestPacket . frames . push ( new StreamMaxMoneyFrame ( stream . id , stream . receiveMax , stream . totalReceived ) ) ; requestPacket . frames . push ( new StreamMaxDataFrame ( stream . id , stream . _getIncomingOffsets ( ) . maxAcceptable ) ) ; } } if ( this . closed && this . remoteState === RemoteState . Connected ) { this . log . trace ( 'sending connection close frame' ) ; requestPacket . frames . push ( new ConnectionCloseFrame ( ErrorCode . NoError , '' ) ) ; this . remoteState = RemoteState . Closed ; } let maxAmountFromNextStream = this . congestion . testMaximumPacketAmount ; if ( this . exchangeRate ! . greaterThanOne ( ) ) { maxAmountFromNextStream = minLong ( maxAmountFromNextStream , this . exchangeRate ! . reciprocal ( ) . multiplyByLong ( Long . MAX_UNSIGNED_VALUE ) ) ; } const streamsSentFrom = [ ] ; for ( let [ _ , stream ] of this . streams ) { if ( stream . _sentEnd ) { continue ; } let amountToSendFromStream = minLong ( stream . _getAmountAvailableToSend ( ) , maxAmountFromNextStream ) ; const maxDestinationAmount = checkedSubtract ( stream . _remoteReceiveMax , stream . _remoteReceived ) . difference ; const maxSourceAmount = this . exchangeRate ! . reciprocal ( ) . multiplyByLongCeil ( maxDestinationAmount ) ; if ( maxSourceAmount . lessThan ( amountToSendFromStream ) ) { this . log . trace ( 'stream %d could send %s but that would be more than the receiver says they can receive, so we\'ll send %s instead' , stream . id , amountToSendFromStream , maxSourceAmount ) ; amountToSendFromStream = maxSourceAmount ; } this . log . trace ( 'amount to send from stream %d: %s, exchange rate: %s, remote total received: %s, remote receive max: %s' , stream . id , amountToSendFromStream , this . exchangeRate , stream . _remoteReceived , stream . _remoteReceiveMax ) ; if ( amountToSendFromStream . greaterThan ( 0 ) ) { stream . _holdOutgoing ( requestPacket . sequence . toString ( ) , amountToSendFromStream ) ; requestPacket . frames . push ( new StreamMoneyFrame ( stream . id , amountToSendFromStream ) ) ; amountToSend = amountToSend . add ( amountToSendFromStream ) ; maxAmountFromNextStream = maxAmountFromNextStream . subtract ( amountToSendFromStream ) ; streamsSentFrom . push ( stream ) ; } const amountLeftStreamWantsToSend = Long . fromString ( stream . sendMax , true ) . subtract ( stream . totalSent ) . subtract ( amountToSendFromStream ) ; if ( this . exchangeRate ! . multiplyByLong ( amountLeftStreamWantsToSend ) . greaterThan ( checkedSubtract ( stream . _remoteReceiveMax , stream . _remoteReceived ) . difference ) ) { requestPacket . frames . push ( new StreamMoneyBlockedFrame ( stream . id , stream . sendMax , stream . totalSent ) ) ; } if ( maxAmountFromNextStream . equals ( 0 ) ) { break ; } } let bytesLeftInPacket = MAX_DATA_SIZE - requestPacket . byteLength ( ) ; const maxBytesRemoteConnectionCanReceive = this . remoteMaxOffset - this . getOutgoingOffsets ( ) . currentOffset ; if ( bytesLeftInPacket > maxBytesRemoteConnectionCanReceive ) { const outgoingMaxOffset = this . getOutgoingOffsets ( ) . maxOffset ; this . log . debug ( 'peer is blocking us from sending more data. they will only accept up to offset: %d, but we want to send up to: %d' , this . remoteMaxOffset , outgoingMaxOffset ) ; requestPacket . frames . push ( new ConnectionDataBlockedFrame ( outgoingMaxOffset ) ) ; bytesLeftInPacket = maxBytesRemoteConnectionCanReceive ; } for ( let [ _ , stream ] of this . streams ) { if ( bytesLeftInPacket - 20 <= 0 ) { break ; } const { data , offset } = stream . _getAvailableDataToSend ( bytesLeftInPacket - 20 ) ; if ( data && data . length > 0 ) { const streamDataFrame = new StreamDataFrame ( stream . id , offset , data ) ; this . log . trace ( 'sending %d bytes from stream %d' , data . length , stream . id ) ; bytesLeftInPacket -= streamDataFrame . byteLength ( ) ; requestPacket . frames . push ( streamDataFrame ) ; } const maxOutgoingOffset = stream . _isDataBlocked ( ) ; if ( maxOutgoingOffset ) { this . log . trace ( 'telling remote that stream %d is blocked and has more data to send' , stream . id ) ; requestPacket . frames . push ( new StreamDataBlockedFrame ( stream . id , maxOutgoingOffset ) ) ; } } if ( amountToSend . equals ( 0 ) ) { if ( requestPacket . frames . length === 0 ) { this . sending = false ; return ; } else { if ( ! requestPacket . frames . find ( frame => ( ( frame . type === FrameType . StreamClose ) || ( frame . type === FrameType . StreamData ) || ( frame . type === FrameType . StreamMoney ) ) ) ) { this . sending = false ; } } } const minimumDestinationAmount = this . slippage . complement ( ) . multiplyByLong ( this . exchangeRate ! . multiplyByLong ( amountToSend ) ) ; if ( minimumDestinationAmount . greaterThan ( 0 ) ) { requestPacket . prepareAmount = minimumDestinationAmount ; } const responsePacket = await this . sendPacket ( requestPacket , amountToSend , false ) ; if ( responsePacket ) { if ( this . remoteState === RemoteState . Init ) { this . remoteState = RemoteState . Connected ; } this . remoteKnowsOurAccount = true ; this . handleControlFrames ( responsePacket . frames ) ; if ( amountToSend . greaterThan ( 0 ) ) { this . _lastPacketExchangeRate = new Rational ( responsePacket . prepareAmount , amountToSend , true ) ; } if ( responsePacket . ilpPacketType === IlpPacketType . Fulfill ) { for ( let frame of responsePacket . frames ) { if ( frame . type === FrameType . StreamReceipt ) { const stream = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( stream ) { stream . _setReceipt ( frame . receipt ) ; } else { this . log . debug ( 'received receipt for unknown stream %d: %h' , frame . streamId , frame . receipt ) ; } } } for ( let stream of streamsSentFrom ) { stream . _executeHold ( requestPacket . sequence . toString ( ) ) ; } this . addTotalDelivered ( responsePacket . prepareAmount ) ; this . addTotalSent ( amountToSend ) ; this . congestion . onFulfill ( amountToSend ) ; this . retryDelay = RETRY_DELAY_START ; } } } private protected async sendTestPacketVolley ( testPacketAmounts ) { const results = await Promise . all ( testPacketAmounts . map ( async ( amount ) => { try { return this . sendTestPacket ( Long . fromNumber ( amount , true ) ) ; } catch ( err ) { this . log . error ( 'Error sending test packet for amount %d: %s' , amount , err ) ; return null ; } } ) ) ; const maxPacketAmounts = testPacketAmounts . map ( ( sourceAmount , index ) => { if ( results [ index ] && ( results [ index ] as IlpPacket . IlpReject ) . code === 'F08' ) { try { const reader = Reader . from ( ( results [ index ] as IlpPacket . IlpReject ) . data ) ; const receivedAmount = reader . readUInt64Long ( ) ; const maximumAmount = reader . readUInt64Long ( ) ; const maximumPacketAmount = multiplyDivideFloor ( Long . fromNumber ( sourceAmount , true ) , maximumAmount , receivedAmount ) ; this . log . debug ( 'sending test packet of %d resulted in F08 error that told us maximum packet amount is %s' , testPacketAmounts [ index ] , maximumPacketAmount ) ; return maximumPacketAmount ; } catch ( err ) { return Long . MAX_UNSIGNED_VALUE ; } } return Long . MAX_UNSIGNED_VALUE ; } ) ; const { maxDigits , exchangeRate , packetErrors } = results . reduce < any > ( ( { maxDigits , exchangeRate , packetErrors } , result , index ) => { const sourceAmount = testPacketAmounts [ index ] ; if ( result && ( result as IlpPacket . IlpReject ) . code ) { packetErrors . push ( { sourceAmount : sourceAmount , code : ( result as IlpPacket . IlpReject ) . code } ) ; } if ( result && ( result as Packet ) . prepareAmount ) { const prepareAmount = ( result as Packet ) . prepareAmount ; const exchangeRate = new Rational ( prepareAmount , Long . fromNumber ( sourceAmount , true ) , true ) ; this . log . debug ( 'sending test packet of %d delivered %s (exchange rate: %s)' , sourceAmount , prepareAmount , exchangeRate ) ; if ( countDigits ( prepareAmount ) >= maxDigits ) { return { maxDigits : countDigits ( prepareAmount ) , exchangeRate , packetErrors } ; } } return { maxDigits , exchangeRate , packetErrors } ; } , { maxDigits : 0 , exchangeRate : Rational . UZERO , packetErrors : [ ] } ) ; return { maxDigits , exchangeRate , maxPacketAmounts , packetErrors } ; } protected async setupExchangeRate ( ) { if ( ! this . exchangeRate ) { this . log . trace ( 'determining exchange rate' ) ; await this . determineExchangeRate ( ) ; } process . nextTick ( ( ) => { this . safeEmit ( 'connect' ) ; this . log . trace ( 'connected' ) ; } ) ; } private protected async determineExchangeRate ( ) { this . log . trace ( 'determineExchangeRate' ) ; if ( ! this . _destinationAccount ) { throw new Error ( 'Cannot determine exchange rate. Destination account is unknown' ) ; } let retryDelay = RETRY_DELAY_START ; let testPacketAmounts = [ 1 , 1e3 , 1e6 , 1e9 , 1e12 ] ; let attempts = 0 ; while ( ! this . exchangeRate && testPacketAmounts . length > 0 && attempts < TEST_PACKET_MAX_ATTEMPTS ) { attempts ++ ; const { maxDigits , exchangeRate , maxPacketAmounts , packetErrors } = await this . sendTestPacketVolley ( testPacketAmounts ) ; this . congestion . setMaximumAmounts ( minLongs ( maxPacketAmounts . concat ( this . congestion . maximumPacketAmount ) ) ) ; if ( this . congestion . maximumPacketAmount . equals ( 0 ) ) { this . log . error ( 'cannot send anything through this path. the maximum packet amount is 0' ) ; throw new Error ( 'Cannot send. Path has a Maximum Packet Amount of 0' ) ; } if ( maxDigits >= this . minExchangeRatePrecision ) { this . log . debug ( 'determined exchange rate to be %s with %d digits precision' , exchangeRate , maxDigits ) ; this . exchangeRate = exchangeRate ; return ; } testPacketAmounts = maxPacketAmounts . filter ( ( amount ) => ! amount . equals ( Long . MAX_UNSIGNED_VALUE ) ) . reduce ( ( acc , curr ) => [ ... new Set ( [ ... acc , curr . toString ( ) ] ) ] , [ ] ) ; if ( packetErrors . some ( ( error ) => error . code [ 0 ] === 'T' ) ) { const smallestPacketAmount = packetErrors . reduce ( ( min , error ) => { return minLong ( min , Long . fromNumber ( error . sourceAmount , true ) ) ; } , Long . MAX_UNSIGNED_VALUE ) ; const reducedPacketAmount = smallestPacketAmount . subtract ( smallestPacketAmount . divide ( 3 ) ) ; this . log . debug ( 'got Txx error(s), waiting %dms and reducing packet amount to %s before sending another test packet' , retryDelay , reducedPacketAmount ) ; testPacketAmounts = [ ... testPacketAmounts , reducedPacketAmount ] ; await this . rateRetryTimer . wait ( retryDelay ) . catch ( ( _err ) => { this . log . debug ( 'connection terminated before rate could be determined; delay=%d' , retryDelay ) ; throw new Error ( 'Connection terminated before rate could be determined.' ) ; } ) ; retryDelay *= RETRY_DELAY_INCREASE_FACTOR ; } this . log . debug ( 'retry with packet amounts %j' , testPacketAmounts ) ; } throw new Error ( `Unable to establish connection, no packets meeting the minimum exchange precision of ${ this . minExchangeRatePrecision } digits made it through the path.` ) ; } private stopTimers ( ) { if ( this . rateRetryTimer ) this . rateRetryTimer . stop ( ) ; clearTimeout ( this . idleTimer ) ; this . done = true ; } private protected async sendTestPacket ( amount , timeout = DEFAULT_PACKET_TIMEOUT ) { const requestPacket = new Packet ( this . getNextPacketSequence ( ) , IlpPacketType . Prepare ) ; this . log . trace ( 'sending test packet %s for amount: %s. timeout: %d' , requestPacket . sequence , amount , timeout ) ; this . maybePushAccountFrames ( requestPacket ) ; const prepare = { destination : this . _destinationAccount ! , amount : amount . toString ( ) , data : await requestPacket . serializeAndEncrypt ( this . _pskKey ) , executionCondition : cryptoHelper . generateRandomCondition ( ) , expiresAt : this . getExpiry ( this . _destinationAccount ! ) } ; const responseData = await ( new Promise ( async ( resolve , reject ) => { const timer = setTimeout ( ( ) => { this . log . error ( 'test packet %s timed out before we got a response' , requestPacket . sequence ) ; resolve ( null ) ; } , timeout ) ; const result = await this . plugin . sendData ( IlpPacket . serializeIlpPrepare ( prepare ) ) ; clearTimeout ( timer ) ; resolve ( result ) ; } ) as Promise < Buffer | null > ) ; if ( ! responseData ) { return null ; } this . bumpIdle ( ) ; const ilpReject = IlpPacket . deserializeIlpReject ( responseData ) ; let responsePacket ; if ( ilpReject . code === 'F99' && ilpReject . data . length > 0 ) { responsePacket = await Packet . decryptAndDeserialize ( this . _pskKey , ilpReject . data ) ; if ( ! responsePacket . sequence . equals ( requestPacket . sequence ) ) { this . log . error ( 'response packet sequence does not match the request packet. expected sequence: %s, got response packet: %j' , requestPacket . sequence , responsePacket ) ; throw new Error ( `Response packet sequence does not correspond to the request. Actual: ${ responsePacket . sequence }, expected: ${ requestPacket . sequence }` ) ; } if ( responsePacket . ilpPacketType !== responseData [ 0 ] ) { this . log . error ( 'response packet was on wrong ILP packet type. expected ILP packet type: %d, got: %j' , responseData [ 0 ] , responsePacket ) ; throw new Error ( `Response says it should be on an ILP packet of type: ${ responsePacket . ilpPacketType } but it was carried on an ILP packet of type: ${ responseData [ 0 ] }` ) ; } } else { this . log . debug ( 'test packet %s was rejected with a %s triggered by %s error%s' , requestPacket . sequence , ilpReject . code , ilpReject . triggeredBy , ilpReject . message ? ' with the message: "' + ilpReject . message + '"' : '' ) ; } if ( responsePacket ) { this . remoteKnowsOurAccount = true ; this . handleControlFrames ( responsePacket . frames ) ; return responsePacket ; } else { return ilpReject ; } } /**
     * Send a ConnectionClose frame to the other side
     */ protected async sendConnectionClose ( err ? ) { if ( this . remoteState === RemoteState . Closed ) { this . log . debug ( 'not sending connection error because remote is already closed' ) ; return ; } let errorCode ; let errorMessage ; if ( err && err instanceof ConnectionError ) { errorCode = err . streamErrorCode ; errorMessage = err . message ; } else if ( err ) { errorCode = ErrorCode . InternalError ; errorMessage = err . message ; } else { errorCode = ErrorCode . NoError ; errorMessage = '' ; } const packet = new Packet ( this . nextPacketSequence ++ , IlpPacketType . Prepare , 0 , [ new ConnectionCloseFrame ( errorCode , errorMessage ) ] ) ; try { const prepare = { destination : this . _destinationAccount ! , amount : '0' , data : await packet . serializeAndEncrypt ( this . _pskKey ) , executionCondition : cryptoHelper . generateRandomCondition ( ) , expiresAt : this . getExpiry ( this . _destinationAccount ! ) } ; await this . plugin . sendData ( IlpPacket . serializeIlpPrepare ( prepare ) ) ; } catch ( err ) { this . log . error ( 'error while trying to inform peer that connection is closing, but closing anyway' , err ) ; } this . remoteState = RemoteState . Closed ; } /**
     * Helper function used to send all ILP Prepare packets.
     * This automatically generates the condition and sets the packet expiry.
     * It also ensures that responses are valid and match the outgoing request.
     */ protected async sendPacket ( packet , sourceAmount , unfulfillable = false ) { this . log . trace ( 'sending packet %s with source amount: %s: %j' , packet . sequence , sourceAmount , packet ) ; const data = await packet . serializeAndEncrypt ( this . _pskKey , ( this . enablePadding ? MAX_DATA_SIZE : undefined ) ) ; let fulfillment ; let executionCondition ; if ( unfulfillable ) { fulfillment = undefined ; executionCondition = cryptoHelper . generateRandomCondition ( ) ; } else { fulfillment = await cryptoHelper . generateFulfillment ( this . _fulfillmentKey , data ) ; executionCondition = await cryptoHelper . hash ( fulfillment ) ; } const prepare = { destination : this . _destinationAccount ! , amount : sourceAmount . toString ( ) , data , executionCondition , expiresAt : this . getExpiry ( this . _destinationAccount ! ) } ; const responseData = await this . plugin . sendData ( IlpPacket . serializeIlpPrepare ( prepare ) ) ; this . bumpIdle ( ) ; let response ; try { if ( responseData [ 0 ] === IlpPacket . Type . TYPE_ILP_FULFILL ) { response = IlpPacket . deserializeIlpFulfill ( responseData ) ; } else if ( responseData [ 0 ] === IlpPacket . Type . TYPE_ILP_REJECT ) { response = IlpPacket . deserializeIlpReject ( responseData ) ; } else { throw new Error ( `Invalid response packet type: ${ responseData [ 0 ] }` ) ; } } catch ( err ) { this . log . error ( 'got invalid response from sending packet %s: %s %h' , packet . sequence , err , responseData ) ; throw new Error ( `Invalid response when sending packet ${ packet . sequence }: ${ err . message }` ) ; } if ( fulfillment && isFulfill ( response ) ) { if ( ! ( await cryptoHelper . hash ( response . fulfillment ) ) . equals ( executionCondition ) ) { this . log . error ( 'got invalid fulfillment for packet %s: %h. expected: %h for condition: %h' , packet . sequence , response . fulfillment , fulfillment , executionCondition ) ; throw new Error ( `Got invalid fulfillment for packet ${ packet . sequence }. Actual: ${ response . fulfillment . toString ( 'hex' ) }, expected: ${ fulfillment . toString ( 'hex' ) }` ) ; } } else { response = response as IlpPacket . IlpReject ; this . undoRejectedPacket ( packet ) ; if ( response . code !== 'F99' ) { return this . handleConnectorError ( response , sourceAmount ) ; } } if ( response . data . length === 0 ) { return undefined ; } let responsePacket ; try { responsePacket = await Packet . decryptAndDeserialize ( this . _pskKey , response . data ) ; } catch ( err ) { this . log . error ( 'unable to decrypt and parse response data: %s %h' , err ) ; throw new Error ( 'Unable to decrypt and parse response data: ' + err . message ) ; } if ( ! responsePacket . sequence . equals ( packet . sequence ) ) { this . log . error ( 'response packet sequence does not match the request packet. expected sequence: %s, got response packet: %j' , packet . sequence , responsePacket ) ; throw new Error ( `Response packet sequence does not correspond to the request. Actual: ${ responsePacket . sequence }, expected: ${ packet . sequence }` ) ; } if ( responsePacket . ilpPacketType !== responseData [ 0 ] ) { this . log . error ( 'response packet was on wrong ILP packet type. expected ILP packet type: %d, got: %j' , responseData [ 0 ] , responsePacket ) ; throw new Error ( `Response says it should be on an ILP packet of type: ${ responsePacket . ilpPacketType } but it was carried on an ILP packet of type: ${ responseData [ 0 ] }` ) ; } this . log . debug ( 'got response to packet: %s: %j' , packet . sequence , responsePacket ) ; return responsePacket ; } private protected undoRejectedPacket ( requestPacket ) { this . log . debug ( 'packet %s was rejected' , requestPacket . sequence ) ; for ( let frame of requestPacket . frames ) { switch ( frame . type ) { case FrameType . StreamMoney : const stream1 = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( stream1 ) stream1 . _cancelHold ( requestPacket . sequence . toString ( ) ) ; break ; case FrameType . StreamData : const stream2 = this . streams . get ( frame . streamId . toNumber ( ) ) ; if ( stream2 ) stream2 . _resendOutgoingData ( frame . data , frame . offset . toNumber ( ) ) ; break ; case FrameType . StreamClose : this . queuedFrames . push ( frame ) ; break ; default : continue ; } } } private protected async handleConnectorError ( reject , amountSent ) { this . log . debug ( 'handling reject triggered by: %s error: %s message: %s data: %h' , reject . triggeredBy , reject . code , reject . message , reject . data ) ; if ( reject . code === 'F08' ) { const maximumPacketAmount = this . congestion . onAmountTooLargeError ( reject , amountSent ) ; if ( maximumPacketAmount . equals ( 0 ) ) { this . log . error ( 'cannot send anything through this path. the maximum packet amount is 0' ) ; throw new Error ( 'Cannot send. Path has a Maximum Packet Amount of 0' ) ; } } else if ( reject . code [ 0 ] === 'T' ) { if ( reject . code === 'T04' ) { this . congestion . onInsufficientLiquidityError ( reject , amountSent ) ; } this . log . warn ( 'got %s temporary error triggered by: %s. waiting %dms before trying again' , reject . code , reject . triggeredBy , this . retryDelay ) ; const delay = this . retryDelay ; this . retryDelay = Math . min ( this . retryDelay * 2 , RETRY_DELAY_MAX ) ; await new Promise ( ( resolve , reject ) => setTimeout ( resolve , delay ) ) ; } else { this . log . error ( 'unexpected error. code: %s, triggered by: %s, message: %s, data: %h' , reject . code , reject . triggeredBy , reject . message , reject . data ) ; const error = new Error ( `Unexpected error while sending packet. Code: ${ reject . code }, triggered by: ${ reject . triggeredBy }, message: ${ reject . message }` ) ; error [ 'ilpReject' ] = reject ; throw error ; } } protected safeEmit ( event , ... args ) { try { args . unshift ( event ) ; this . emit . apply ( this , args ) ; } catch ( err ) { this . log . debug ( 'error in %s handler: %s' , event , err ) ; } } protected getOutgoingOffsets ( ) { let currentOffset = 0 ; let maxOffset = 0 ; for ( let [ _ , stream ] of this . streams ) { const streamOffsets = stream . _getOutgoingOffsets ( ) ; currentOffset += streamOffsets . current ; maxOffset += streamOffsets . max ; } return { currentOffset , maxOffset } ; } protected getIncomingOffsets ( ) { let totalMaxOffset = 0 ; let totalReadOffset = 0 ; let totalBufferedData = 0 ; for ( let [ _ , stream ] of this . streams ) { const { max , current } = stream . _getIncomingOffsets ( ) ; totalMaxOffset += max ; totalReadOffset += current ; totalBufferedData += stream . readableLength ; } return { current : totalReadOffset , max : totalMaxOffset , maxAcceptable : totalReadOffset - totalBufferedData + this . maxBufferedData } ; } protected removeStreamRecord ( stream ) { this . log . debug ( 'removing record of stream %d' , stream . id ) ; this . streams . delete ( stream . id ) ; this . closedStreams . add ( stream . id ) ; if ( ! stream . _sentEnd ) { stream . _sentEnd = true ; const streamEndFrame = ( stream . _errorMessage ? new StreamCloseFrame ( stream . id , ErrorCode . ApplicationError , stream . _errorMessage ) : new StreamCloseFrame ( stream . id , ErrorCode . NoError , '' ) ) ; this . queuedFrames . push ( streamEndFrame ) ; setImmediate ( ( ) => this . startSendLoop ( ) ) ; } } private startIdleTimer ( ) { if ( this . idleTimeout === 0 ) return ; const idle = Date . now ( ) - this . lastActive . getTime ( ) ; this . idleTimer = setTimeout ( ( ) => this . testIdle ( ) , this . idleTimeout - idle ) ; if ( typeof this . idleTimer . unref === 'function' ) { this . idleTimer . unref ( ) ; } this . log . trace ( '(re)starting idle timeout for %dms from now' , this . idleTimeout ) ; } private testIdle ( ) { const idle = Date . now ( ) - this . lastActive . getTime ( ) ; if ( idle >= this . idleTimeout ) { this . log . error ( 'Connection timed out due to inactivity, destroying connection' ) ; this . destroy ( new Error ( 'Connection timed out due to inactivity' ) ) ; } else { this . startIdleTimer ( ) ; } } private bumpIdle ( ) { this . lastActive = new Date ( ) ; } private addIncomingHold ( value ) { let result = checkedAdd ( this . _totalReceived , this . _incomingHold ) ; result = checkedAdd ( result . sum , value ) ; if ( result . overflow ) { const err = new IlpPacket . Errors . BadRequestError ( 'Total received exceeded MaxUint64' ) ; err [ 'ilpErrorMessage' ] = err . message ; this . destroy ( err ) ; throw err ; } else { this . _incomingHold = result . sum ; } } private removeIncomingHold ( value ) { this . _incomingHold = checkedSubtract ( this . _incomingHold , value ) . difference ; } private addTotalReceived ( value ) { this . _totalReceived = checkedAdd ( this . _totalReceived , value ) . sum ; } private addTotalSent ( value ) { const result = checkedAdd ( this . _totalSent , value ) ; if ( result . overflow ) { const err = new IlpPacket . Errors . BadRequestError ( 'Total sent exceeded MaxUint64' ) ; err [ 'ilpErrorMessage' ] = err . message ; this . destroy ( err ) ; throw err ; } else { this . _totalSent = result . sum ; } } private addTotalDelivered ( value ) { const result = checkedAdd ( this . _totalDelivered , value ) ; if ( result . overflow ) { const err = new IlpPacket . Errors . BadRequestError ( 'Total delivered exceeded MaxUint64' ) ; err [ 'ilpErrorMessage' ] = err . message ; this . destroy ( err ) ; throw err ; } else { this . _totalDelivered = result . sum ; } } private getNextPacketSequence ( ) { const sequence = this . nextPacketSequence ++ ; if ( sequence >= 2 ** 31 ) { throw new ConnectionError ( 'Connection exceeded maximum number of packets' , ErrorCode . InternalError ) ; } return sequence ; } private maybePushAccountFrames ( requestPacket ) { if ( ! this . remoteKnowsOurAccount ) { requestPacket . frames . push ( new ConnectionNewAddressFrame ( this . _sourceAccount ) ) ; requestPacket . frames . push ( new ConnectionAssetDetailsFrame ( this . _sourceAssetCode , this . _sourceAssetScale ) ) ; } } } function isFulfill ( packet ) { return packet . hasOwnProperty ( 'fulfillment' ) ; }