From bc430b255f36a0cb2d9cb15467deeacb760d3b8e Mon Sep 17 00:00:00 2001
From: Denny <denny@denny-desktop>
Date: Tue, 3 Jan 2017 09:27:17 -0500
Subject: [PATCH] #30 - changes internal MP3 temporary recording stream
 management so that the broadcast model manages the recordings and sends the
 completed recordings to the broadcasters to avoid 2 or more streams recording
 MP3 files for the same audio call.

Note: there is a bug where temporary recordings are not being consistently deleted that I'm still working on.
---
 src/audio/broadcast/AudioBroadcaster.java     |  87 ++++--------
 ...udioRecording.java => AudioRecording.java} |  33 ++++-
 src/audio/broadcast/BroadcastModel.java       | 125 +++++++++++++++---
 src/audio/broadcast/StreamManager.java        |  23 ++--
 src/record/AudioRecorder.java                 |   6 +-
 5 files changed, 184 insertions(+), 90 deletions(-)
 rename src/audio/broadcast/{StreamableAudioRecording.java => AudioRecording.java} (70%)

diff --git a/src/audio/broadcast/AudioBroadcaster.java b/src/audio/broadcast/AudioBroadcaster.java
index be544ae1..cda9f47e 100644
--- a/src/audio/broadcast/AudioBroadcaster.java
+++ b/src/audio/broadcast/AudioBroadcaster.java
@@ -36,7 +36,7 @@ import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-public abstract class AudioBroadcaster implements IAudioPacketListener
+public abstract class AudioBroadcaster implements Listener<AudioRecording>
 {
     private final static Logger mLog = LoggerFactory.getLogger( AudioBroadcaster.class );
 
@@ -46,14 +46,13 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
     private ScheduledFuture mScheduledTask;
 
     private RecordingQueueProcessor mRecordingQueueProcessor = new RecordingQueueProcessor();
-    private LinkedTransferQueue<StreamableAudioRecording> mAudioRecordingQueue = new LinkedTransferQueue<>();
+    private LinkedTransferQueue<AudioRecording> mAudioRecordingQueue = new LinkedTransferQueue<>();
 
     private ISilenceGenerator mSilenceGenerator;
 
     private Listener<BroadcastEvent> mBroadcastEventListener;
     private BroadcastState mBroadcastState = BroadcastState.READY;
 
-    private StreamManager mStreamManager;
     private int mStreamedAudioCount = 0;
     private BroadcastConfiguration mBroadcastConfiguration;
     private long mDelay;
@@ -85,11 +84,7 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
         mThreadPoolManager = threadPoolManager;
         mBroadcastConfiguration = broadcastConfiguration;
         mDelay = mBroadcastConfiguration.getDelay();
-
         mSilenceGenerator = BroadcastFactory.getSilenceGenerator(broadcastConfiguration.getBroadcastFormat());
-
-        mStreamManager = new StreamManager(threadPoolManager, this,
-                SystemProperties.getInstance().getApplicationFolder(BroadcastModel.TEMPORARY_STREAM_DIRECTORY));
     }
 
     protected ThreadPoolManager getThreadPoolManager()
