import Event from "rx.mini" ; import { bufferReader , bufferWriter } from "../../../common/src" ; import { RtcpSrPacket , RtpPacket } from ".." ; import { RtcpPacket } from "../rtcp/rtcp" ; import { Pipeline } from "./base" ; export class LipSync extends Pipeline { baseNtpTimestamp ? : bigint ; baseRtpTimestamp ? : number ; private rtpPackets : { [ pt ] : RtpPacket [ ] ; } = { } ; constructor ( public clockRate , public mismatch , streams ? ) { super ( streams ) ; } pushRtcpPackets ( packets ) { packets . forEach ( ( sr ) => { if ( sr instanceof RtcpSrPacket ) { this . srReceived ( sr ) ; } } ) ; this . children ?. pushRtcpPackets ?. ( packets ) ; } private srReceived ( sr ) { const { ntpTimestamp , rtpTimestamp } = sr . senderInfo ; this . baseNtpTimestamp = ntpTimestamp ; this . baseRtpTimestamp = rtpTimestamp ; } pushRtpPackets ( packets ) { packets . forEach ( ( p ) => { this . rtpPackets [ p . header . payloadType ] = this . rtpPackets [ p . header . payloadType ] ?? [ ] ; this . rtpPackets [ p . header . payloadType ] . push ( p ) ; } ) ; if ( Object . keys ( this . rtpPackets ) . length === 2 ) { const [ a , b ] = Object . values ( this . rtpPackets ) ; const lastA = this . calcNtpTime ( a . slice ( - 1 ) [ 0 ] . header . timestamp ) ; const lastB = this . calcNtpTime ( b . slice ( - 1 ) [ 0 ] . header . timestamp ) ; if ( lastA == undefined || lastB == undefined ) { this . children ?. pushRtpPackets ?. ( packets ) ; } else { } } } private calcNtpTime ( rtpTimestamp ) { if ( ! this . baseRtpTimestamp || ! this . baseNtpTimestamp ) { return ; } if ( rtpTimestamp - this . baseRtpTimestamp > Max32bit - this . clockRate * 60 ) { this . baseRtpTimestamp += Max32bit ; } else if ( rtpTimestamp + ( Max32bit - this . clockRate * 60 ) - this . baseRtpTimestamp < 0 ) { rtpTimestamp += Max32bit ; } const elapsed = ( rtpTimestamp - this . baseRtpTimestamp ) / this . clockRate ; return ntpTime2Time ( this . baseNtpTimestamp ) + elapsed ; } } export const ntpTime2Time = ( ntp ) => { const [ ntpSec , ntpMsec ] = bufferReader ( bufferWriter ( [ 8 ] , [ ntp ] ) , [ 4 , 4 ] ) ; return Number ( `${ ntpSec }.${ ntpMsec }` ) ; } ; /**4294967295 */ export const Max32bit = Number ( ( 0x01n << 32n ) - 1n ) ; export interface BufferResolve { packets : { packet : RtpPacket ; offset : number ; } [ ] ; /**NTP seconds */ startAtNtpTime : number ; }