sdrtrunk/src/main/java/io/github/dsheirer/audio/broadcast/StreamManager.java

/*
 *
 *  * ******************************************************************************
 *  * Copyright (C) 2014-2019 Dennis Sheirer
 *  *
 *  * This program is free software: you can redistribute it and/or modify
 *  * it under the terms of the GNU General Public License as published by
 *  * the Free Software Foundation, either version 3 of the License, or
 *  * (at your option) any later version.
 *  *
 *  * This program is distributed in the hope that it will be useful,
 *  * but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  * GNU General Public License for more details.
 *  *
 *  * You should have received a copy of the GNU General Public License
 *  * along with this program.  If not, see <http://www.gnu.org/licenses/>
 *  * *****************************************************************************
 *
 *
 */
package io.github.dsheirer.audio.broadcast;

import io.github.dsheirer.preference.UserPreferences;
import io.github.dsheirer.record.AudioRecorder;
import io.github.dsheirer.sample.Listener;
import io.github.dsheirer.sample.buffer.ReusableAudioPacket;
import io.github.dsheirer.util.ThreadPool;
import io.github.dsheirer.util.TimeStamp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class StreamManager implements Listener<ReusableAudioPacket>
{
    private final static Logger mLog = LoggerFactory.getLogger(StreamManager.class);
    private static final long MAXIMUM_RECORDER_LIFESPAN_MILLIS = 30000; //30 seconds

    private static AtomicInteger sNextRecordingNumber = new AtomicInteger();

    private Listener<AudioRecording> mAudioRecordingListener;
    private BroadcastFormat mBroadcastFormat;
    private UserPreferences mUserPreferences;
    private Map<Integer,AudioRecorder> mStreamRecorders = new HashMap<>();
    private Runnable mRecorderMonitor;
    private ScheduledFuture<?> mRecorderMonitorFuture;
    private AtomicBoolean mRunning = new AtomicBoolean();

    /**
     * Stream manager processes all incoming audio packets and reassembles individual audio streams, converts audio
     * to desired output format and persists each stream to disc.  Each recording is capped at a maximum length to
     * ensure that recordings don't run too long before they are streamed out and to ensure that inactive recordings
     * are closed in a timely fashion.
     *
     * Completed streamable audio recordings are nominated to the output listener (for broadcast) upon completion
     *
     * @param listener to receive completed audio recordings
     * @param broadcastFormat for the stream
     * @param userPreferences to obtain the temporary streaming recording directory
     */
    public StreamManager(Listener<AudioRecording> listener, BroadcastFormat broadcastFormat, UserPreferences userPreferences)
    {
        mAudioRecordingListener = listener;
        mBroadcastFormat = broadcastFormat;
        mUserPreferences = userPreferences;
    }

    /**
     * Starts the stream manager.  Schedules a processor to monitor for inactive temporary stream recorders
     * every 2 seconds.
     */
    public void start()
    {
        if(mRunning.compareAndSet(false, true))
        {
            if(mRecorderMonitor == null)
            {
                mRecorderMonitor = new RecorderMonitor();
            }

            mRecorderMonitorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(mRecorderMonitor,
                0,2, TimeUnit.SECONDS);
        }
    }

    /**
     * Stops the stream manager.  Stops the inactive temporary stream recorder monitoring thread.
     */
    public void stop()
    {
        if(mRunning.compareAndSet(true, false))
        {
            if(mRecorderMonitorFuture != null)
            {
                mRecorderMonitorFuture.cancel(true);
            }

            synchronized(mStreamRecorders)
            {
                List<Integer> streamKeys = new ArrayList<>(mStreamRecorders.keySet());

                for(Integer streamKey : streamKeys)
                {
                    removeRecorder(streamKey);
                }
            }
        }
    }

    @Override
    public void receive(ReusableAudioPacket audioPacket)
    {
        if(mRunning.get() && audioPacket.hasIdentifierCollection())
        {
            synchronized(mStreamRecorders)
            {
                int channelMetadataID = audioPacket.getAudioChannelId();

                ReusableAudioPacket.Type type = audioPacket.getType();

                if(type == ReusableAudioPacket.Type.AUDIO)
                {
                    if(mStreamRecorders.containsKey(channelMetadataID))
                    {
                        AudioRecorder recorder = mStreamRecorders.get(channelMetadataID);

                        if(recorder != null)
                        {
                            audioPacket.incrementUserCount();
                            recorder.receive(audioPacket);
                        }
                    }
                    else
                    {
                        AudioRecorder recorder = BroadcastFactory.getAudioRecorder(getTemporaryRecordingPath(), mBroadcastFormat);
                        recorder.start();
                        audioPacket.incrementUserCount();
                        recorder.receive(audioPacket);
                        mStreamRecorders.put(channelMetadataID, recorder);
                    }
                }
                else if(type == ReusableAudioPacket.Type.END)
                {
                    removeRecorder(channelMetadataID);
                }
                else
                {
                    mLog.info("Unrecognized Audio Packet Type: " + type);
                }
            }
        }

        audioPacket.decrementUserCount();
    }

    /**
     * Removes the recorder associated with the source channel ID.
     *
     * Note: this method invocation is not thread safe and must be invoked by a thread safe mechanism that protects the
     * mStreamRecorders map.
     *
     * @param sourceChannelID identifying the recorder
     */
    private void removeRecorder(Integer sourceChannelID)
    {
        if(mStreamRecorders.containsKey(sourceChannelID))
        {
            AudioRecorder recorder = mStreamRecorders.remove(sourceChannelID);

            recorder.close(new Listener<AudioRecorder>()
            {
                @Override
                public void receive(AudioRecorder audioRecorder)
                {
                    AudioRecording audioRecording =
                        new AudioRecording(audioRecorder.getPath(), audioRecorder.getIdentifierCollection(),
                            audioRecorder.getTimeRecordingStart(), audioRecorder.getRecordingLength());

                    if(mAudioRecordingListener != null)
                    {
                        mAudioRecordingListener.receive(audioRecording);
                    }
                }
            });

        }
    }

    /**
     * Creates a temporary streaming recording file path
     */
    private Path getTemporaryRecordingPath()
    {
        StringBuilder sb = new StringBuilder();
        sb.append(BroadcastModel.TEMPORARY_STREAM_FILE_SUFFIX);

        int recordingNumber = sNextRecordingNumber.incrementAndGet();

        //Check for integer overflow and readjust negative value to 0
        if(recordingNumber < 0)
        {
            sNextRecordingNumber.set(0);
            recordingNumber = sNextRecordingNumber.incrementAndGet();
        }
        sb.append(recordingNumber).append("_");
        sb.append(TimeStamp.getLongTimeStamp("_"));
        sb.append(mBroadcastFormat.getFileExtension());

        Path temporaryRecordingPath = mUserPreferences.getDirectoryPreference().getDirectoryStreaming().resolve(sb.toString());

        return temporaryRecordingPath;
    }

    /**
     * Monitors recorders to ensure they don't exceed the maximum life-span allowed for a recording.
     */
    public class RecorderMonitor implements Runnable
    {
        @Override
        public void run()
        {
            synchronized(mStreamRecorders)
            {
                long now = System.currentTimeMillis();

                mStreamRecorders.entrySet().stream()
                    .filter(entry -> entry.getValue().getTimeRecordingStart() + MAXIMUM_RECORDER_LIFESPAN_MILLIS < now)
                    .forEach(entry ->
                    {
                        mLog.info("cycling recorder - max temporary streaming recording time limit reached [" +
                            entry.getValue().getPath().toString() + "]");

                        removeRecorder(entry.getKey());
                    });
            }
        }
    }
}