private void sendMessage ( RTMPMessage messageIn ) { IRTMPEvent eventIn = messageIn . getBody ( ) ; IRTMPEvent event ; switch ( eventIn . getDataType ( ) ) { case Constants . TYPE_AGGREGATE : event = new Aggregate ( ( ( Aggregate ) eventIn ) . getData ( ) ) ; break ; case Constants . TYPE_AUDIO_DATA : event = new AudioData ( ( ( AudioData ) eventIn ) . getData ( ) ) ; break ; case Constants . TYPE_VIDEO_DATA : event = new VideoData ( ( ( VideoData ) eventIn ) . getData ( ) ) ; break ; default : event = new Notify ( ( ( Notify ) eventIn ) . getData ( ) ) ; break ; } // get the incoming event time int eventTime = eventIn . getTimestamp ( ) ; // get the incoming event source type and set on the outgoing event event . setSourceType ( eventIn . getSourceType ( ) ) ; // instance the outgoing message RTMPMessage messageOut = RTMPMessage . build ( event , eventTime ) ; if ( log . isTraceEnabled ( ) ) { log . trace ( "Source type - in: {} out: {}" , eventIn . getSourceType ( ) , messageOut . getBody ( ) . getSourceType ( ) ) ; long delta = System . currentTimeMillis ( ) - playbackStart ; log . trace ( "sendMessage: streamStartTS {}, length {}, streamOffset {}, timestamp {} last timestamp {} delta {} buffered {}" , new Object [ ] { streamStartTS . get ( ) , currentItem . get ( ) . getLength ( ) , streamOffset , eventTime , lastMessageTs , delta , lastMessageTs - delta } ) ; } if ( playDecision == 1 ) { // 1 == vod/file if ( eventTime > 0 && streamStartTS . compareAndSet ( - 1 , eventTime ) ) { log . debug ( "sendMessage: set streamStartTS" ) ; messageOut . getBody ( ) . setTimestamp ( 0 ) ; } long length = currentItem . get ( ) . getLength ( ) ; if ( length >= 0 ) { int duration = eventTime - streamStartTS . get ( ) ; if ( log . isTraceEnabled ( ) ) { log . trace ( "sendMessage duration={} length={}" , duration , length ) ; } if ( duration - streamOffset >= length ) { // sent enough data to client stop ( ) ; return ; } } } else { // don't reset streamStartTS to 0 for live streams if ( eventTime > 0 && streamStartTS . compareAndSet ( - 1 , eventTime ) ) { log . debug ( "sendMessage: set streamStartTS" ) ; } // relative timestamp adjustment for live streams int startTs = streamStartTS . get ( ) ; if ( startTs > 0 ) { // subtract the offset time of when the stream started playing for the client eventTime -= startTs ; messageOut . getBody ( ) . setTimestamp ( eventTime ) ; if ( log . isTraceEnabled ( ) ) { log . trace ( "sendMessage (updated): streamStartTS={}, length={}, streamOffset={}, timestamp={}" , new Object [ ] { startTs , currentItem . get ( ) . getLength ( ) , streamOffset , eventTime } ) ; } } } doPushMessage ( messageOut ) ; }
Send an RTMP message