import { createHmac , randomBytes } from "crypto" ; import debug from "debug" ; import { jspack } from "jspack" ; import { range } from "lodash" ; import { Event } from "rx.mini" ; import { random32 , uint16Add , uint16Gt , uint32Gt , uint32Gte , } from "../../common/src" ; import { AbortChunk , Chunk , CookieAckChunk , CookieEchoChunk , DataChunk , ErrorChunk , ForwardTsnChunk , HeartbeatAckChunk , HeartbeatChunk , InitAckChunk , InitChunk , parsePacket , ReConfigChunk , ReconfigChunk , SackChunk , serializePacket , ShutdownAckChunk , ShutdownChunk , ShutdownCompleteChunk , } from "./chunk" ; import { SCTP_STATE } from "./const" ; import { createEventsFromList , enumerate , Unpacked } from "./helper" ; import { OutgoingSSNResetRequestParam , RECONFIG_PARAM_BY_TYPES , ReconfigResponseParam , reconfigResult , StreamAddOutgoingParam , StreamParam , } from "./param" ; import { Transport } from "./transport" ; const log = debug ( "werift/sctp/sctp" ) ; const COOKIE_LENGTH = 24 ; const COOKIE_LIFETIME = 60 ; const MAX_STREAMS = 65535 ; const USERDATA_MAX_LENGTH = 1200 ; const SCTP_DATA_LAST_FRAG = 0x01 ; const SCTP_DATA_FIRST_FRAG = 0x02 ; const SCTP_DATA_UNORDERED = 0x04 ; const SCTP_MAX_ASSOCIATION_RETRANS = 10 ; const SCTP_MAX_INIT_RETRANS = 8 ; const SCTP_RTO_ALPHA = 1 / 8 ; const SCTP_RTO_BETA = 1 / 4 ; const SCTP_RTO_INITIAL = 3 ; const SCTP_RTO_MIN = 1 ; const SCTP_RTO_MAX = 60 ; const SCTP_TSN_MODULO = 2 ** 32 ; const RECONFIG_MAX_STREAMS = 135 ; const SCTP_STATE_COOKIE = 0x0007 ; const SCTP_SUPPORTED_CHUNK_EXT = 0x8008 ; const SCTP_PRSCTP_SUPPORTED = 0xc000 ; const SCTPConnectionStates = [ "new" , "closed" , "connected" , "connecting" , ] as const ; type SCTPConnectionState = Unpacked < typeof SCTPConnectionStates > ; export class SCTP { readonly stateChanged : { [ key in SCTPConnectionState ] : Event < [ ] > ; } = createEventsFromList ( SCTPConnectionStates ) ; readonly onReconfigStreams = new Event < [ number [ ] ] > ( ) ; /**streamId: number, ppId: number, data: Buffer */ readonly onReceive = new Event < [ number , number , Buffer ] > ( ) ; onSackReceived : ( ) => Promise < void > = async ( ) => { } ; associationState = SCTP_STATE . CLOSED ; started = false ; state : SCTPConnectionState = "new" ; isServer = true ; private hmacKey = randomBytes ( 16 ) ; private localPartialReliability = true ; private localPort = this . port ; private localVerificationTag = random32 ( ) ; remoteExtensions : number [ ] = [ ] ; remotePartialReliability = true ; private remotePort ? : number ; private remoteVerificationTag = 0 ; private advertisedRwnd = 1024 * 1024 ; private inboundStreams : { [ key ] : InboundStream ; } = { } ; _inboundStreamsCount = 0 ; _inboundStreamsMax = MAX_STREAMS ; private lastReceivedTsn ? : number ; private sackDuplicates : number [ ] = [ ] ; private sackMisOrdered = new Set < number > ( ) ; private sackNeeded = false ; private cwnd = 3 * USERDATA_MAX_LENGTH ; private fastRecoveryExit ? : number ; private fastRecoveryTransmit = false ; private forwardTsnChunk ? : ForwardTsnChunk ; private flightSize = 0 ; outboundQueue : DataChunk [ ] = [ ] ; private outboundStreamSeq : { [ streamId ] : number ; } = { } ; _outboundStreamsCount = MAX_STREAMS ; /**local transmission sequence number */ private localTsn = Number ( random32 ( ) ) ; private lastSackedTsn = tsnMinusOne ( this . localTsn ) ; private advancedPeerAckTsn = tsnMinusOne ( this . localTsn ) ; private partialBytesAcked = 0 ; private sentQueue : DataChunk [ ] = [ ] ; /**初期TSNと同じ値に初期化される単調に増加する数です. これは、新しいre-configuration requestパラメーターを送信するたびに1ずつ増加します */ reconfigRequestSeq = this . localTsn ; /**このフィールドは、incoming要求のre-configuration requestシーケンス番号を保持します. 他の場合では、次に予想されるre-configuration requestシーケンス番号から1を引いた値が保持されます */ reconfigResponseSeq = 0 ; reconfigRequest ? : OutgoingSSNResetRequestParam ; reconfigQueue : number [ ] = [ ] ; private srtt ? : number ; private rttvar ? : number ; private rto = SCTP_RTO_INITIAL ; /**t1 is wait for initAck or cookieAck */ private timer1Handle ? : any ; private timer1Chunk ? : Chunk ; private timer1Failures = 0 ; /**t2 is wait for shutdown */ private timer2Handle ? : any ; private timer2Chunk ? : Chunk ; private timer2Failures = 0 ; /**t3 is wait for data sack */ private timer3Handle ? : any ; /**Re-configuration Timer */ private timerReconfigHandle ? : any ; private timerReconfigFailures = 0 ; private ssthresh ? : number ; constructor ( public transport , public port = 5000 ) { this . transport . onData = ( buf ) => { this . handleData ( buf ) ; } ; } get maxChannels ( ) { if ( this . _inboundStreamsCount > 0 ) return Math . min ( this . _inboundStreamsCount , this . _outboundStreamsCount ) ; } static client ( transport , port = 5000 ) { const sctp = new SCTP ( transport , port ) ; sctp . isServer = false ; return sctp ; } static server ( transport , port = 5000 ) { const sctp = new SCTP ( transport , port ) ; sctp . isServer = true ; return sctp ; } private async handleData ( data ) { let expectedTag ; const [ , , verificationTag , chunks ] = parsePacket ( data ) ; const initChunk = chunks . filter ( ( v ) => v . type === InitChunk . type ) . length ; if ( initChunk > 0 ) { if ( chunks . length != 1 ) { throw new Error ( ) ; } expectedTag = 0 ; } else { expectedTag = this . localVerificationTag ; } if ( verificationTag !== expectedTag ) { return ; } for ( const chunk of chunks ) { await this . receiveChunk ( chunk ) ; } if ( this . sackNeeded ) { await this . sendSack ( ) ; } } private async sendSack ( ) { const gaps = [ ] ; let gapNext ; [ ... this . sackMisOrdered ] . sort ( ) . forEach ( ( tsn ) => { const pos = ( tsn - this . lastReceivedTsn ! ) % SCTP_TSN_MODULO ; if ( tsn === gapNext ) { gaps [ gaps . length - 1 ] [ 1 ] = pos ; } else { gaps . push ( [ pos , pos ] ) ; } gapNext = tsnPlusOne ( tsn ) ; } ) ; const sack = new SackChunk ( 0 , undefined ) ; sack . cumulativeTsn = this . lastReceivedTsn ! ; sack . advertisedRwnd = Math . max ( 0 , this . advertisedRwnd ) ; sack . duplicates = [ ... this . sackDuplicates ] ; sack . gaps = gaps ; await this . sendChunk ( sack ) . catch ( ( err ) => { log ( "send sack failed" , err . message ) ; } ) ; this . sackDuplicates = [ ] ; this . sackNeeded = false ; } private async receiveChunk ( chunk ) { switch ( chunk . type ) { case DataChunk . type : { this . receiveDataChunk ( chunk as DataChunk ) ; } break ; case InitChunk . type : { if ( ! this . isServer ) return ; const init = chunk as InitChunk ; log ( "receive init" , init ) ; this . lastReceivedTsn = tsnMinusOne ( init . initialTsn ) ; this . reconfigResponseSeq = tsnMinusOne ( init . initialTsn ) ; this . remoteVerificationTag = init . initiateTag ; this . ssthresh = init . advertisedRwnd ; this . getExtensions ( init . params ) ; this . _inboundStreamsCount = Math . min ( init . outboundStreams , this . _inboundStreamsMax ) ; this . _outboundStreamsCount = Math . min ( this . _outboundStreamsCount , init . inboundStreams ) ; const ack = new InitAckChunk ( ) ; ack . initiateTag = this . localVerificationTag ; ack . advertisedRwnd = this . advertisedRwnd ; ack . outboundStreams = this . _outboundStreamsCount ; ack . inboundStreams = this . _inboundStreamsCount ; ack . initialTsn = this . localTsn ; this . setExtensions ( ack . params ) ; const time = Date . now ( ) / 1000 ; let cookie = Buffer . from ( jspack . Pack ( "!L" , [ time ] ) ) ; cookie = Buffer . concat ( [ cookie , createHmac ( "sha1" , this . hmacKey ) . update ( cookie ) . digest ( ) , ] ) ; ack . params . push ( [ SCTP_STATE_COOKIE , cookie ] ) ; log ( "send initAck" , ack ) ; await this . sendChunk ( ack ) . catch ( ( err ) => { log ( "send initAck failed" , err . message ) ; } ) ; } break ; case InitAckChunk . type : { if ( this . associationState != SCTP_STATE . COOKIE_WAIT ) return ; const initAck = chunk as InitAckChunk ; this . timer1Cancel ( ) ; this . lastReceivedTsn = tsnMinusOne ( initAck . initialTsn ) ; this . reconfigResponseSeq = tsnMinusOne ( initAck . initialTsn ) ; this . remoteVerificationTag = initAck . initiateTag ; this . ssthresh = initAck . advertisedRwnd ; this . getExtensions ( initAck . params ) ; this . _inboundStreamsCount = Math . min ( initAck . outboundStreams , this . _inboundStreamsMax ) ; this . _outboundStreamsCount = Math . min ( this . _outboundStreamsCount , initAck . inboundStreams ) ; const echo = new CookieEchoChunk ( ) ; for ( const [ k , v ] of initAck . params ) { if ( k === SCTP_STATE_COOKIE ) { echo . body = v ; break ; } } await this . sendChunk ( echo ) . catch ( ( err ) => { log ( "send echo failed" , err . message ) ; } ) ; this . timer1Start ( echo ) ; this . setState ( SCTP_STATE . COOKIE_ECHOED ) ; } break ; case SackChunk . type : { await this . receiveSackChunk ( chunk as SackChunk ) ; } break ; case HeartbeatChunk . type : { const ack = new HeartbeatAckChunk ( ) ; ack . params = ( chunk as HeartbeatChunk ) . params ; await this . sendChunk ( ack ) . catch ( ( err ) => { log ( "send heartbeat ack failed" , err . message ) ; } ) ; } break ; case AbortChunk . type : { this . setState ( SCTP_STATE . CLOSED ) ; } break ; case ShutdownChunk . type : { this . timer2Cancel ( ) ; this . setState ( SCTP_STATE . SHUTDOWN_RECEIVED ) ; const ack = new ShutdownAckChunk ( ) ; await this . sendChunk ( ack ) . catch ( ( err ) => { log ( "send shutdown ack failed" , err . message ) ; } ) ; this . t2Start ( ack ) ; this . setState ( SCTP_STATE . SHUTDOWN_SENT ) ; } break ; case ErrorChunk . type : { log ( "ErrorChunk" , ( chunk as ErrorChunk ) . descriptions ) ; } break ; case CookieEchoChunk . type : { if ( ! this . isServer ) return ; const data = chunk as CookieEchoChunk ; const cookie = data . body ! ; const digest = createHmac ( "sha1" , this . hmacKey ) . update ( cookie . slice ( 0 , 4 ) ) . digest ( ) ; if ( cookie ?. length != COOKIE_LENGTH || ! cookie . slice ( 4 ) . equals ( digest ) ) { log ( "x State cookie is invalid" ) ; return ; } const now = Date . now ( ) / 1000 ; const stamp = jspack . Unpack ( "!L" , cookie ) [ 0 ] ; if ( stamp < now - COOKIE_LIFETIME || stamp > now ) { const error = new ErrorChunk ( 0 , undefined ) ; error . params . push ( [ ErrorChunk . CODE . StaleCookieError , Buffer . concat ( [ ... Array ( 8 ) ] . map ( ( ) => Buffer . from ( "\x00" ) ) ) , ] ) ; await this . sendChunk ( error ) . catch ( ( err ) => { log ( "send errorChunk failed" , err . message ) ; } ) ; return ; } const ack = new CookieAckChunk ( ) ; await this . sendChunk ( ack ) . catch ( ( err ) => { log ( "send cookieAck failed" , err . message ) ; } ) ; this . setState ( SCTP_STATE . ESTABLISHED ) ; } break ; case CookieAckChunk . type : { if ( this . associationState != SCTP_STATE . COOKIE_ECHOED ) return ; this . timer1Cancel ( ) ; this . setState ( SCTP_STATE . ESTABLISHED ) ; } break ; case ShutdownCompleteChunk . type : { if ( this . associationState != SCTP_STATE . SHUTDOWN_ACK_SENT ) return ; this . timer2Cancel ( ) ; this . setState ( SCTP_STATE . CLOSED ) ; } break ; case ReconfigChunk . type : { if ( this . associationState != SCTP_STATE . ESTABLISHED ) return ; const reconfig = chunk as ReConfigChunk ; for ( const [ type , body ] of reconfig . params ) { const target = RECONFIG_PARAM_BY_TYPES [ type ] ; if ( target ) { await this . receiveReconfigParam ( target . parse ( body ) ) ; } } } break ; case ForwardTsnChunk . type : { this . receiveForwardTsnChunk ( chunk as ForwardTsnChunk ) ; } break ; } } private getExtensions ( params ) { for ( const [ k , v ] of params ) { if ( k === SCTP_PRSCTP_SUPPORTED ) { this . remotePartialReliability = true ; } else if ( k === SCTP_SUPPORTED_CHUNK_EXT ) { this . remoteExtensions = [ ... v ] ; } } } private async receiveReconfigParam ( param ) { log ( "receiveReconfigParam" , RECONFIG_PARAM_BY_TYPES [ param . type ] ) ; switch ( param . type ) { case OutgoingSSNResetRequestParam . type : { const p = param as OutgoingSSNResetRequestParam ; const response = new ReconfigResponseParam ( p . requestSequence , reconfigResult . ReconfigResultSuccessPerformed ) ; this . reconfigResponseSeq = p . requestSequence ; await this . sendReconfigParam ( response ) ; await Promise . all ( p . streams . map ( async ( streamId ) => { delete this . inboundStreams [ streamId ] ; if ( this . outboundStreamSeq [ streamId ] ) { this . reconfigQueue . push ( streamId ) ; } } ) ) ; await this . transmitReconfigRequest ( ) ; this . onReconfigStreams . execute ( p . streams ) ; } break ; case ReconfigResponseParam . type : { const reset = param as ReconfigResponseParam ; if ( reset . result !== reconfigResult . ReconfigResultSuccessPerformed ) { log ( "OutgoingSSNResetRequestParam failed" , Object . keys ( reconfigResult ) . find ( ( key ) => reconfigResult [ key as never ] === reset . result ) ) ; } else if ( reset . responseSequence === this . reconfigRequest ?. requestSequence ) { const streamIds = this . reconfigRequest . streams . map ( ( streamId ) => { delete this . outboundStreamSeq [ streamId ] ; return streamId ; } ) ; this . onReconfigStreams . execute ( streamIds ) ; this . reconfigRequest = undefined ; this . timerReconfigCancel ( ) ; if ( this . reconfigQueue . length > 0 ) { await this . transmitReconfigRequest ( ) ; } } } break ; case StreamAddOutgoingParam . type : { const add = param as StreamAddOutgoingParam ; this . _inboundStreamsCount += add . newStreams ; const res = new ReconfigResponseParam ( add . requestSequence , 1 ) ; this . reconfigResponseSeq = add . requestSequence ; await this . sendReconfigParam ( res ) ; } break ; } } private receiveDataChunk ( chunk ) { this . sackNeeded = true ; if ( this . markReceived ( chunk . tsn ) ) return ; const inboundStream = this . getInboundStream ( chunk . streamId ) ; inboundStream . addChunk ( chunk ) ; this . advertisedRwnd -= chunk . userData . length ; for ( const message of inboundStream . popMessages ( ) ) { this . advertisedRwnd += message [ 2 ] . length ; this . receive ( ... message ) ; } } private async receiveSackChunk ( chunk ) { if ( uint32Gt ( this . lastSackedTsn , chunk . cumulativeTsn ) ) return ; const receivedTime = Date . now ( ) / 1000 ; this . lastSackedTsn = chunk . cumulativeTsn ; const cwndFullyUtilized = this . flightSize >= this . cwnd ; let done = 0 , doneBytes = 0 ; while ( this . sentQueue . length > 0 && uint32Gte ( this . lastSackedTsn , this . sentQueue [ 0 ] . tsn ) ) { const sChunk = this . sentQueue . shift ( ) ! ; done ++ ; if ( ! sChunk ?. acked ) { doneBytes += sChunk . bookSize ; this . flightSizeDecrease ( sChunk ) ; } if ( done === 1 && sChunk . sentCount === 1 ) { this . updateRto ( receivedTime - sChunk . sentTime ! ) ; } } let loss = false ; if ( chunk . gaps . length > 0 ) { const seen = new Set ( ) ; let highestSeenTsn ; chunk . gaps . forEach ( ( gap ) => range ( gap [ 0 ] , gap [ 1 ] + 1 ) . forEach ( ( pos ) => { highestSeenTsn = ( chunk . cumulativeTsn + pos ) % SCTP_TSN_MODULO ; seen . add ( highestSeenTsn ) ; } ) ) ; let highestNewlyAcked = chunk . cumulativeTsn ; for ( const sChunk of this . sentQueue ) { if ( uint32Gt ( sChunk . tsn , highestSeenTsn ! ) ) { break ; } if ( seen . has ( sChunk . tsn ) && ! sChunk . acked ) { doneBytes += sChunk . bookSize ; sChunk . acked = true ; this . flightSizeDecrease ( sChunk ) ; highestNewlyAcked = sChunk . tsn ; } } for ( const sChunk of this . sentQueue ) { if ( uint32Gt ( sChunk . tsn , highestNewlyAcked ) ) { break ; } if ( ! seen . has ( sChunk . tsn ) ) { sChunk . misses ++ ; if ( sChunk . misses === 3 ) { sChunk . misses = 0 ; if ( ! this . maybeAbandon ( sChunk ) ) { sChunk . retransmit = true ; } sChunk . acked = false ; this . flightSizeDecrease ( sChunk ) ; loss = true ; } } } } if ( this . fastRecoveryExit === undefined ) { if ( done && cwndFullyUtilized ) { if ( this . cwnd <= this . ssthresh ! ) { this . cwnd += Math . min ( doneBytes , USERDATA_MAX_LENGTH ) ; } else { this . partialBytesAcked += doneBytes ; if ( this . partialBytesAcked >= this . cwnd ) { this . partialBytesAcked -= this . cwnd ; this . cwnd += USERDATA_MAX_LENGTH ; } } } if ( loss ) { this . ssthresh = Math . max ( Math . floor ( this . cwnd / 2 ) , 4 * USERDATA_MAX_LENGTH ) ; this . cwnd = this . ssthresh ; this . partialBytesAcked = 0 ; this . fastRecoveryExit = this . sentQueue [ this . sentQueue . length - 1 ] . tsn ; this . fastRecoveryTransmit = true ; } } else if ( uint32Gte ( chunk . cumulativeTsn , this . fastRecoveryExit ) ) { this . fastRecoveryExit = undefined ; } if ( this . sentQueue . length === 0 ) { this . timer3Cancel ( ) ; } else if ( done > 0 ) { this . timer3Restart ( ) ; } this . updateAdvancedPeerAckPoint ( ) ; await this . onSackReceived ( ) ; await this . transmit ( ) ; } receiveForwardTsnChunk ( chunk ) { this . sackNeeded = true ; if ( uint32Gte ( this . lastReceivedTsn ! , chunk . cumulativeTsn ) ) { return ; } const isObsolete = ( x ) => uint32Gt ( x , this . lastReceivedTsn ! ) ; this . lastReceivedTsn = chunk . cumulativeTsn ; this . sackMisOrdered = new Set ( [ ... this . sackMisOrdered ] . filter ( isObsolete ) ) ; for ( const tsn of [ ... this . sackMisOrdered ] . sort ( ) ) { if ( tsn === tsnPlusOne ( this . lastReceivedTsn ) ) { this . lastReceivedTsn = tsn ; } else { break ; } } this . sackDuplicates = this . sackDuplicates . filter ( isObsolete ) ; this . sackMisOrdered = new Set ( [ ... this . sackMisOrdered ] . filter ( isObsolete ) ) ; for ( const [ streamId , streamSeqNum ] of chunk . streams ) { const inboundStream = this . getInboundStream ( streamId ) ; inboundStream . streamSequenceNumber = uint16Add ( streamSeqNum , 1 ) ; for ( const message of inboundStream . popMessages ( ) ) { this . advertisedRwnd += message [ 2 ] . length ; this . receive ( ... message ) ; } } Object . values ( this . inboundStreams ) . forEach ( ( inboundStream ) => { this . advertisedRwnd += inboundStream . pruneChunks ( this . lastReceivedTsn ! ) ; } ) ; } private updateRto ( R ) { if ( ! this . srtt ) { this . rttvar = R / 2 ; this . srtt = R ; } else { this . rttvar = ( 1 - SCTP_RTO_BETA ) * this . rttvar ! + SCTP_RTO_BETA * Math . abs ( this . srtt - R ) ; this . srtt = ( 1 - SCTP_RTO_ALPHA ) * this . srtt + SCTP_RTO_ALPHA * R ; } this . rto = Math . max ( SCTP_RTO_MIN , Math . min ( this . srtt + 4 * this . rttvar , SCTP_RTO_MAX ) ) ; } private receive ( streamId , ppId , data ) { this . onReceive . execute ( streamId , ppId , data ) ; } private getInboundStream ( streamId ) { if ( ! this . inboundStreams [ streamId ] ) { this . inboundStreams [ streamId ] = new InboundStream ( ) ; } return this . inboundStreams [ streamId ] ; } private markReceived ( tsn ) { if ( uint32Gte ( this . lastReceivedTsn ! , tsn ) || this . sackMisOrdered . has ( tsn ) ) { this . sackDuplicates . push ( tsn ) ; return true ; } this . sackMisOrdered . add ( tsn ) ; for ( const tsn of [ ... this . sackMisOrdered ] . sort ( ) ) { if ( tsn === tsnPlusOne ( this . lastReceivedTsn ! ) ) { this . lastReceivedTsn = tsn ; } else { break ; } } const isObsolete = ( x ) => uint32Gt ( x , this . lastReceivedTsn ! ) ; this . sackDuplicates = this . sackDuplicates . filter ( isObsolete ) ; this . sackMisOrdered = new Set ( [ ... this . sackMisOrdered ] . filter ( isObsolete ) ) ; return false ; } send = async ( streamId , ppId , userData , { expiry , maxRetransmits , ordered , } = { expiry : undefined , maxRetransmits : undefined , ordered : true } ) => { const streamSeqNum = ordered ? this . outboundStreamSeq [ streamId ] || 0 : 0 ; const fragments = Math . ceil ( userData . length / USERDATA_MAX_LENGTH ) ; let pos = 0 ; const chunks = [ ] ; for ( const fragment of range ( 0 , fragments ) ) { const chunk = new DataChunk ( 0 , undefined ) ; chunk . flags = 0 ; if ( ! ordered ) { chunk . flags = SCTP_DATA_UNORDERED ; } if ( fragment === 0 ) { chunk . flags |= SCTP_DATA_FIRST_FRAG ; } if ( fragment === fragments - 1 ) { chunk . flags |= SCTP_DATA_LAST_FRAG ; } chunk . tsn = this . localTsn ; chunk . streamId = streamId ; chunk . streamSeqNum = streamSeqNum ; chunk . protocol = ppId ; chunk . userData = userData . slice ( pos , pos + USERDATA_MAX_LENGTH ) ; chunk . bookSize = chunk . userData . length ; chunk . expiry = expiry ; chunk . maxRetransmits = maxRetransmits ; pos += USERDATA_MAX_LENGTH ; this . localTsn = tsnPlusOne ( this . localTsn ) ; chunks . push ( chunk ) ; } chunks . forEach ( ( chunk ) => { this . outboundQueue . push ( chunk ) ; } ) ; if ( ordered ) { this . outboundStreamSeq [ streamId ] = uint16Add ( streamSeqNum , 1 ) ; } if ( ! this . timer3Handle ) { await this . transmit ( ) ; } else { await new Promise ( ( r ) => setImmediate ( r ) ) ; } } ; private async transmit ( ) { if ( this . forwardTsnChunk ) { await this . sendChunk ( this . forwardTsnChunk ) . catch ( ( err ) => { log ( "send forwardTsn failed" , err . message ) ; } ) ; this . forwardTsnChunk = undefined ; if ( ! this . timer3Handle ) { this . timer3Start ( ) ; } } const burstSize = this . fastRecoveryExit != undefined ? 2 * USERDATA_MAX_LENGTH : 4 * USERDATA_MAX_LENGTH ; const cwnd = Math . min ( this . flightSize + burstSize , this . cwnd ) ; let retransmitEarliest = true ; for ( const dataChunk of this . sentQueue ) { if ( dataChunk . retransmit ) { if ( this . fastRecoveryTransmit ) { this . fastRecoveryTransmit = false ; } else if ( this . flightSize >= cwnd ) { return ; } this . flightSizeIncrease ( dataChunk ) ; dataChunk . misses = 0 ; dataChunk . retransmit = false ; dataChunk . sentCount ++ ; await this . sendChunk ( dataChunk ) . catch ( ( err ) => { log ( "send data failed" , err . message ) ; } ) ; if ( retransmitEarliest ) { this . timer3Restart ( ) ; } } retransmitEarliest = false ; } while ( this . outboundQueue . length > 0 ) { const chunk = this . outboundQueue . shift ( ) ; if ( ! chunk ) return ; this . sentQueue . push ( chunk ) ; this . flightSizeIncrease ( chunk ) ; chunk . sentCount ++ ; chunk . sentTime = Date . now ( ) / 1000 ; await this . sendChunk ( chunk ) . catch ( ( err ) => { log ( "send data outboundQueue failed" , err . message ) ; } ) ; if ( ! this . timer3Handle ) { this . timer3Start ( ) ; } } } async transmitReconfigRequest ( ) { if ( this . reconfigQueue . length > 0 && this . associationState === SCTP_STATE . ESTABLISHED && ! this . reconfigRequest ) { const streams = this . reconfigQueue . slice ( 0 , RECONFIG_MAX_STREAMS ) ; this . reconfigQueue = this . reconfigQueue . slice ( RECONFIG_MAX_STREAMS ) ; const param = new OutgoingSSNResetRequestParam ( this . reconfigRequestSeq , this . reconfigResponseSeq , tsnMinusOne ( this . localTsn ) , streams ) ; this . reconfigRequestSeq = tsnPlusOne ( this . reconfigRequestSeq ) ; this . reconfigRequest = param ; await this . sendReconfigParam ( param ) ; this . timerReconfigHandleStart ( ) ; } } async sendReconfigParam ( param ) { log ( "sendReconfigParam" , param ) ; const chunk = new ReconfigChunk ( ) ; chunk . params . push ( [ param . type , param . bytes ] ) ; await this . sendChunk ( chunk ) . catch ( ( err ) => { log ( "send reconfig failed" , err . message ) ; } ) ; } private async sendResetRequest ( streamId ) { log ( "sendResetRequest" , streamId ) ; const chunk = new DataChunk ( 0 , undefined ) ; chunk . streamId = streamId ; this . outboundQueue . push ( chunk ) ; if ( ! this . timer3Handle ) { await this . transmit ( ) ; } } private flightSizeIncrease ( chunk ) { this . flightSize += chunk . bookSize ; } private flightSizeDecrease ( chunk ) { this . flightSize = Math . max ( 0 , this . flightSize - chunk . bookSize ) ; } /**t1 is wait for initAck or cookieAck */ private timer1Start ( chunk ) { if ( this . timer1Handle ) throw new Error ( ) ; this . timer1Chunk = chunk ; this . timer1Failures = 0 ; this . timer1Handle = setTimeout ( this . timer1Expired , this . rto * 1000 ) ; } private timer1Expired = ( ) => { this . timer1Failures ++ ; this . timer1Handle = undefined ; if ( this . timer1Failures > SCTP_MAX_INIT_RETRANS ) { this . setState ( SCTP_STATE . CLOSED ) ; } else { setImmediate ( ( ) => { this . sendChunk ( this . timer1Chunk ! ) . catch ( ( err ) => { log ( "send timer1 chunk failed" , err . message ) ; } ) ; } ) ; this . timer1Handle = setTimeout ( this . timer1Expired , this . rto * 1000 ) ; } } ; private timer1Cancel ( ) { if ( this . timer1Handle ) { clearTimeout ( this . timer1Handle ) ; this . timer1Handle = undefined ; this . timer1Chunk = undefined ; } } /**t2 is wait for shutdown */ private t2Start ( chunk ) { if ( this . timer2Handle ) throw new Error ( ) ; this . timer2Chunk = chunk ; this . timer2Failures = 0 ; this . timer2Handle = setTimeout ( this . timer2Expired , this . rto * 1000 ) ; } private timer2Expired = ( ) => { this . timer2Failures ++ ; this . timer2Handle = undefined ; if ( this . timer2Failures > SCTP_MAX_ASSOCIATION_RETRANS ) { this . setState ( SCTP_STATE . CLOSED ) ; } else { setImmediate ( ( ) => { this . sendChunk ( this . timer2Chunk ! ) . catch ( ( err ) => { log ( "send timer2Chunk failed" , err . message ) ; } ) ; } ) ; this . timer2Handle = setTimeout ( this . timer2Expired , this . rto * 1000 ) ; } } ; private timer2Cancel ( ) { if ( this . timer2Handle ) { clearTimeout ( this . timer2Handle ) ; this . timer2Handle = undefined ; this . timer2Chunk = undefined ; } } /**t3 is wait for data sack */ private timer3Start ( ) { if ( this . timer3Handle ) throw new Error ( ) ; this . timer3Handle = setTimeout ( this . timer3Expired , this . rto * 1000 ) ; } private timer3Restart ( ) { this . timer3Cancel ( ) ; this . timer3Handle = setTimeout ( this . timer3Expired , this . rto ) ; } private timer3Expired = ( ) => { this . timer3Handle = undefined ; this . sentQueue . forEach ( ( chunk ) => { if ( ! this . maybeAbandon ( chunk ) ) { chunk . retransmit = true ; } } ) ; this . updateAdvancedPeerAckPoint ( ) ; this . fastRecoveryExit = undefined ; this . flightSize = 0 ; this . partialBytesAcked = 0 ; this . ssthresh = Math . max ( Math . floor ( this . cwnd / 2 ) , 4 * USERDATA_MAX_LENGTH ) ; this . cwnd = USERDATA_MAX_LENGTH ; this . transmit ( ) ; } ; private timer3Cancel ( ) { if ( this . timer3Handle ) { clearTimeout ( this . timer3Handle ) ; this . timer3Handle = undefined ; } } /**Re-configuration Timer */ private timerReconfigHandleStart ( ) { if ( this . timerReconfigHandle ) return ; log ( "timerReconfigHandleStart" , { rto : this . rto } ) ; this . timerReconfigFailures = 0 ; this . timerReconfigHandle = setTimeout ( this . timerReconfigHandleExpired , this . rto * 1000 ) ; } private timerReconfigHandleExpired = async ( ) => { this . timerReconfigFailures ++ ; this . rto = Math . ceil ( this . rto * 1.5 ) ; if ( this . timerReconfigFailures > SCTP_MAX_ASSOCIATION_RETRANS ) { log ( "timerReconfigFailures" , this . timerReconfigFailures ) ; this . setState ( SCTP_STATE . CLOSED ) ; this . timerReconfigHandle = undefined ; } else if ( this . reconfigRequest ) { log ( "timerReconfigHandleExpired" , this . timerReconfigFailures , this . rto ) ; await this . sendReconfigParam ( this . reconfigRequest ) ; this . timerReconfigHandle = setTimeout ( this . timerReconfigHandleExpired , this . rto * 1000 ) ; } } ; private timerReconfigCancel ( ) { if ( this . timerReconfigHandle ) { log ( "timerReconfigCancel" ) ; clearTimeout ( this . timerReconfigHandle ) ; this . timerReconfigHandle = undefined ; } } private updateAdvancedPeerAckPoint ( ) { if ( uint32Gt ( this . lastSackedTsn , this . advancedPeerAckTsn ) ) { this . advancedPeerAckTsn = this . lastSackedTsn ; } let done = 0 ; const streams = { } ; while ( this . sentQueue . length > 0 && this . sentQueue [ 0 ] . abandoned ) { const chunk = this . sentQueue . shift ( ) ! ; this . advancedPeerAckTsn = chunk . tsn ; done ++ ; if ( ! ( chunk . flags & SCTP_DATA_UNORDERED ) ) { streams [ chunk . streamId ] = chunk . streamSeqNum ; } } if ( done ) { this . forwardTsnChunk = new ForwardTsnChunk ( 0 , undefined ) ; this . forwardTsnChunk . cumulativeTsn = this . advancedPeerAckTsn ; this . forwardTsnChunk . streams = Object . entries ( streams ) . map ( ( [ k , v ] ) => [ Number ( k ) , v , ] ) ; } } private maybeAbandon ( chunk ) { if ( chunk . abandoned ) return true ; const abandon = ( ! ! chunk . maxRetransmits && chunk . maxRetransmits < chunk . sentCount ! ) || ( ! ! chunk . expiry && chunk . expiry < Date . now ( ) / 1000 ) ; if ( ! abandon ) return false ; const chunkPos = this . sentQueue . findIndex ( ( v ) => v . type === chunk . type ) ; for ( const pos of range ( chunkPos , - 1 , - 1 ) ) { const oChunk = this . sentQueue [ pos ] ; oChunk . abandoned = true ; oChunk . retransmit = false ; if ( oChunk . flags & SCTP_DATA_LAST_FRAG ) { break ; } } for ( const pos of range ( chunkPos , this . sentQueue . length ) ) { const oChunk = this . sentQueue [ pos ] ; oChunk . abandoned = true ; oChunk . retransmit = false ; if ( oChunk . flags & SCTP_DATA_LAST_FRAG ) { break ; } } return true ; } static getCapabilities ( ) { return new RTCSctpCapabilities ( 65536 ) ; } setRemotePort ( port ) { this . remotePort = port ; } async start ( remotePort ? ) { if ( ! this . started ) { this . started = true ; this . setConnectionState ( "connecting" ) ; if ( remotePort ) { this . setRemotePort ( remotePort ) ; } if ( ! this . isServer ) { await this . init ( ) ; } } } private async init ( ) { const init = new InitChunk ( ) ; init . initiateTag = this . localVerificationTag ; init . advertisedRwnd = this . advertisedRwnd ; init . outboundStreams = this . _outboundStreamsCount ; init . inboundStreams = this . _inboundStreamsMax ; init . initialTsn = this . localTsn ; this . setExtensions ( init . params ) ; log ( "send init" , init ) ; try { await this . sendChunk ( init ) ; this . timer1Start ( init ) ; this . setState ( SCTP_STATE . COOKIE_WAIT ) ; } catch ( error ) { log ( "send init failed" , error . message ) ; } } private setExtensions ( params ) { const extensions = [ ] ; if ( this . localPartialReliability ) { params . push ( [ SCTP_PRSCTP_SUPPORTED , Buffer . from ( "" ) ] ) ; extensions . push ( ForwardTsnChunk . type ) ; } extensions . push ( ReConfigChunk . type ) ; params . push ( [ SCTP_SUPPORTED_CHUNK_EXT , Buffer . from ( extensions ) ] ) ; } async sendChunk ( chunk ) { if ( this . state === "closed" ) return ; if ( this . remotePort === undefined ) { throw new Error ( "invalid remote port" ) ; } const packet = serializePacket ( this . localPort , this . remotePort , this . remoteVerificationTag , chunk ) ; await this . transport . send ( packet ) ; } setState ( state ) { if ( state != this . associationState ) { this . associationState = state ; } if ( state === SCTP_STATE . ESTABLISHED ) { this . setConnectionState ( "connected" ) ; } else if ( state === SCTP_STATE . CLOSED ) { this . timer1Cancel ( ) ; this . timer2Cancel ( ) ; this . timer3Cancel ( ) ; this . setConnectionState ( "closed" ) ; this . removeAllListeners ( ) ; } } setConnectionState ( state ) { this . state = state ; log ( "setConnectionState" , state ) ; this . stateChanged [ state ] . execute ( ) ; } async stop ( ) { if ( this . associationState !== SCTP_STATE . CLOSED ) { await this . abort ( ) ; } this . setState ( SCTP_STATE . CLOSED ) ; clearTimeout ( this . timer1Handle ) ; clearTimeout ( this . timer2Handle ) ; clearTimeout ( this . timer3Handle ) ; } async abort ( ) { const abort = new AbortChunk ( ) ; await this . sendChunk ( abort ) . catch ( ( err ) => { log ( "send abort failed" , err . message ) ; } ) ; } private removeAllListeners ( ) { Object . values ( this . stateChanged ) . forEach ( ( v ) => v . allUnsubscribe ( ) ) ; } } export class InboundStream { reassembly : DataChunk [ ] = [ ] ; streamSequenceNumber = 0 ; constructor ( ) { } addChunk ( chunk ) { if ( this . reassembly . length === 0 || uint32Gt ( chunk . tsn , this . reassembly [ this . reassembly . length - 1 ] . tsn ) ) { this . reassembly . push ( chunk ) ; return ; } for ( const [ i , v ] of enumerate ( this . reassembly ) ) { if ( v . tsn === chunk . tsn ) throw new Error ( "duplicate chunk in reassembly" ) ; if ( uint32Gt ( v . tsn , chunk . tsn ) ) { this . reassembly . splice ( i , 0 , chunk ) ; break ; } } } * popMessages ( ) { let pos = 0 ; let startPos ; let expectedTsn ; let ordered ; while ( pos < this . reassembly . length ) { const chunk = this . reassembly [ pos ] ; if ( startPos === undefined ) { ordered = ! ( chunk . flags & SCTP_DATA_UNORDERED ) ; if ( ! ( chunk . flags & SCTP_DATA_FIRST_FRAG ) ) { if ( ordered ) { break ; } else { pos ++ ; continue ; } } if ( ordered && uint16Gt ( chunk . streamSeqNum , this . streamSequenceNumber ) ) { break ; } expectedTsn = chunk . tsn ; startPos = pos ; } else if ( chunk . tsn !== expectedTsn ! ) { if ( ordered ! ) { break ; } else { startPos = undefined ; pos ++ ; continue ; } } if ( chunk . flags & SCTP_DATA_LAST_FRAG ) { const arr = this . reassembly . slice ( startPos , pos + 1 ) . map ( ( c ) => c . userData ) . reduce ( ( acc , cur ) => { acc . push ( cur ) ; acc . push ( Buffer . from ( "" ) ) ; return acc ; } , [ ] as Buffer [ ] ) ; arr . pop ( ) ; const userData = Buffer . concat ( arr ) ; this . reassembly = [ ... this . reassembly . slice ( 0 , startPos ) , ... this . reassembly . slice ( pos + 1 ) , ] ; if ( ordered && chunk . streamSeqNum === this . streamSequenceNumber ) { this . streamSequenceNumber = uint16Add ( this . streamSequenceNumber , 1 ) ; } pos = startPos ; yield [ chunk . streamId , chunk . protocol , userData ] ; } else { pos ++ ; } expectedTsn = tsnPlusOne ( expectedTsn ) ; } } pruneChunks ( tsn ) { let pos = - 1 , size = 0 ; for ( const [ i , chunk ] of this . reassembly . entries ( ) ) { if ( uint32Gte ( tsn , chunk . tsn ) ) { pos = i ; size += chunk . userData . length ; } else { break ; } } this . reassembly = this . reassembly . slice ( pos + 1 ) ; return size ; } } export class RTCSctpCapabilities { constructor ( public maxMessageSize ) { } } function tsnMinusOne ( a ) { return ( a - 1 ) % SCTP_TSN_MODULO ; } function tsnPlusOne ( a ) { return ( a + 1 ) % SCTP_TSN_MODULO ; }