connection.ts

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 // 12 hours should be long enough
const RETRY_DELAY_INCREASE_FACTOR = 1.5
const DEFAULT_PACKET_TIMEOUT = 30000
const DEFAULT_IDLE_TIMEOUT = 60000 // 1 minute
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`]{@link Server#generateAddressAndSecret} */
  connectionTag?: string,
  /** User-specified receipt nonce that was passed into [`generateAddressAndSecret`]{@link Server#generateAddressAndSecret} */
  receiptNonce?: Buffer,
  /** User-specified receipt secret that was passed into [`generateAddressAndSecret`]{@link Server#generateAddressAndSecret} */
  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`]{@link Server#generateAddressAndSecret} */
  sharedSecret: Buffer
}

interface NewConnectionOpts extends BuildConnectionOpts {
  pskKey: Buffer,
  fulfillmentKey: Buffer,
}

export class ConnectionError extends Error {
  streamErrorCode: ErrorCode

  constructor (message: string, streamErrorCode?: ErrorCode) {
    super(message)
    this.streamErrorCode = streamErrorCode || ErrorCode.InternalError
  }
}

enum RemoteState {
  Init,
  Connected,
  Closed
}

function defaultGetExpiry (): Date {
  return new Date(Date.now() + DEFAULT_PACKET_TIMEOUT)
}

