code: private void processQueue() {
          CachedEvent cachedEvent;
          try {
              IRTMPEvent event = null;
              RTMPMessage message = null;
              // get first event in the queue
              cachedEvent = queue.poll();
              if (cachedEvent != null) {
                  // get the data type
                  final byte dataType = cachedEvent.getDataType();
                  // get the data
                  IoBuffer buffer = cachedEvent.getData();
                  // get the current size of the buffer / data
                  int bufferLimit = buffer.limit();
                  if (bufferLimit > 0) {
                      // create new RTMP message and push to the consumer
                      switch (dataType) {
                          case Constants.TYPE_AGGREGATE:
                              event = new Aggregate(buffer);
                              event.setTimestamp(cachedEvent.getTimestamp());
                              message = RTMPMessage.build(event);
                              break;
                          case Constants.TYPE_AUDIO_DATA:
                              event = new AudioData(buffer);
                              event.setTimestamp(cachedEvent.getTimestamp());
                              message = RTMPMessage.build(event);
                              break;
                          case Constants.TYPE_VIDEO_DATA:
                              event = new VideoData(buffer);
                              event.setTimestamp(cachedEvent.getTimestamp());
                              message = RTMPMessage.build(event);
                              break;
                          default:
                              event = new Notify(buffer);
                              event.setTimestamp(cachedEvent.getTimestamp());
                              message = RTMPMessage.build(event);
                              break;
                      }
                      // push it down to the recorder
                      recordingConsumer.pushMessage(null, message);
                  } else if (bufferLimit == 0 && dataType == Constants.TYPE_AUDIO_DATA) {
                      log.debug("Stream data size was 0, sending empty audio message");
                      // allow for 0 byte audio packets
                      event = new AudioData(IoBuffer.allocate(0));
                      event.setTimestamp(cachedEvent.getTimestamp());
                      message = RTMPMessage.build(event);
                      // push it down to the recorder
                      recordingConsumer.pushMessage(null, message);
                  } else {
                      log.debug("Stream data size was 0, recording pipe will not be notified");
                  }
              }
          } catch (Exception e) {
              log.warn("Exception while pushing to consumer", e);
          }
      }
docs: Process the queued items.