import debug from 'debug' ; import { Tube } from '../component' ; import { merge } from '../../utils/config' ; import { Message , MessageType , RtspMessage , RtcpMessage , RtpMessage , SdpMessage , } from '../message' ; import { Transform } from 'stream' ; import { Sdp } from '../../utils/protocols/sdp' ; import { statusCode , connectionEnded , sequence , sessionId , contentBase , range , sessionTimeout , contentLocation , } from '../../utils/protocols/rtsp' ; import { isRtcpSR , Rtcp } from '../../utils/protocols/rtcp' ; import { getTime } from '../../utils/protocols/ntp' ; import { timestamp } from '../../utils/protocols/rtp' ; function isAbsolute ( url ) { return /^[^:]+:\/\// . test ( url ) ; } enum STATE { IDLE = 'idle' , PLAYING = 'playing' , PAUSED = 'paused' } export enum RTSP_METHOD { OPTIONS = 'OPTIONS' , DESCRIBE = 'DESCRIBE' , SETUP = 'SETUP' , PLAY = 'PLAY' , PAUSE = 'PAUSE' , TEARDOWN = 'TEARDOWN' } const MIN_SESSION_TIMEOUT = 5 ; interface Headers { [ key ] : string ; } interface Command { method : RTSP_METHOD ; headers ? : Headers ; uri ? : string ; } interface MethodHeaders { [ key ] : Headers ; } export interface RtspConfig { hostname ? : string ; parameters ? : string [ ] ; uri ? : string ; headers ? : MethodHeaders ; defaultHeaders ? : Headers ; } const defaultConfig = ( hostname = typeof window === 'undefined' ? '' : window . location . hostname , parameters = [ ] ) : RtspConfig => { const uri = parameters . length > 0 ? `rtsp://${ hostname }/axis-media/media.amp?${ parameters . join ( '&' ) }` : `rtsp://${ hostname }/axis-media/media.amp` ; return { uri } ; } ; export class RTSPResponseError extends Error { public code : number ; constructor ( message , code ) { super ( message ) ; this . name = 'RTSPResponseError' ; this . code = code ; } } extends Component export class RtspSession extends Tube { public uri ? : string ; public headers ? : MethodHeaders ; public defaultHeaders ? : Headers ; public t0 ? : { [ key ] : number ; } ; public n0 ? : { [ key ] : number ; } ; public clockrates ? : { [ key ] : number ; } ; public startTime ? : number ; public onRtcp ? : ( rtcp : Rtcp ) => void ; public onSdp ? : ( sdp : Sdp ) => void ; public onError ? : ( err : RTSPResponseError ) => void ; public onPlay ? : ( range ? : string [ ] ) => void ; public retry ? : ( ) => void ; private _outgoingClosed : boolean ; private _sequence ? : number ; private _callStack ? : Command [ ] ; private _callHistory ? : any [ ] ; private _state ? : STATE ; private _waiting ? : boolean ; private _contentBase ? : string | null ; private _contentLocation ? : string | null ; private _sessionId ? : string | null ; private _sessionControlURL : string ; private _renewSessionInterval ? : number | null ; param config param config . hostname param config . parameters param config . uri param config . defaultHeaders param config . headers constructor ( config = { } ) { const { uri , headers , defaultHeaders } = merge ( defaultConfig ( config . hostname , config . parameters ) , config ) ; const incoming = new Transform ( { objectMode : true , transform : ( msg , _ , callback ) => { if ( msg . type === MessageType . RTSP ) { this . _onRtsp ( msg ) ; callback ( ) ; } else if ( msg . type === MessageType . RTCP ) { this . _onRtcp ( msg ) ; this . onRtcp && this . onRtcp ( msg . rtcp ) ; callback ( undefined , msg ) ; } else if ( msg . type === MessageType . RTP ) { this . _onRtp ( msg ) ; callback ( undefined , msg ) ; } else if ( msg . type === MessageType . SDP ) { this . _onSdp ( msg ) ; this . onSdp && this . onSdp ( msg . sdp ) ; callback ( undefined , msg ) ; } else { callback ( undefined , msg ) ; } } , } ) ; incoming . on ( 'end' , ( ) => { this . _outgoingClosed = true ; } ) ; super ( incoming ) ; this . _outgoingClosed = false ; this . _reset ( ) ; this . update ( uri , headers , defaultHeaders ) ; this . _sessionControlURL = this . _controlURL ( ) ; } param uri param headers param defaultHeaders update ( uri , headers = { } , defaultHeaders = { } ) { if ( uri === undefined ) { throw new Error ( 'You must supply an uri when creating a RtspSessionComponent' ) ; } this . uri = uri ; this . defaultHeaders = defaultHeaders ; this . headers = Object . assign ( { [ RTSP_METHOD . OPTIONS ] : { } , [ RTSP_METHOD . PLAY ] : { } , [ RTSP_METHOD . SETUP ] : { Blocksize : '64000' } , [ RTSP_METHOD . DESCRIBE ] : { Accept : 'application/sdp' } , [ RTSP_METHOD . PAUSE ] : { } , } , headers ) ; } /**
     * Restore the initial values to the state they were in before any RTSP
     * connection was made.
     */ _reset ( ) { this . _sequence = 1 ; this . retry = ( ) => console . error ( "No request sent, can't retry" ) ; this . _callStack = [ ] ; this . _callHistory = [ ] ; this . _state = STATE . IDLE ; this . _waiting = false ; this . _contentBase = null ; this . _sessionId = null ; if ( this . _renewSessionInterval !== null ) { clearInterval ( this . _renewSessionInterval ) ; } this . _renewSessionInterval = null ; this . t0 = undefined ; this . n0 = undefined ; this . clockrates = undefined ; } _controlURL ( attribute ? ) { if ( attribute !== undefined && isAbsolute ( attribute ) ) { return attribute ; } const baseURL = this . _contentBase ?? this . _contentLocation ?? this . uri ; if ( baseURL === null || baseURL === undefined ) { throw new Error ( 'relative or missing control attribute but no base URL available' ) ; } if ( attribute === undefined || attribute === '*' ) { return baseURL ; } return new URL ( attribute , baseURL ) . href ; } param msg _onRtsp ( msg ) { this . _waiting = false ; const status = statusCode ( msg . data ) ; const ended = connectionEnded ( msg . data ) ; const seq = sequence ( msg . data ) ; if ( seq === null ) { throw new Error ( 'rtsp: expected sequence number' ) ; } if ( this . _callHistory === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } const method = this . _callHistory [ seq - 1 ] ; debug ( 'msl:rtsp:incoming' ) ( `${ msg . data }` ) ; if ( ! this . _sessionId && ! ended ) { this . _sessionId = sessionId ( msg . data ) ; const _sessionTimeout = sessionTimeout ( msg . data ) ; if ( _sessionTimeout !== null ) { if ( this . _renewSessionInterval !== null ) { clearInterval ( this . _renewSessionInterval ) ; } this . _renewSessionInterval = setInterval ( ( ) => { this . _enqueue ( { method : RTSP_METHOD . OPTIONS } ) ; this . _dequeue ( ) ; } , Math . max ( MIN_SESSION_TIMEOUT , _sessionTimeout - 5 ) * 1000 ) as unknown as number ; } } if ( ! this . _contentBase ) { this . _contentBase = contentBase ( msg . data ) ; } if ( ! this . _contentLocation ) { this . _contentLocation = contentLocation ( msg . data ) ; } if ( status >= 400 ) { this . onError && this . onError ( new RTSPResponseError ( msg . data . toString ( 'ascii' ) , status ) ) ; } if ( method === RTSP_METHOD . PLAY ) { this . onPlay && this . onPlay ( range ( msg . data ) ) ; } if ( ended ) { debug ( 'msl:rtsp:incoming' ) ( `RTSP Session ${ this . _sessionId } ended with statusCode: ${ status }` ) ; this . _sessionId = null ; } this . _dequeue ( ) ; } _onRtcp ( msg ) { if ( this . t0 === undefined || this . n0 === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } if ( isRtcpSR ( msg . rtcp ) ) { const rtpChannel = msg . channel - 1 ; this . t0 [ rtpChannel ] = msg . rtcp . rtpTimestamp ; this . n0 [ rtpChannel ] = getTime ( msg . rtcp . ntpMost , msg . rtcp . ntpLeast ) ; } } _onRtp ( msg ) { if ( this . t0 === undefined || this . n0 === undefined || this . clockrates === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } const rtpChannel = msg . channel ; const t0 = this . t0 [ rtpChannel ] ; const n0 = this . n0 [ rtpChannel ] ; if ( typeof t0 !== 'undefined' && typeof n0 !== 'undefined' ) { const clockrate = this . clockrates [ rtpChannel ] ; const t = timestamp ( msg . data ) ; const dt = ( t - t0 ) | 0 ; msg . ntpTimestamp = ( dt / clockrate ) * 1000 + n0 ; } } param msg _onSdp ( msg ) { this . n0 = { } ; this . t0 = { } ; this . clockrates = { } ; this . _sessionControlURL = this . _controlURL ( msg . sdp . session . control ) ; msg . sdp . media . forEach ( ( media , index ) => { if ( media . rtpmap === undefined ) { return ; } const { clockrate } = media . rtpmap ; const rtp = index * 2 ; const rtcp = rtp + 1 ; const uri = media . control === undefined ? this . _sessionControlURL : this . _controlURL ( media . control ) ; this . _enqueue ( { method : RTSP_METHOD . SETUP , headers : { Transport : 'RTP/AVP/TCP;unicast;interleaved=' + rtp + '-' + rtcp , } , uri , } ) ; if ( this . clockrates === undefined ) { return ; } this . clockrates [ rtp ] = clockrate ; } ) ; if ( this . _state === STATE . PLAYING ) { this . _enqueue ( { method : RTSP_METHOD . PLAY , headers : { Range : `npt=${ this . startTime || 0 }-` , } , uri : this . _sessionControlURL , } ) ; } this . _dequeue ( ) ; } param startTime play ( startTime = 0 ) { if ( this . _state === STATE . IDLE ) { this . startTime = Number ( startTime ) || 0 ; this . _enqueue ( { method : RTSP_METHOD . OPTIONS } ) ; this . _enqueue ( { method : RTSP_METHOD . DESCRIBE } ) ; } else if ( this . _state === STATE . PAUSED ) { if ( this . _sessionId === null || this . _sessionId === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } this . _enqueue ( { method : RTSP_METHOD . PLAY , headers : { Session : this . _sessionId , } , uri : this . _sessionControlURL , } ) ; } this . _state = STATE . PLAYING ; this . _dequeue ( ) ; } return { undefined } pause ( ) { this . _enqueue ( { method : RTSP_METHOD . PAUSE } ) ; this . _state = STATE . PAUSED ; this . _dequeue ( ) ; } return { undefined } stop ( ) { if ( this . _sessionId ) { this . _enqueue ( { method : RTSP_METHOD . TEARDOWN } ) ; } else { this . _callStack = [ ] ; } this . _state = STATE . IDLE ; if ( this . _renewSessionInterval !== null ) { clearInterval ( this . _renewSessionInterval ) ; this . _renewSessionInterval = null ; } this . _dequeue ( ) ; } param cmd send ( cmd ) { const { method , headers , uri } = cmd ; if ( method === undefined ) { throw new Error ( 'missing method when send request' ) ; } this . _waiting = true ; this . retry = this . send . bind ( this , cmd ) ; if ( this . _sequence === undefined || this . headers === undefined || this . _callHistory === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } const message = Object . assign ( { type : MessageType . RTSP , uri : uri || this . _sessionControlURL , data : Buffer . alloc ( 0 ) , } , { method , headers } , { headers : Object . assign ( { CSeq : this . _sequence ++ } , this . defaultHeaders , this . headers [ method ] , headers ) , } ) ; this . _sessionId && ( message . headers . Session = this . _sessionId ) ; this . _callHistory . push ( method ) ; if ( ! this . _outgoingClosed ) { this . outgoing . push ( message ) ; } else { debug ( 'msl:rtsp:outgoing' ) ( `Unable to send ${ method }, connection closed` ) ; } } param cmd _enqueue ( cmd ) { if ( this . _callStack === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } this . _callStack . push ( cmd ) ; } /**
     * If possible, send the next command on the call stack.
     */ _dequeue ( ) { if ( this . _callStack === undefined ) { throw new Error ( 'rtsp: internal error' ) ; } if ( ! this . _waiting && this . _callStack . length > 0 ) { const cmd = this . _callStack . shift ( ) ; if ( cmd !== undefined ) { this . send ( cmd ) ; } } } }