@@ -133,8 +128,6 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
     {
         if(mStreaming.compareAndSet(false, true))
         {
-            mStreamManager.start();
-
             if(mScheduledTask == null)
             {
                 if(mThreadPoolManager != null)
@@ -153,8 +146,6 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
     {
         if(mStreaming.compareAndSet(true, false))
         {
-            mStreamManager.stop();
-
             if(mThreadPoolManager != null && mScheduledTask != null)
             {
                 mThreadPoolManager.cancel(mScheduledTask);
@@ -201,7 +192,7 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
      *
      * @param recording to queue for broadcasting
      */
-    public void receive(StreamableAudioRecording recording)
+    public void receive(AudioRecording recording)
     {
         if(connected())
         {
@@ -210,24 +201,6 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
         }
     }
 
-    /**
-     * Cleanup method to remove a temporary recording file from disk.
-     *
-     * @param recording to remove
-     */
-    private void removeRecording(StreamableAudioRecording recording)
-    {
-        try
-        {
-            Files.delete(recording.getPath());
-        }
-        catch(IOException ioe)
-        {
-            mLog.error("Stream [" + getBroadcastConfiguration().getName() + "] - error deleting temporary internet " +
-                "recording file: " + recording.getPath().toString() + " - " + ioe.getMessage());
-        }
-    }
-
     /**
      * Broadcast configuration used by this broadcaster
      */
@@ -236,15 +209,6 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
         return mBroadcastConfiguration;
     }
 
-    /**
-     * IAudioPacketListener interface method to gain access to the internal stream manager for enqueuing audio packets.
-     */
-    @Override
-    public Listener<AudioPacket> getAudioPacketListener()
-    {
-        return mStreamManager;
-    }
-
     /**
      * Registers the listener to receive broadcast events/state changes
      */
@@ -291,11 +255,11 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
 
             if(!connected())
             {
-                StreamableAudioRecording recording = mAudioRecordingQueue.poll();
+                //Remove all pending audio recordings
+                AudioRecording recording = mAudioRecordingQueue.poll();
 
                 while(recording != null)
                 {
-                    removeRecording(recording);
                     recording = mAudioRecordingQueue.poll();
                 }
             }
@@ -426,37 +390,38 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
 
             mInputStream = null;
 
-            StreamableAudioRecording nextRecording = mAudioRecordingQueue.peek();
+            AudioRecording nextRecording = mAudioRecordingQueue.peek();
 
             if(nextRecording != null && nextRecording.getStartTime() + mDelay <= System.currentTimeMillis())
             {
-                StreamableAudioRecording recording = mAudioRecordingQueue.poll();
+                AudioRecording recording = mAudioRecordingQueue.poll();
 
                 try
                 {
-                    byte[] audio = Files.readAllBytes(recording.getPath());
-
-                    if(audio != null && audio.length > 0)
+                    if(Files.exists(recording.getPath()))
                     {
-                        mInputStream = new ByteArrayInputStream(audio);
+                        byte[] audio = Files.readAllBytes(recording.getPath());
 
-                        mFinalSilencePadding = PROCESSOR_RUN_INTERVAL_MS -
-                            (recording.getRecordingLength() % PROCESSOR_RUN_INTERVAL_MS);
-
-                        while(mFinalSilencePadding >= PROCESSOR_RUN_INTERVAL_MS)
+                        if(audio != null && audio.length > 0)
                         {
-                            mFinalSilencePadding -= PROCESSOR_RUN_INTERVAL_MS;
-                        }
+                            mInputStream = new ByteArrayInputStream(audio);
 
-                        if(connected())
-                        {
-                            broadcastMetadata(recording.getAudioMetadata());
-                        }
+                            mFinalSilencePadding = PROCESSOR_RUN_INTERVAL_MS -
+                                (recording.getRecordingLength() % PROCESSOR_RUN_INTERVAL_MS);
 
-                        metadataUpdateRequired = false;
+                            while(mFinalSilencePadding >= PROCESSOR_RUN_INTERVAL_MS)
+                            {
+                                mFinalSilencePadding -= PROCESSOR_RUN_INTERVAL_MS;
+                            }
+
+                            if(connected())
+                            {
+                                broadcastMetadata(recording.getAudioMetadata());
+                            }
+
+                            metadataUpdateRequired = false;
+                        }
                     }
-
-                    removeRecording(recording);
                 }
                 catch(IOException ioe)
                 {
@@ -467,6 +432,8 @@ public abstract class AudioBroadcaster implements IAudioPacketListener
                     metadataUpdateRequired = false;
                 }
 
+                recording.removePendingReplay();
+
                 broadcast(new BroadcastEvent(AudioBroadcaster.this, BroadcastEvent.Event.BROADCASTER_QUEUE_CHANGE));
             }
 
diff --git a/src/audio/broadcast/StreamableAudioRecording.java b/src/audio/broadcast/AudioRecording.java
similarity index 70%
rename from src/audio/broadcast/StreamableAudioRecording.java
rename to src/audio/broadcast/AudioRecording.java
index 507a9a5e..a8538035 100644
--- a/src/audio/broadcast/StreamableAudioRecording.java
+++ b/src/audio/broadcast/AudioRecording.java
@@ -21,13 +21,15 @@ package audio.broadcast;
 import audio.metadata.AudioMetadata;
 
 import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicInteger;
 
-public class StreamableAudioRecording implements Comparable<StreamableAudioRecording>
+public class AudioRecording implements Comparable<AudioRecording>
 {
     private Path mPath;
     private AudioMetadata mAudioMetadata;
     private long mStartTime;
     private long mRecordingLength;
+    private AtomicInteger mPendingReplayCount = new AtomicInteger();
 
     /**
      * Audio recording that is ready to be streamed
@@ -36,7 +38,7 @@ public class StreamableAudioRecording implements Comparable<StreamableAudioRecor
      * @param start time of recording in milliseconds since epoch
      * @param recordingLength in milliseconds
      */
-    public StreamableAudioRecording(Path path, AudioMetadata audioMetadata, long start, long recordingLength)
+    public AudioRecording(Path path, AudioMetadata audioMetadata, long start, long recordingLength)
     {
         mPath = path;
         mAudioMetadata = audioMetadata;
@@ -81,8 +83,33 @@ public class StreamableAudioRecording implements Comparable<StreamableAudioRecor
      * Implements comparable for sorting recordings based on start time in ascending order
      */
     @Override
-    public int compareTo(StreamableAudioRecording otherRecording)
+    public int compareTo(AudioRecording otherRecording)
     {
         return Long.compare(getStartTime(), otherRecording.getStartTime());
     }
+
+    /**
+     * Increments the count of pending replays.
+     */
+    public void addPendingReplay()
+    {
+        mPendingReplayCount.incrementAndGet();
+    }
+
+    /**
+     * Decrements the count of pending replays.
+     */
+    public void removePendingReplay()
+    {
+        mPendingReplayCount.decrementAndGet();
+    }
+
+    /**
+     * Indicates if there are any remaining pending replays.  Once the pending replay count is less than or equal to
+     * zero, the recording can be deleted.
+     */
+    public boolean hasPendingReplays()
+    {
+        return mPendingReplayCount.get() <= 0;
+    }
 }
diff --git a/src/audio/broadcast/BroadcastModel.java b/src/audio/broadcast/BroadcastModel.java
index 9f0416a5..703c350c 100644
--- a/src/audio/broadcast/BroadcastModel.java
+++ b/src/audio/broadcast/BroadcastModel.java
@@ -20,6 +20,7 @@ package audio.broadcast;
 
 import alias.id.broadcast.BroadcastChannel;
 import audio.AudioPacket;
+import audio.metadata.AudioMetadata;
 import controller.ThreadPoolManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -36,6 +37,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -60,12 +62,13 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
     public static final int COLUMN_BROADCASTER_STREAMED_COUNT = 4;
 
     private List<BroadcastConfiguration> mBroadcastConfigurations = new CopyOnWriteArrayList<>();
+    private List<AudioRecording> mRecordingQueue = new CopyOnWriteArrayList<>();
 
     private Map<String,BroadcastConfiguration> mBroadcastConfigurationMap = new HashMap<>();
     private Map<String,AudioBroadcaster> mBroadcasterMap = new HashMap<>();
     private ThreadPoolManager mThreadPoolManager;
     private SettingsManager mSettingsManager;
-
+    private StreamManager mStreamManager;
     private Broadcaster<BroadcastEvent> mBroadcastEventBroadcaster = new Broadcaster<>();
 
     /**
@@ -75,6 +78,13 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
     {
         mThreadPoolManager = threadPoolManager;
         mSettingsManager = settingsManager;
+        mStreamManager = new StreamManager(threadPoolManager, new CompletedRecordingListener(), BroadcastFormat.MP3,
+            SystemProperties.getInstance().getApplicationFolder(TEMPORARY_STREAM_DIRECTORY));
+        mStreamManager.start();
+
+        //Monitor to remove temporary recording files that have been streamed by all audio broadcasters
+        mThreadPoolManager.scheduleFixedRate(ThreadPoolManager.ThreadType.AUDIO_RECORDING,
+            new RecordingDeletionMonitor(), 15l, TimeUnit.SECONDS );
 
         removeOrphanedTemporaryRecordings();
     }
@@ -265,22 +275,10 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
     @Override
     public void receive(AudioPacket audioPacket)
     {
-        if (audioPacket.hasAudioMetadata() && audioPacket.getAudioMetadata().isStreamable())
+        if (audioPacket.hasAudioMetadata() &&
+            audioPacket.getAudioMetadata().isStreamable())
         {
-            for (BroadcastChannel broadcastChannel : audioPacket.getAudioMetadata().getBroadcastChannels())
-            {
-                String channelName = broadcastChannel.getChannelName();
-
-                if (channelName != null)
-                {
-                    AudioBroadcaster audioBroadcaster = getBroadcaster(channelName);
-
-                    if (audioBroadcaster != null)
-                    {
-                        audioBroadcaster.getAudioPacketListener().receive(audioPacket);
-                    }
-                }
-            }
+            mStreamManager.receive(audioPacket);
         }
     }
 
@@ -600,6 +598,25 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
         return null;
     }
 
+    /**
+     * Cleanup method to remove a temporary recording file from disk.
+     *
+     * @param recording to remove
+     */
+    private void removeRecording(AudioRecording recording)
+    {
+        try
+        {
+            Files.delete(recording.getPath());
+        }
+        catch(IOException ioe)
+        {
+            mLog.error("Error deleting temporary internet recording file: " + recording.getPath().toString() + " - " +
+                ioe.getMessage());
+        }
+    }
+
+
     /**
      * Removes any temporary stream recordings left-over from the previous application run.
      *
@@ -659,4 +676,80 @@ public class BroadcastModel extends AbstractTableModel implements Listener<Audio
             createBroadcaster(mBroadcastConfiguration);
         }
     }
+
+    /**
+     * Processes completed audio recordings and distributes them to the audio broadcasters.  Adds the recording to
+     * the audio recording queue to be monitored for deletion.
+     */
+    public class CompletedRecordingListener implements Listener<AudioRecording>
+    {
+        @Override
+        public void receive(AudioRecording audioRecording)
+        {
+            AudioMetadata metadata = audioRecording.getAudioMetadata();
+
+            if(metadata != null && metadata.isStreamable())
+            {
+                for (BroadcastChannel broadcastChannel : metadata.getBroadcastChannels())
+                {
+                    String channelName = broadcastChannel.getChannelName();
+
+                    if (channelName != null)
+                    {
+                        AudioBroadcaster audioBroadcaster = getBroadcaster(channelName);
+
+                        if (audioBroadcaster != null)
+                        {
+                            audioRecording.addPendingReplay();
+                            audioBroadcaster.receive(audioRecording);
+                        }
+                    }
+                }
+            }
+
+            mRecordingQueue.add(audioRecording);
+        }
+    }
+
+    /**
+     * Monitors the recording queue and removes any recordings that have no pending replays by audio broadcasters
+     */
+    public class RecordingDeletionMonitor implements Runnable
+    {
+        @Override
+        public void run()
+        {
+            try
+            {
+                List<AudioRecording> recordingsToDelete = new ArrayList<>();
+
+                Iterator<AudioRecording> it = mRecordingQueue.iterator();
+
+                AudioRecording recording;
+
+                while(it.hasNext())
+                {
+                    recording = it.next();
+
+                    if(!recording.hasPendingReplays())
+                    {
+                        recordingsToDelete.add(recording);
+                    }
+                }
+
+                if(!recordingsToDelete.isEmpty())
+                {
+                    for(AudioRecording recordingToDelete: recordingsToDelete)
+                    {
+                        mRecordingQueue.remove(recordingToDelete);
+                        removeRecording(recordingToDelete);
+                    }
+                }
+            }
+            catch(Exception e)
+            {
+                mLog.error("Error while checking audio recording queue for recordings to delete", e);
+            }
+        }
+    }
 }
diff --git a/src/audio/broadcast/StreamManager.java b/src/audio/broadcast/StreamManager.java
index 3de030e6..b658df7f 100644
--- a/src/audio/broadcast/StreamManager.java
+++ b/src/audio/broadcast/StreamManager.java
@@ -45,7 +45,8 @@ public class StreamManager implements Listener<AudioPacket>
     private static AtomicInteger sNextRecordingNumber = new AtomicInteger();
 
     private ThreadPoolManager mThreadPoolManager;
-    private AudioBroadcaster mAudioBroadcaster;
+    private Listener<AudioRecording> mAudioRecordingListener;
+    private BroadcastFormat mBroadcastFormat;
     private Path mTempDirectory;
     private Map<Integer, AudioRecorder> mStreamRecorders = new HashMap<>();
     private Runnable mRecorderMonitor;
@@ -61,16 +62,18 @@ public class StreamManager implements Listener<AudioPacket>
      * Completed streamable audio recordings are nominated to the output listener (for broadcast) upon completion
      *
      * @param threadPoolManager for scheduling runnables
-     * @param audioBroadcaster to receive completed streamable audio recordings
+     * @param listener to receive completed audio recordings
      * @param tempDirectory where to store temporary audio recordings
      */
-    public StreamManager(ThreadPoolManager threadPoolManager, AudioBroadcaster audioBroadcaster, Path tempDirectory)
+    public StreamManager(ThreadPoolManager threadPoolManager, Listener<AudioRecording> listener,
+                         BroadcastFormat broadcastFormat, Path tempDirectory)
     {
         assert(tempDirectory != null && Files.isDirectory(tempDirectory));
         assert(mThreadPoolManager != null);
 
         mThreadPoolManager = threadPoolManager;
-        mAudioBroadcaster = audioBroadcaster;
+        mAudioRecordingListener = listener;
+        mBroadcastFormat = broadcastFormat;
         mTempDirectory = tempDirectory;
     }
 
@@ -141,7 +144,7 @@ public class StreamManager implements Listener<AudioPacket>
                     else
                     {
                         AudioRecorder recorder = BroadcastFactory.getAudioRecorder(getTemporaryRecordingPath(),
-                                mAudioBroadcaster.getBroadcastConfiguration().getBroadcastFormat());
+                                mBroadcastFormat);
                         recorder.start(mThreadPoolManager.getScheduledExecutorService());
                         recorder.receive(audioPacket);
                         mStreamRecorders.put(sourceChannelID, recorder);
@@ -178,13 +181,13 @@ public class StreamManager implements Listener<AudioPacket>
                 @Override
                 public void receive(AudioRecorder audioRecorder)
                 {
-                    StreamableAudioRecording streamableAudioRecording =
-                        new StreamableAudioRecording(audioRecorder.getPath(), audioRecorder.getMetadata(),
+                    AudioRecording audioRecording =
+                        new AudioRecording(audioRecorder.getPath(), audioRecorder.getMetadata(),
                             audioRecorder.getTimeRecordingStart(), audioRecorder.getRecordingLength());
 
-                    if(mAudioBroadcaster != null)
+                    if(mAudioRecordingListener != null)
                     {
-                        mAudioBroadcaster.receive(streamableAudioRecording);
+                        mAudioRecordingListener.receive(audioRecording);
                     }
                 }
             });
@@ -201,7 +204,7 @@ public class StreamManager implements Listener<AudioPacket>
         sb.append(BroadcastModel.TEMPORARY_STREAM_FILE_SUFFIX);
         sb.append(sNextRecordingNumber.incrementAndGet()).append("_");
         sb.append( TimeStamp.getLongTimeStamp( "_" ) );
-        sb.append( mAudioBroadcaster.getBroadcastConfiguration().getBroadcastFormat().getFileExtension() );
+        sb.append( mBroadcastFormat.getFileExtension() );
 
         Path temporaryRecordingPath = mTempDirectory.resolve(sb.toString());
 
diff --git a/src/record/AudioRecorder.java b/src/record/AudioRecorder.java
index 51568016..61f48158 100644
--- a/src/record/AudioRecorder.java
+++ b/src/record/AudioRecorder.java
@@ -169,7 +169,11 @@ public abstract class AudioRecorder extends Module implements Listener<AudioPack
     {
         mRecordingClosedListener = listener;
 
-        mRunning.set(false);
+        if(!mRunning.compareAndSet(true, false))
+        {
+            //If this recorder has already been stopped/closed, immediately inform the listener
+            mRecordingClosedListener.receive(this);
+        }
     }