Package org.red5.server.stream.consumer

Class FileConsumer

java.lang.Object
org.red5.server.stream.consumer.FileConsumer
All Implemented Interfaces:
IConsumer, IMessageComponent, IPipeConnectionListener, IPushableConsumer, Constants, DisposableBean, InitializingBean
public class FileConsumer extends Object implements Constants, IPushableConsumer, IPipeConnectionListener, InitializingBean, DisposableBean
Consumer that pushes messages to file. Used when recording live streams.
Author:
The Red5 Project, Paul Gregoire (mondain@gmail.com), Vladimir Hmelyoff (vlhm@splitmedialabs.com), Octavian Naicu (naicuoctavian@gmail.com)

  • Field Details

    • log

      private static final org.slf4j.Logger log

    • scheduledExecutorService

      private ScheduledExecutorService scheduledExecutorService
      Executor for all instance writer jobs

    • schedulerThreadSize

      private int schedulerThreadSize
      Queue writer thread count

    • queue

      private PriorityQueue<FileConsumer.QueuedData> queue
      Queue to hold data for delayed writing

    • reentrantLock

      private ReentrantReadWriteLock reentrantLock
      Reentrant lock

    • writeLock

      private volatile Lock writeLock
      Write lock

    • readLock

      private volatile Lock readLock
      Read lock

    • scope

      private IScope scope
      Scope

    • file

      private File file
      File

    • writer

      private ITagWriter writer
      Tag writer

    • mode

      private String mode
      Operation mode

    • startTimestamp

      private int startTimestamp
      Start timestamp

    • lastTimestamp

      private int lastTimestamp
      Last write timestamp

    • videoConfigurationTag

      private ITag videoConfigurationTag
      Video decoder configuration

    • audioConfigurationTag

      private ITag audioConfigurationTag
      Audio decoder configuration

    • queueThreshold

      private int queueThreshold
      Number of queued items needed before writes are initiated

    • percentage

      private int percentage
      Percentage of the queue which is sliced for writing

    • delayWrite

      private boolean delayWrite
      Whether or not to use a queue for delaying file writes. The queue is useful for keeping Tag items in their expected order based on their time stamp.

    • lastWrittenTs

      private volatile int lastWrittenTs
      Tracks the last timestamp written to prevent backwards time stamped data.

    • writerFuture

      private volatile Future<?> writerFuture
      Keeps track of the last spawned write worker.

    • gotVideoKeyFrame

      private AtomicBoolean gotVideoKeyFrame

  • Constructor Details

    • FileConsumer

      public FileConsumer()
      Default ctor

    • FileConsumer

      public FileConsumer(IScope scope, File file)
      Creates file consumer
      Parameters:
      scope - Scope of consumer
      file - File

  • Method Details

    • afterPropertiesSet

      public void afterPropertiesSet() throws Exception
      Specified by:
      afterPropertiesSet in interface InitializingBean
      Throws:
      Exception

    • pushMessage

      public void pushMessage(IPipe pipe, IMessage message) throws IOException
      Push message through pipe
      Specified by:
      pushMessage in interface IPushableConsumer
      Parameters:
      pipe - Pipe
      message - Message to push
      Throws:
      IOException - if message could not be written

    • writeQueuedDataSlice

      private void writeQueuedDataSlice(FileConsumer.QueuedData[] slice)

    • createFixedLengthSlice

      private FileConsumer.QueuedData[] createFixedLengthSlice(int sliceLength)

    • createTimestampLimitedSlice

      private FileConsumer.QueuedData[] createTimestampLimitedSlice(int timestamp)

    • acquireWriteFuture

      private boolean acquireWriteFuture(int sliceLength)
      Get the WriteFuture with a timeout based on the length of the slice to write.
      Parameters:
      sliceLength -
      Returns:
      true if successful and false otherwise

    • onOOBControlMessage

      public void onOOBControlMessage(IMessageComponent source, IPipe pipe, OOBControlMessage oobCtrlMsg)
      Out-of-band control message handler
      Specified by:
      onOOBControlMessage in interface IMessageComponent
      Parameters:
      source - Source of message
      pipe - Pipe that is used to transmit OOB message
      oobCtrlMsg - OOB control message

    • onPipeConnectionEvent

      public void onPipeConnectionEvent(PipeConnectionEvent event)
      Pipe connection event handler
      Specified by:
      onPipeConnectionEvent in interface IPipeConnectionListener
      Parameters:
      event - Pipe connection event

    • init

      private void init() throws IOException
      Initialization
      Throws:
      IOException - I/O exception

    • uninit

      public void uninit()
      Reset or uninitialize

    • doWrites

      public final void doWrites()
      Write all the queued items to the writer.

    • doWrites

      public final void doWrites(FileConsumer.QueuedData[] slice)
      Write a slice of the queued items to the writer.
      Parameters:
      slice - set of queued data

    • write

      private final void write(int timestamp, IRTMPEvent msg)
      Write incoming data to the file.
      Parameters:
      timestamp - adjusted timestamp
      msg - stream data

    • write

      private final void write(FileConsumer.QueuedData queued)
      Adjust timestamp and write to the file.
      Parameters:
      queued - queued data for write

    • setVideoDecoderConfiguration

      public void setVideoDecoderConfiguration(IRTMPEvent decoderConfig)
      Sets a video decoder configuration; some codecs require this, such as AVC.
      Parameters:
      decoderConfig - video codec configuration

    • setAudioDecoderConfiguration

      public void setAudioDecoderConfiguration(IRTMPEvent decoderConfig)
      Sets a audio decoder configuration; some codecs require this, such as AAC.
      Parameters:
      decoderConfig - audio codec configuration

    • setScope

      public void setScope(IScope scope)
      Sets the scope for this consumer.
      Parameters:
      scope - scope

    • setFile

      public void setFile(File file)
      Sets the file we're writing to.
      Parameters:
      file - file

    • getFile

      public File getFile()
      Returns the file.
      Returns:
      file

    • setQueueThreshold

      public void setQueueThreshold(int queueThreshold)
      Sets the threshold for the queue. When the threshold is met a worker is spawned to empty the sorted queue to the writer.
      Parameters:
      queueThreshold - number of items to queue before spawning worker

    • getQueueThreshold

      public int getQueueThreshold()
      Returns the size of the delayed writing queue.
      Returns:
      queue length

    • isDelayWrite

      public boolean isDelayWrite()
      Whether or not the queue should be utilized.
      Returns:
      true if using the queue, false if sending directly to the writer

    • setDelayWrite

      public void setDelayWrite(boolean delayWrite)
      Sets whether or not to use the queue.
      Parameters:
      delayWrite - true to use the queue, false if not

    • getSchedulerThreadSize

      public int getSchedulerThreadSize()
      Returns:
      the schedulerThreadSize

    • setSchedulerThreadSize

      public void setSchedulerThreadSize(int schedulerThreadSize)
      Parameters:
      schedulerThreadSize - the schedulerThreadSize to set

    • setMode

      public void setMode(String mode)
      Sets the recording mode.
      Parameters:
      mode - either "record" or "append" depending on the type of action to perform

    • destroy

      public void destroy() throws Exception
      Specified by:
      destroy in interface DisposableBean
      Throws:
      Exception