diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index 32957c31..cb30aa08 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); + mBufferProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferProcessor(), 0, + 1, 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"); + } + } + }); } /** @@ -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,56 +100,97 @@ 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) { - if(mRecorders.containsKey(identifier)) + String identifier = audioPacket.getMetadata().getUniqueIdentifier(); + + synchronized(mRecorders) { - RealBufferWaveRecorder recorder = mRecorders.get(identifier); - - if(audioPacket.getType() == AudioPacket.Type.AUDIO) + if(mRecorders.containsKey(identifier)) { - recorder.receive(audioPacket.getAudioBuffer()); - } - else if(audioPacket.getType() == AudioPacket.Type.END) - { - RealBufferWaveRecorder finished = mRecorders.remove(identifier); - finished.stop(); - } - } - else if(audioPacket.getType() == AudioPacket.Type.AUDIO) - { - if(mCanStartNewRecorders) - { - String filePrefix = getFilePrefix(audioPacket); + RealBufferWaveRecorder recorder = mRecorders.get(identifier); - RealBufferWaveRecorder recorder = null; - - try + if(audioPacket.getType() == AudioPacket.Type.AUDIO) { - recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix); - - recorder.start(ThreadPool.SCHEDULED); - recorder.receive(audioPacket.getAudioBuffer()); - mRecorders.put(identifier, recorder); } - catch(Exception ioe) + else if(audioPacket.getType() == AudioPacket.Type.END) { - mCanStartNewRecorders = false; + RealBufferWaveRecorder finished = mRecorders.remove(identifier); + finished.stop(); + } + } + else if(audioPacket.getType() == AudioPacket.Type.AUDIO) + { + if(mCanStartNewRecorders) + { + String filePrefix = getFilePrefix(audioPacket); - mLog.error("Error attempting to start new audio wave recorder. All (future) audio recording " + - "is disabled", ioe); + RealBufferWaveRecorder recorder = null; - if(recorder != null) + try { - recorder.stop(); + recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix); + + recorder.start(ThreadPool.SCHEDULED); + + recorder.receive(audioPacket.getAudioBuffer()); + mRecorders.put(identifier, recorder); + } + catch(Exception ioe) + { + mCanStartNewRecorders = false; + + mLog.error("Error attempting to start new audio wave recorder. All (future) audio recording " + + "is disabled", ioe); + + if(recorder != null) + { + recorder.stop(); + } } } } } } + + 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,28 +237,18 @@ 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(); } } }