/**
 * Class representing the connection between a [`Client`]{@link createConnection} and a [`Server`]{@link Server}.
 * A single connection can be used to send or receive on [Streams]{@link DataAndMoneyStream}.
 *
 * Streams are created using the [`createStream`]{@link 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 // whether there is a running send-loop
  protected congestion: CongestionController
  protected minExchangeRatePrecision: number
  protected connected: boolean
  // Set to `true`:
  // - Until `'connect'` event.
  // - When a `ConnectionCloseFrame` is received.
  // - When `end()` is called, but before sending the `ConnectionCloseFrame` to remote.
  protected closed: boolean
  // Set to `true` when either `end()` or `destroy()` has finished.
  protected done: boolean = false
  protected exchangeRate?: Rational
  protected retryDelay: number
  protected queuedFrames: Frame[]

  protected remoteState: RemoteState = RemoteState.Init
  protected remoteMaxStreamId: number
  protected remoteKnowsOurAccount: boolean

  // TODO use longs for byte offsets
  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: NewConnectionOpts) {
    super()

    // Use the same connectionId for logging on both client & server
    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
    // TODO should streams be a Map or just an object?
    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: BuildConnectionOpts): Promise<Connection> {
    const pskKey = await cryptoHelper.generatePskEncryptionKey(opts.sharedSecret)
    const fulfillmentKey = await cryptoHelper.generateFulfillmentKey(opts.sharedSecret)
    return new Connection(Object.assign({ pskKey, fulfillmentKey }, opts))
  }

  /**
   * New incoming stream event
   * @event stream
   * @type {DataAndMoneyStream}
   */

  /**
   * Start sending or receiving.
   * @fires stream
   */
  async connect (): Promise<void> {
    if (!this.closed) {
      return Promise.resolve()
    }
    /* tslint:disable-next-line:no-floating-promises */
    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?: 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
   */
  // TODO should this be sync or async?
  async end (): Promise<void> {
    this.log.info('closing connection')
    // Create Promises on each stream that resolve on the 'end' event so
    // we can wait for them all to be completed before closing the connection
    let streamEndPromises: Promise<any>[] = []
    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)

      /* tslint:disable-next-line:no-floating-promises */
      this.startSendLoop()
    })
    // Wait for the send loop to finish & all the streams to end
    // before marking the connection as closed so the streams
    // can finish sending data or money.
    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.
   *
   * This function must *never* return a rejection, since `startSendLoop` is run
   * as a floating promise and returns `destroy()` on errors, so the rejection
   * would be unhandled.
   */
  // TODO should this be sync or async?
  async destroy (err?: Error): Promise<void> {
    this.log.error('destroying connection with error:', err)
    // Don't destroy the connection (most importantly: don't emit "close" & "error"
    // if the connection has already done so, since it will confuse clients.
    if (this.done) return

    if (err) {
      this.safeEmit('error', err)
    }
    // Create Promises on each stream that resolve on the 'close' event so
    // we can wait for them all to be completed before closing the connection
    let streamClosePromises: Promise<any>[] = []
    for (let [_, stream] of this.streams) {
      streamClosePromises.push(new Promise((resolve, reject) => {
        stream.on('close', resolve)
      }))
      // TODO should we pass the error to each stream?
      stream.destroy()
    }
    // Send an error to ensure that the other side doesn't get a NoError.
    await this.sendConnectionClose(err || new ConnectionError('Connection destroyed', ErrorCode.ApplicationError))
    // wait for all the streams to be closed before emitting the connection 'close'
    await Promise.all(streamClosePromises)
    this.safeEmit('close')
    this.stopTimers()
  }

  /**
   * Returns a new bidirectional [`DataAndMoneyStream`]{@link DataAndMoneyStream}
   */
  createStream (): DataAndMoneyStream {
    // Make sure we don't open more streams than the remote will allow
    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`)
    }

    // TODO should this inform the other side?
    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()
  }

  /**
   * (Internal) Handle incoming ILP Prepare packets.
   * This will automatically fulfill all valid and expected Prepare packets, or
   * defer to custom application logic using the `shouldFulfill` callback, if provided.
   * It passes the incoming money and/or data to the relevant streams.
   * @private
   */
  async handlePrepare (prepare: IlpPacket.IlpPrepare): Promise<IlpPacket.IlpFulfill> {
    // Parse packet
    let requestPacket: Packet
    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: Frame[] = []

    // Tell peer how much data connection can receive
    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)))
    }

    // Handle new streams
    for (let frame of requestPacket.frames) {
      if (frame.type === FrameType.StreamMoney
        || frame.type === FrameType.StreamData
        // TODO should frames that set the max values open the stream?
        || frame.type === FrameType.StreamMaxMoney
        || frame.type === FrameType.StreamMaxData) {
        const streamId = frame.streamId.toNumber()

        // Check if the stream was already closed
        if (this.closedStreams.has(streamId)) {
          this.log.trace('got packet with frame for stream %d, which was already closed', streamId)

          // Don't bother sending an error frame back unless they've actually sent money or data
          if (frame.type !== FrameType.StreamMoney && frame.type !== FrameType.StreamData) {
            continue
          }

          // Respond with a StreamClose frame (unless there is already one queued)
          const testStreamClose = (frame: 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 {
          // Note this will throw if the stream was already closed
          this.handleNewStream(frame.streamId.toNumber())
        } catch (err) {
          this.log.debug('error handling new stream %s: %s', frame.streamId, err && err.message)
          await throwFinalApplicationError()
        }
      }
    }

    // TODO don't throw errors in expected cases -- they are slower than just returning a value
    try {
      this.handleControlFrames(requestPacket.frames)
    } catch (err) {
      this.log.debug('error handling frames:', err && err.message)
      await throwFinalApplicationError()
    }

    // TODO keep a running total of the offsets so we don't need to recalculate each time
    const incomingOffsets = this.getIncomingOffsets()
    if (incomingOffsets.max > incomingOffsets.maxAcceptable) {
      /* tslint:disable-next-line:no-floating-promises */
      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()
    }

    // Ensure we can generate correct fulfillment
    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()
    }

    // Determine amount to receive on each frame
    const amountsToReceive: { stream: DataAndMoneyStream, amount: Long }[] = []
    const totalMoneyShares = requestPacket.frames.reduce((sum: Long, frame: 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()
      // TODO make sure we don't lose any because of rounding issues
      const streamAmount = multiplyDivideFloor(
        incomingAmount, frame.shares, totalMoneyShares)
      const stream = this.streams.get(streamId)!
      amountsToReceive.push({
        stream,
        amount: streamAmount
      })

      // Ensure that this amount isn't more than the stream can receive
      const maxStreamCanReceive = this.allowableReceiveExtra
        .multiplyByLongCeil(stream._getAmountStreamCanReceive())
      if (maxStreamCanReceive.lessThan(streamAmount)) {
        // TODO should this be distributed to other streams if it can be?
        this.log.debug('peer sent too much for stream: %d. got: %s, max receivable: %s', streamId, streamAmount, maxStreamCanReceive)
        // Tell peer how much the streams they sent for can receive
        responseFrames.push(new StreamMaxMoneyFrame(streamId, stream.receiveMax, stream.totalReceived))

        // TODO include error frame
        await throwFinalApplicationError()
      }

      // Reject the packet if any of the streams is already closed
      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)

    // Allow consumer to choose to fulfill each packet and/or perform other logic before fulfilling
    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()
      })
    }

    // Add incoming amounts to each stream
    const totalsReceived: Map<number, string> = new Map()
    for (let { stream, amount } of amountsToReceive) {
      stream._addToIncoming(amount, prepare)
      totalsReceived.set(stream.id, stream.totalReceived)
    }

    // Tell peer about closed streams and how much each stream can receive
    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, ''))
          }
          // TODO confirm that they get this
          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))

          // TODO only send these frames when we need to
          responseFrames.push(new StreamMaxDataFrame(stream.id, stream._getIncomingOffsets().maxAcceptable))
        }
      }
    }

    // Add receipt frame(s)
    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
        })))
      }
    }

    // TODO make sure the queued frames aren't too big
    responseFrames = responseFrames.concat(this.queuedFrames)
    this.queuedFrames = []

    // Return fulfillment and response packet
    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: Frame[]): void {
    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)

          // Tell the other side our max stream id and asset details
          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')
          }
          // TODO reset the exchange rate and send a test packet to make sure they haven't spoofed the address
          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:
          // TODO end the connection in some other way
          this.sending = false
          this.closed = true
          this.remoteState = RemoteState.Closed
          if (frame.errorCode === ErrorCode.NoError) {
            this.log.info('remote closed connection')
            /* tslint:disable-next-line:no-floating-promises */
            this.end().catch((err) => {
              this.log.warn('close failed with error=%s', err)
              return this.destroy()
            })
          } else {
            this.log.error('remote connection error. code: %s, message: %s', ErrorCode[frame.errorCode], frame.errorMessage)
            /* tslint:disable-next-line:no-floating-promises */
            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 {
            // We assumed their size was 64kb but it turned out to be less
            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:
          // TODO make sure the number isn't lowered
          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)) {
            /* tslint:disable-next-line:no-floating-promises */
            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())

          // Make sure the peer hasn't exceeded the flow control limits
          const incomingOffsets = stream._getIncomingOffsets()
          if (incomingOffsets.max > incomingOffsets.maxAcceptable) {
            /* tslint:disable-next-line:no-floating-promises */
            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
            /* tslint:disable-next-line:no-floating-promises */
            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: number): void {
    if (this.streams.has(streamId) || this.closedStreams.has(streamId)) {
      return
    }

    // Validate stream ID
    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`))
      // TODO this should probably call this.destroy
      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
    }

    // Make sure there aren't too many open streams
    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
    }

    // Let the other side know if they're getting close to the number of streams
    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: StreamCloseFrame) {
    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)
    // TODO should we confirm with the other side that we closed it?
    stream._sentEnd = true
    let err
    if (frame.errorMessage) {
      err = new Error(frame.errorMessage)
      err.name = ErrorCode[frame.errorCode]
    }
    stream._remoteEnded(err)

    // TODO make sure we don't send more than one of these frames per packet
    this.maxStreamId += 2
    this.log.trace('raising maxStreamId to %d', this.maxStreamId)
    this.queuedFrames.push(new ConnectionMaxStreamIdFrame(this.maxStreamId))
    // Start send loop to make sure this frame is sent
    /* tslint:disable-next-line:no-floating-promises */
    this.startSendLoop()
  }

  /**
   * (Internal) Start sending packets with money and/or data, as necessary.
   * @private
   */
  protected async startSendLoop () {
    if (this.looping) {
      // `sending` may be `false`, but the loop has not yet exited.
      // Setting `true` ensures that it will try to loop at least once more before stopping,
      // ensuring whatever changes triggered this `startSendLoop` call are sent.
      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 {
          // TODO Send multiple packets at the same time (don't await promise)
          // TODO Figure out if we need to wait before sending the next one
          await this.loadAndSendPacket()
        }
      }
      this.looping = false
    } catch (err) {
      this.looping = false
      // TODO should a connection error be an error on all of the streams?
      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')
    }
  }

  /**
   * Load up a packet money and/or data, send it to the other party, and handle the result.
   * @private
   */
  protected async loadAndSendPacket (): Promise<void> {
    // Actually send on the next tick of the event loop in case multiple streams
    // have their limits raised at the same time
    await new Promise((resolve, reject) => setImmediate(resolve))

    this.log.trace('loadAndSendPacket')
    let amountToSend = Long.UZERO

    // TODO make sure the queued frames aren't too big
    const requestPacket = new Packet(this.getNextPacketSequence(), IlpPacketType.Prepare, undefined, this.queuedFrames)
    this.queuedFrames = []

    // This is usually handled by the first test packet, but when the exchange rate
    // is fixed it happens here instead.
    this.maybePushAccountFrames(requestPacket)

    // Send control frames
    // TODO only send the max amount when it changes
    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) {
      // TODO how do we know if there was an error?
      this.log.trace('sending connection close frame')
      requestPacket.frames.push(new ConnectionCloseFrame(ErrorCode.NoError, ''))
      // TODO don't put any more frames because the connection is closed
      // TODO only mark this as closed once we confirm that with the receiver
      this.remoteState = RemoteState.Closed
    }

    // Determine how much to send based on amount frames and path maximum packet amount
    let maxAmountFromNextStream = this.congestion.testMaximumPacketAmount
    if (this.exchangeRate!.greaterThanOne()) {
      // Ensure that the packet's PrepareAmount will never be larger than MAX_UNSIGNED_VALUE.
      maxAmountFromNextStream = minLong(
        maxAmountFromNextStream,
        this.exchangeRate!.reciprocal()
          .multiplyByLong(Long.MAX_UNSIGNED_VALUE))
    }
    const streamsSentFrom = []
    for (let [_, stream] of this.streams) {
      if (stream._sentEnd) {
        // TODO just remove closed streams?
        continue
      }
      // Determine how much to send from this stream based on how much it has available
      // and how much the receiver side of this stream can receive
      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)

      // Hold the money and add a frame to the packet
      if (amountToSendFromStream.greaterThan(0)) {
        stream._holdOutgoing(requestPacket.sequence.toString(), amountToSendFromStream)
        // TODO make sure the length of the frames doesn't exceed packet data limit
        requestPacket.frames.push(new StreamMoneyFrame(stream.id, amountToSendFromStream))
        amountToSend = amountToSend.add(amountToSendFromStream)
        maxAmountFromNextStream = maxAmountFromNextStream.subtract(amountToSendFromStream)
        streamsSentFrom.push(stream)
      }

      // Tell peer if they're blocking us from sending money
      const amountLeftStreamWantsToSend = Long.fromString(stream.sendMax, true)
        .subtract(stream.totalSent)
        .subtract(amountToSendFromStream)
      /* tslint:disable-next-line:no-unnecessary-type-assertion */
      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)) {
        // TODO make sure that we start with those later frames the next time around
        break
      }
    }

    let bytesLeftInPacket = MAX_DATA_SIZE - requestPacket.byteLength()

    // Respect connection-level flow control
    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) {
      // TODO use a sensible estimate for the StreamDataFrame overhead
      if (bytesLeftInPacket - 20 <= 0) {
        // Never pass a negative offset to _getAmountAvailableToSend.
        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)
      }

      // Inform remote which streams are blocked
      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))
      }
    }

    // Check if we can stop sending
    if (amountToSend.equals(0)) {
      if (requestPacket.frames.length === 0) {
        this.sending = false
        return
      } else {
        // Check if any Close, Data, or Money Frames are present in the packet.
        // If any of those are do not sent sending to false so the send loop
        // has an opportunity to retry if those packets are rejected.
        if (!requestPacket.frames.find(frame =>
            ((frame.type === FrameType.StreamClose)
            || (frame.type === FrameType.StreamData)
            || (frame.type === FrameType.StreamMoney)))) {
          this.sending = false
        }
      }
    }

    // Set minimum destination amount
    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)

      // Track the exchange rate for the last packet (whether it was fulfilled or rejected)
      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())
        }

        // Update stats based on amount sent
        this.addTotalDelivered(responsePacket.prepareAmount)
        this.addTotalSent(amountToSend)
        this.congestion.onFulfill(amountToSend)

        // Reset the retry delay
        this.retryDelay = RETRY_DELAY_START
      }
    }
  }

  /**
   * (Internal) Send volley of test packets to find the exchange rate, its precision, and potential other amounts to try.
   * @private
   */
  protected async sendTestPacketVolley (testPacketAmounts: number[]): Promise<any> {
    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
      }
    }))

    // parse F08 packets and get the max packet amounts from them
    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
    })

    // Figure out which test packet discovered the exchange rate with the most precision and gather packet error codes
    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 (): Promise<void> {
    if (!this.exchangeRate) {
      this.log.trace('determining exchange rate')
      await this.determineExchangeRate()
    }
    // 'connect' is not emitted immediately because `connect()` needs to be listening,
    // and when `exchangeRate` is fixed it isn't yet.
    process.nextTick(() => {
      this.safeEmit('connect')
      this.log.trace('connected')
    })
  }

  /**
   * (Internal) Probe using test packets to find the exchange rate.
   * @private
   */
  protected async determineExchangeRate (): Promise<void> {
    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

    // set a max attempts in case F08 & TXX errors keep occurring
    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
      }

      // If we get here the first volley failed, try new volley using all unique packet amounts based on the max packets
      testPacketAmounts = maxPacketAmounts
        .filter((amount: any) => !amount.equals(Long.MAX_UNSIGNED_VALUE))
        .reduce((acc: any, curr: any) => [...new Set([...acc, curr.toString()])], [])

      // Check for any Txx Errors
      if (packetErrors.some((error: any) => error.code[0] === 'T')) {
        // Find the smallest packet amount we tried in case we ran into Txx errors
        const smallestPacketAmount = packetErrors.reduce((min: Long, error: any) => {
          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 (): void {
    if (this.rateRetryTimer) this.rateRetryTimer.stop()
    clearTimeout(this.idleTimer)
    this.done = true
  }

  /**
   * (Internal) Send an unfulfillable test packet. Primarily used for determining the path exchange rate.
   * @private
   */
  protected async sendTestPacket (amount: Long, timeout = DEFAULT_PACKET_TIMEOUT): Promise<Packet | IlpPacket.IlpReject | null> {
    // Set packet number to correlate response with request
    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!)
    }

    /* tslint:disable-next-line:no-unnecessary-type-assertion */
    const responseData = await (new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.log.error('test packet %s timed out before we got a response', requestPacket.sequence)
        resolve(null)
      }, timeout)
      this.plugin.sendData(IlpPacket.serializeIlpPrepare(prepare))
        .then((result) => {
          clearTimeout(timer)
          resolve(result)
        })
        .catch(reject)
    }) as Promise<Buffer | null>)

    if (!responseData) {
      return null
    }
    this.bumpIdle()

    const ilpReject = IlpPacket.deserializeIlpReject(responseData)

    // Return the receiver's response if there was one
    let responsePacket
    if (ilpReject.code === 'F99' && ilpReject.data.length > 0) {
      responsePacket = await Packet.decryptAndDeserialize(this._pskKey, ilpReject.data)

      // Ensure the response corresponds to the request
      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?: ConnectionError | Error): Promise<void> {
    if (this.remoteState === RemoteState.Closed) {
      this.log.debug('not sending connection error because remote is already closed')
      return
    }

    let errorCode: 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: Packet, sourceAmount: Long, unfulfillable = false): Promise<Packet | void> {
    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: Buffer | undefined
    let executionCondition: Buffer
    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: IlpPacket.IlpFulfill | IlpPacket.IlpReject
    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}`)
    }

    // Handle fulfillment
    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)
      }
    }

    // TODO correctly handle fulfills that come back without data attached (this will be treated like a reject)
    if (response.data.length === 0) {
      return undefined
    }

    // Parse response data from receiver
    let responsePacket: Packet
    try {
      responsePacket = await Packet.decryptAndDeserialize(this._pskKey, response.data)
    } catch (err) {
      this.log.error('unable to decrypt and parse response data: %s %h', err)
      // TODO should we continue processing anyway? what if it was fulfilled?
      throw new Error('Unable to decrypt and parse response data: ' + err.message)
    }

    // Ensure the response corresponds to the request
    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
  }

  /**
   * Roll back the effects of an outgoing packet that was rejected
   * @private
   */
  protected undoRejectedPacket (requestPacket: Packet) {
    this.log.debug('packet %s was rejected', requestPacket.sequence)

    // TODO resend control frames
    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
      }
    }
  }

  /**
   * (Internal) Handle final and temporary errors that were not generated by the receiver.
   * @private
   */
  protected async handleConnectorError (reject: IlpPacket.IlpReject, amountSent: Long) {
    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)
      }

      // TODO should we reduce the packet amount on other TXX errors too?
      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)
      // This error will terminate the connection and bubble out to the caller.
      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: string, ...args: any[]) {
    try {
      args.unshift(event)
      this.emit.apply(this, args)
    } catch (err) {
      this.log.debug('error in %s handler: %s', event, err)
    }
  }

  protected getOutgoingOffsets (): { currentOffset: number, maxOffset: number } {
    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 (): { current: number, max: number, maxAcceptable: number } {
    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: DataAndMoneyStream) {
    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 (): void {
    if (this.idleTimeout === 0) return
    const idle = Date.now() - this.lastActive.getTime()
    this.idleTimer = setTimeout(() => this.testIdle(), this.idleTimeout - idle)
    // browser timers don't support unref
    /* tslint:disable-next-line:strict-type-predicates */
    if (typeof this.idleTimer.unref === 'function') {
      this.idleTimer.unref()
    }
    this.log.trace('(re)starting idle timeout for %dms from now', this.idleTimeout)
  }

  private testIdle (): void {
    const idle = Date.now() - this.lastActive.getTime()
    if (idle >= this.idleTimeout) {
      this.log.error('Connection timed out due to inactivity, destroying connection')
      /* tslint:disable-next-line:no-floating-promises */
      this.destroy(new Error('Connection timed out due to inactivity'))
    } else {
      this.startIdleTimer()
    }
  }

  private bumpIdle (): void { this.lastActive = new Date() }

  private addIncomingHold (value: Long): void {
    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
      /* tslint:disable-next-line:no-floating-promises */
      this.destroy(err)
      throw err
    } else {
      this._incomingHold = result.sum
    }
  }

  private removeIncomingHold (value: Long): void {
    // As long as this is called after `addIncomingHold` for the same amount,
    // this should never underflow
    this._incomingHold = checkedSubtract(this._incomingHold, value).difference
  }

  private addTotalReceived (value: Long): void {
    // As long as this is called after `addIncomingHold` for the same amount,
    // this should never overflow
    this._totalReceived = checkedAdd(this._totalReceived, value).sum
  }

  private addTotalSent (value: Long): void {
    const result = checkedAdd(this._totalSent, value)
    if (result.overflow) {
      const err = new IlpPacket.Errors.BadRequestError('Total sent exceeded MaxUint64')
      err['ilpErrorMessage'] = err.message
      /* tslint:disable-next-line:no-floating-promises */
      this.destroy(err)
      throw err
    } else {
      this._totalSent = result.sum
    }
  }

  private addTotalDelivered (value: Long): void {
    const result = checkedAdd(this._totalDelivered, value)
    if (result.overflow) {
      const err = new IlpPacket.Errors.BadRequestError('Total delivered exceeded MaxUint64')
      err['ilpErrorMessage'] = err.message
      /* tslint:disable-next-line:no-floating-promises */
      this.destroy(err)
      throw err
    } else {
      this._totalDelivered = result.sum
    }
  }

  private getNextPacketSequence (): number {
    const sequence = this.nextPacketSequence++
    if (sequence >= 2 ** 31) {
      // Destroy the connection when too many packets are sent (see https://github.com/interledger/rfcs/blob/master/0029-stream/0029-stream.md#513-maximum-number-of-packets-per-connection).
      //
      // Throwing here will abort the current send and `destroy()` the connection.
      throw new ConnectionError('Connection exceeded maximum number of packets', ErrorCode.InternalError)
    }
    return sequence
  }

  private maybePushAccountFrames (requestPacket: Packet) {
    if (!this.remoteKnowsOurAccount) {
      // TODO attach a token to the account?
      requestPacket.frames.push(new ConnectionNewAddressFrame(this._sourceAccount))
      requestPacket.frames.push(new ConnectionAssetDetailsFrame(this._sourceAssetCode, this._sourceAssetScale))
    }
  }
}

function isFulfill (packet: IlpPacket.IlpFulfill | IlpPacket.IlpReject): packet is IlpPacket.IlpFulfill {
  return packet.hasOwnProperty('fulfillment')
}