diff --git a/build/build.xml b/build/build.xml index 6105114e..d3dd6483 100644 --- a/build/build.xml +++ b/build/build.xml @@ -7,7 +7,7 @@ - + diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index 32957c31..4d9789e0 100644 --- a/src/record/RecorderManager.java +++ b/src/record/RecorderManager.java @@ -25,11 +25,15 @@ import properties.SystemProperties; import record.wave.ComplexBufferWaveRecorder; import record.wave.RealBufferWaveRecorder; import sample.Listener; +import sample.OverflowableTransferQueue; +import sample.real.IOverflowListener; import util.ThreadPool; import java.io.File; +import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -42,7 +46,8 @@ public class RecorderManager implements Listener public static final long IDLE_RECORDER_REMOVAL_THRESHOLD = 6000; //6 seconds private Map mRecorders = new HashMap<>(); - private ScheduledFuture mRecorderMonitorFuture; + private OverflowableTransferQueue mAudioPacketQueue = new OverflowableTransferQueue<>(1000, 100); + private ScheduledFuture mBufferProcessorFuture; private boolean mCanStartNewRecorders = true; @@ -55,8 +60,24 @@ public class RecorderManager implements Listener */ public RecorderManager() { - mRecorderMonitorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new RecorderMonitor(), 2, - 2, TimeUnit.SECONDS); + mAudioPacketQueue.setOverflowListener(new IOverflowListener() + { + @Override + public void sourceOverflow(boolean overflow) + { + if(overflow) + { + mLog.warn("overflow - audio packets will be dropped until recording catches up"); + } + else + { + mLog.info("audio recorder packet processing has returned to normal"); + } + } + }); + + mBufferProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferProcessor(), 0, + 1, TimeUnit.SECONDS); } /** @@ -64,9 +85,9 @@ public class RecorderManager implements Listener */ public void dispose() { - if(mRecorderMonitorFuture != null) + if(mBufferProcessorFuture != null) { - mRecorderMonitorFuture.cancel(true); + mBufferProcessorFuture.cancel(true); } } @@ -79,10 +100,25 @@ public class RecorderManager implements Listener { if(audioPacket.hasMetadata() && audioPacket.getMetadata().isRecordable()) { - String identifier = audioPacket.getMetadata().getUniqueIdentifier(); + mAudioPacketQueue.offer(audioPacket); + } + } - synchronized(mRecorders) + /** + * Process any queued audio buffers and dispatch them to the audio recorders + */ + private void processBuffers() + { + List audioPackets = new ArrayList<>(); + + mAudioPacketQueue.drainTo(audioPackets, 50); + + while(!audioPackets.isEmpty()) + { + for(AudioPacket audioPacket: audioPackets) { + String identifier = audioPacket.getMetadata().getUniqueIdentifier(); + if(mRecorders.containsKey(identifier)) { RealBufferWaveRecorder recorder = mRecorders.get(identifier); @@ -129,6 +165,29 @@ public class RecorderManager implements Listener } } } + + audioPackets.clear(); + mAudioPacketQueue.drainTo(audioPackets, 50); + } + } + + /** + * Removes recorders that have not received any new audio buffers in the last 6 seconds. + */ + private void removeIdleRecorders() + { + Iterator> it = mRecorders.entrySet().iterator(); + + while(it.hasNext()) + { + Map.Entry entry = it.next(); + + if(entry.getValue().getLastBufferReceived() + IDLE_RECORDER_REMOVAL_THRESHOLD < System.currentTimeMillis()) + { + mLog.info("Removing idle recorder [" + entry.getKey() + "]"); + it.remove(); + entry.getValue().stop(); + } } } @@ -175,29 +234,16 @@ public class RecorderManager implements Listener } /** - * Monitors currently running recorders and removes/closes any recorders that are idle more than 6 seconds + * Processes queued audio packets and distributes to each of the audio recorders. Removes any idle recorders + * that have not been updated according to an idle threshold period */ - public class RecorderMonitor implements Runnable + public class BufferProcessor implements Runnable { @Override public void run() { - synchronized(mRecorders) - { - Iterator> it = mRecorders.entrySet().iterator(); - - while(it.hasNext()) - { - Map.Entry entry = it.next(); - - if(entry.getValue().getLastBufferReceived() + IDLE_RECORDER_REMOVAL_THRESHOLD < System.currentTimeMillis()) - { - mLog.info("Removing idle recorder [" + entry.getKey() + "]"); - it.remove(); - entry.getValue().stop(); - } - } - } + processBuffers(); + removeIdleRecorders(); } } }