From 648b12fd1622fd03c0563d71b72d55a314a4e49f Mon Sep 17 00:00:00 2001 From: Denny Date: Sat, 7 Jan 2017 07:31:47 -0500 Subject: [PATCH] #30 - resolves logic issue that was preventing deletion of temporary stream recordings. Changes the streamable recording queue in the AudioBroadcaster to find the recording with the earliest start time to prevent situations where a long recording might close out and enter the broadcast queue after a shorter recording that started later, but would have been replayed first. --- src/audio/broadcast/AudioBroadcaster.java | 43 +++++++++++++++-------- src/audio/broadcast/AudioRecording.java | 6 +++- 2 files changed, 34 insertions(+), 15 deletions(-) diff --git a/src/audio/broadcast/AudioBroadcaster.java b/src/audio/broadcast/AudioBroadcaster.java index cda9f47e..979dbad2 100644 --- a/src/audio/broadcast/AudioBroadcaster.java +++ b/src/audio/broadcast/AudioBroadcaster.java @@ -31,6 +31,9 @@ import sample.Listener; import java.io.ByteArrayInputStream; import java.io.IOException; import java.nio.file.Files; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -46,7 +49,7 @@ public abstract class AudioBroadcaster implements Listener private ScheduledFuture mScheduledTask; private RecordingQueueProcessor mRecordingQueueProcessor = new RecordingQueueProcessor(); - private LinkedTransferQueue mAudioRecordingQueue = new LinkedTransferQueue<>(); + private List mAudioRecordingQueue = new CopyOnWriteArrayList<>(); private ISilenceGenerator mSilenceGenerator; @@ -256,11 +259,10 @@ public abstract class AudioBroadcaster implements Listener if(!connected()) { //Remove all pending audio recordings - AudioRecording recording = mAudioRecordingQueue.poll(); - - while(recording != null) + while(!mAudioRecordingQueue.isEmpty()) { - recording = mAudioRecordingQueue.poll(); + AudioRecording recording = mAudioRecordingQueue.get(0); + recording.removePendingReplay(); } } } @@ -390,24 +392,37 @@ public abstract class AudioBroadcaster implements Listener mInputStream = null; - AudioRecording nextRecording = mAudioRecordingQueue.peek(); + AudioRecording nextRecording = null; + + //Find the recording with the earliest start time + Iterator it = mAudioRecordingQueue.iterator(); + + while(it.hasNext()) + { + AudioRecording recording = it.next(); + + if(nextRecording == null || (recording.getStartTime() < nextRecording.getStartTime())) + { + nextRecording = recording; + } + } if(nextRecording != null && nextRecording.getStartTime() + mDelay <= System.currentTimeMillis()) { - AudioRecording recording = mAudioRecordingQueue.poll(); + mAudioRecordingQueue.remove(nextRecording); try { - if(Files.exists(recording.getPath())) + if(Files.exists(nextRecording.getPath())) { - byte[] audio = Files.readAllBytes(recording.getPath()); + byte[] audio = Files.readAllBytes(nextRecording.getPath()); if(audio != null && audio.length > 0) { mInputStream = new ByteArrayInputStream(audio); mFinalSilencePadding = PROCESSOR_RUN_INTERVAL_MS - - (recording.getRecordingLength() % PROCESSOR_RUN_INTERVAL_MS); + (nextRecording.getRecordingLength() % PROCESSOR_RUN_INTERVAL_MS); while(mFinalSilencePadding >= PROCESSOR_RUN_INTERVAL_MS) { @@ -416,7 +431,7 @@ public abstract class AudioBroadcaster implements Listener if(connected()) { - broadcastMetadata(recording.getAudioMetadata()); + broadcastMetadata(nextRecording.getAudioMetadata()); } metadataUpdateRequired = false; @@ -426,18 +441,18 @@ public abstract class AudioBroadcaster implements Listener catch(IOException ioe) { mLog.error("Stream [" + getBroadcastConfiguration().getName() + "] error reading temporary audio " + - "stream recording [" + recording.getPath().toString() + "] - skipping recording - ", ioe); + "stream recording [" + nextRecording.getPath().toString() + "] - skipping recording - ", ioe); mInputStream = null; metadataUpdateRequired = false; } - recording.removePendingReplay(); + nextRecording.removePendingReplay(); broadcast(new BroadcastEvent(AudioBroadcaster.this, BroadcastEvent.Event.BROADCASTER_QUEUE_CHANGE)); } - //If we closed out a recording and don't have anything new, send an empty metadata update + //If we closed out a recording and don't have a new/next recording, send an empty metadata update if(metadataUpdateRequired && connected()) { broadcastMetadata(null); diff --git a/src/audio/broadcast/AudioRecording.java b/src/audio/broadcast/AudioRecording.java index a8538035..1d4c3f16 100644 --- a/src/audio/broadcast/AudioRecording.java +++ b/src/audio/broadcast/AudioRecording.java @@ -19,12 +19,16 @@ package audio.broadcast; import audio.metadata.AudioMetadata; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.nio.file.Path; import java.util.concurrent.atomic.AtomicInteger; public class AudioRecording implements Comparable { + private final static Logger mLog = LoggerFactory.getLogger(StreamManager.class); + private Path mPath; private AudioMetadata mAudioMetadata; private long mStartTime; @@ -110,6 +114,6 @@ public class AudioRecording implements Comparable */ public boolean hasPendingReplays() { - return mPendingReplayCount.get() <= 0; + return mPendingReplayCount.get() > 0; } }