protected boolean dropMessage(int channelId, IRTMPEvent message) {
        // whether or not to allow dropping functionality
        if (!dropEncoded) {
            log.trace("Not dropping due to flag, source type: {} (0=vod,1=live)", message.getSourceType());
            return false;
        }
        // dont want to drop vod
        boolean isLiveStream = message.getSourceType() == Constants.SOURCE_TYPE_LIVE;
        if (!isLiveStream) {
            log.trace("Not dropping due to vod");
            return false;
        }
        RTMPConnection conn = (RTMPConnection) Red5.getConnectionLocal();
        if (message instanceof Ping) {
            final Ping pingMessage = (Ping) message;
            if (pingMessage.getEventType() == Ping.STREAM_PLAYBUFFER_CLEAR) {
                // client buffer cleared, make sure to reset timestamps for this stream
                final int channel = conn.getChannelIdForStreamId(pingMessage.getValue2());
                log.trace("Ping stream id: {} channel id: {}", pingMessage.getValue2(), channel);
                conn.getState().clearLastTimestampMapping(channel, channel + 1, channel + 2);
            }
            // never drop pings
            return false;
        }
        // whether or not the packet will be dropped
        boolean drop = false;
        // we only drop audio or video data
        boolean isDroppable = message instanceof VideoData || message instanceof AudioData;
        if (isDroppable) {
            if (message.getTimestamp() == 0) {
                // never drop initial packages, also this could be the first packet after
                // MP4 seeking and therefore mess with the timestamp mapping
                return false;
            }
            if (log.isDebugEnabled()) {
                String sourceType = (isLiveStream ? "LIVE" : "VOD");
                log.debug("Connection: {} connType={}", conn.getSessionId(), sourceType);
            }
            RTMP rtmp = conn.getState();
            long timestamp = (message.getTimestamp() & 0xFFFFFFFFL);
            LiveTimestampMapping mapping = rtmp.getLastTimestampMapping(channelId);
            long now = System.currentTimeMillis();
            if (mapping == null || timestamp < mapping.getLastStreamTime()) {
                log.trace("Resetting clock time ({}) to stream time ({})", now, timestamp);
                // either first time through, or time stamps were reset
                mapping = rtmp.new LiveTimestampMapping(now, timestamp);
                rtmp.setLastTimestampMapping(channelId, mapping);
            }
            mapping.setLastStreamTime(timestamp);
            // Calculate when this message should have arrived. Take the time when the stream started, add
            // the current message's timestamp and subtract the timestamp of the first message.
            long clockTimeOfMessage = mapping.getClockStartTime() + timestamp - mapping.getStreamStartTime();

            // Determine how late this message is. This will tell us if the incoming path is congested.
            long incomingLatency = clockTimeOfMessage - now;

            if (log.isDebugEnabled()) {
                log.debug("incomingLatency={} clockTimeOfMessage={} getClockStartTime={} timestamp={} getStreamStartTime={} now={}", new Object[] { incomingLatency, clockTimeOfMessage, mapping.getClockStartTime(), timestamp, mapping.getStreamStartTime(), now });
            }
            //TDJ: EXPERIMENTAL dropping for LIVE packets in future (default false)
            if (isLiveStream && dropLiveFuture) {
                incomingLatency = Math.abs(incomingLatency);
                if (log.isDebugEnabled()) {
                    log.debug("incomingLatency={} clockTimeOfMessage={} now={}", new Object[] { incomingLatency, clockTimeOfMessage, now });
                }
            }
            // NOTE: We could decide to drop the message here because it is very late due to incoming traffic congestion.

            // We need to calculate how long will this message reach the client. If the traffic is congested, we decide to drop the message.
            long outgoingLatency = 0L;
            if (conn != null) {
                // Determine the interval when the last ping was sent and when the pong was received. The
                // duration will be the round-trip time (RTT) of the ping-pong message. If it is high then it
                // means that there is congestion in the connection.
                int lastPingPongInterval = conn.getLastPingSentAndLastPongReceivedInterval();
                if (lastPingPongInterval > 0) {
                    // The outgoingLatency is the ping RTT minus the incoming latency.
                    outgoingLatency = lastPingPongInterval - incomingLatency;
                    if (log.isDebugEnabled()) {
                        log.debug("outgoingLatency={} lastPingTime={}", new Object[] { outgoingLatency, lastPingPongInterval });
                    }
                }
            }
            //TODO: how should we differ handling based on live or vod?
            //TODO: if we are VOD do we "pause" the provider when we are consistently late?
            if (log.isTraceEnabled()) {
                log.trace("Packet timestamp: {}; latency: {}; now: {}; message clock time: {}, dropLiveFuture: {}", new Object[] { timestamp, incomingLatency, now, clockTimeOfMessage, dropLiveFuture });
            }
            if (outgoingLatency < baseTolerance) {
                // no traffic congestion in outgoing direction
            } else if (outgoingLatency > highestTolerance) {
                // traffic congestion in outgoing direction
                if (log.isDebugEnabled()) {
                    log.debug("Outgoing direction congested. outgoingLatency={} highestTolerance={}", new Object[] { outgoingLatency, highestTolerance });
                }
                if (isDroppable) {
                    mapping.setKeyFrameNeeded(true);
                }
                drop = true;
            } else {
                if (isDroppable && message instanceof VideoData) {
                    VideoData video = (VideoData) message;
                    if (video.getFrameType() == FrameType.KEYFRAME) {
                        // if its a key frame the inter and disposable checks can be skipped
                        if (log.isDebugEnabled()) {
                            log.debug("Resuming stream with key frame; message: {}", message);
                        }
                        mapping.setKeyFrameNeeded(false);
                    } else if (incomingLatency >= baseTolerance && incomingLatency < midTolerance) {
                        // drop disposable frames
                        if (video.getFrameType() == FrameType.DISPOSABLE_INTERFRAME) {
                            if (log.isDebugEnabled()) {
                                log.debug("Dropping disposible frame; message: {}", message);
                            }
                            drop = true;
                        }
                    } else if (incomingLatency >= midTolerance && incomingLatency <= highestTolerance) {
                        // drop inter-frames and disposable frames
                        if (log.isDebugEnabled()) {
                            log.debug("Dropping disposible or inter frame; message: {}", message);
                        }
                        drop = true;
                    }
                }
            }
        }
        if (log.isDebugEnabled() && drop) {
            log.debug("Message was dropped");
        }
        return drop;
    }
protected boolean dropMessage ( int channelId , IRTMPEvent message ) { if ( ! dropEncoded ) { log . trace ( "Not dropping due to flag, source type: {} (0=vod,1=live)" , message . getSourceType ( ) ) ; return false ; } boolean isLiveStream = message . getSourceType ( ) == Constants . SOURCE_TYPE_LIVE ; if ( ! isLiveStream ) { log . trace ( "Not dropping due to vod" ) ; return false ; } RTMPConnection conn = ( RTMPConnection ) Red5 . getConnectionLocal ( ) ; if ( message instanceof Ping ) { final Ping pingMessage = ( Ping ) message ; if ( pingMessage . getEventType ( ) == Ping . STREAM_PLAYBUFFER_CLEAR ) { final int channel = conn . getChannelIdForStreamId ( pingMessage . getValue2 ( ) ) ; log . trace ( "Ping stream id: {} channel id: {}" , pingMessage . getValue2 ( ) , channel ) ; conn . getState ( ) . clearLastTimestampMapping ( channel , channel + 1 , channel + 2 ) ; } return false ; } boolean drop = false ; boolean isDroppable = message instanceof VideoData || message instanceof AudioData ; if ( isDroppable ) { if ( message . getTimestamp ( ) == 0 ) { return false ; } if ( log . isDebugEnabled ( ) ) { String sourceType = ( isLiveStream ? "LIVE" : "VOD" ) ; log . debug ( "Connection: {} connType={}" , conn . getSessionId ( ) , sourceType ) ; } RTMP rtmp = conn . getState ( ) ; long timestamp = ( message . getTimestamp ( ) & 0xFFFFFFFFL ) ; LiveTimestampMapping mapping = rtmp . getLastTimestampMapping ( channelId ) ; long now = System . currentTimeMillis ( ) ; if ( mapping == null || timestamp < mapping . getLastStreamTime ( ) ) { log . trace ( "Resetting clock time ({}) to stream time ({})" , now , timestamp ) ; mapping = rtmp . new LiveTimestampMapping ( now , timestamp ) ; rtmp . setLastTimestampMapping ( channelId , mapping ) ; } mapping . setLastStreamTime ( timestamp ) ; long clockTimeOfMessage = mapping . getClockStartTime ( ) + timestamp - mapping . getStreamStartTime ( ) ; long incomingLatency = clockTimeOfMessage - now ; if ( log . isDebugEnabled ( ) ) { log . debug ( "incomingLatency={} clockTimeOfMessage={} getClockStartTime={} timestamp={} getStreamStartTime={} now={}" , new Object [ ] { incomingLatency , clockTimeOfMessage , mapping . getClockStartTime ( ) , timestamp , mapping . getStreamStartTime ( ) , now } ) ; } if ( isLiveStream && dropLiveFuture ) { incomingLatency = Math . abs ( incomingLatency ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "incomingLatency={} clockTimeOfMessage={} now={}" , new Object [ ] { incomingLatency , clockTimeOfMessage , now } ) ; } } long outgoingLatency = 0L ; if ( conn != null ) { int lastPingPongInterval = conn . getLastPingSentAndLastPongReceivedInterval ( ) ; if ( lastPingPongInterval > 0 ) { outgoingLatency = lastPingPongInterval - incomingLatency ; if ( log . isDebugEnabled ( ) ) { log . debug ( "outgoingLatency={} lastPingTime={}" , new Object [ ] { outgoingLatency , lastPingPongInterval } ) ; } } } if ( log . isTraceEnabled ( ) ) { log . trace ( "Packet timestamp: {}; latency: {}; now: {}; message clock time: {}, dropLiveFuture: {}" , new Object [ ] { timestamp , incomingLatency , now , clockTimeOfMessage , dropLiveFuture } ) ; } if ( outgoingLatency < baseTolerance ) { } else if ( outgoingLatency > highestTolerance ) { if ( log . isDebugEnabled ( ) ) { log . debug ( "Outgoing direction congested. outgoingLatency={} highestTolerance={}" , new Object [ ] { outgoingLatency , highestTolerance } ) ; } if ( isDroppable ) { mapping . setKeyFrameNeeded ( true ) ; } drop = true ; } else { if ( isDroppable && message instanceof VideoData ) { VideoData video = ( VideoData ) message ; if ( video . getFrameType ( ) == FrameType . KEYFRAME ) { if ( log . isDebugEnabled ( ) ) { log . debug ( "Resuming stream with key frame; message: {}" , message ) ; } mapping . setKeyFrameNeeded ( false ) ; } else if ( incomingLatency >= baseTolerance && incomingLatency < midTolerance ) { if ( video . getFrameType ( ) == FrameType . DISPOSABLE_INTERFRAME ) { if ( log . isDebugEnabled ( ) ) { log . debug ( "Dropping disposible frame; message: {}" , message ) ; } drop = true ; } } else if ( incomingLatency >= midTolerance && incomingLatency <= highestTolerance ) { if ( log . isDebugEnabled ( ) ) { log . debug ( "Dropping disposible or inter frame; message: {}" , message ) ; } drop = true ; } } } } if ( log . isDebugEnabled ( ) && drop ) { log . debug ( "Message was dropped" ) ; } return drop ; }
Determine if this message should be dropped. If the traffic from server to client is congested, then drop LIVE messages to help alleviate congestion.

- determine latency between server and client using ping
- ping timestamp is unsigned int (4 bytes) and is set from value on sender

1st drop disposable frames - lowest mark 2nd drop interframes - middle 3rd drop key frames - high mark

@param channelId
the channel ID
@param message
the message
@return true to drop; false to send