From f51ae0b5dabace1438a497a69301e75b7db68939 Mon Sep 17 00:00:00 2001 From: Denny Date: Fri, 3 Mar 2017 05:56:32 -0500 Subject: [PATCH] Resolves issue with periodic instances of audio recorders not being shutdown correctly. Adds recorder monitor as a fail-safe to automatically shutdown recorders that don't receive an end-call indication. --- src/record/RecorderManager.java | 120 +++++-- src/record/wave/RealBufferWaveRecorder.java | 380 ++++++++++---------- src/record/wave/WaveWriter.java | 60 ++-- 3 files changed, 296 insertions(+), 264 deletions(-) diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index 871fe914..32957c31 100644 --- a/src/record/RecorderManager.java +++ b/src/record/RecorderManager.java @@ -29,79 +29,102 @@ import util.ThreadPool; import java.io.File; import java.util.HashMap; +import java.util.Iterator; import java.util.Map; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; public class RecorderManager implements Listener { private static final Logger mLog = LoggerFactory.getLogger(RecorderManager.class); public static final int AUDIO_SAMPLE_RATE = 8000; + public static final long IDLE_RECORDER_REMOVAL_THRESHOLD = 6000; //6 seconds private Map mRecorders = new HashMap<>(); + private ScheduledFuture mRecorderMonitorFuture; private boolean mCanStartNewRecorders = true; /** - * Manages all audio recording for all processing channels. Reconstructs audio streams and distributes channel - * audio to each of the recorders, starting and stopping the recorders as needed. + * Audio recording manager. Monitors stream of audio packets produced by decoding channels and automatically starts + * audio recorders when the channel's metadata designates a call as recordable. Routes call audio to each recorder + * based on audio packet metadata. Recorders are shutdown when the channel sends an end-call audio packet + * indicating that the call is complete. A separate recording monitor periodically checks for idled recorders to + * be stopped for cases when the channel fails to send an end-call audio packet. */ public RecorderManager() { + mRecorderMonitorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(new RecorderMonitor(), 2, + 2, TimeUnit.SECONDS); } + /** + * Prepares this class for shutdown + */ public void dispose() { + if(mRecorderMonitorFuture != null) + { + mRecorderMonitorFuture.cancel(true); + } } + /** + * Primary ingest point for audio packets from all decoding channels + * @param audioPacket to process + */ @Override public void receive(AudioPacket audioPacket) { - if(audioPacket.hasMetadata() && - audioPacket.getMetadata().isRecordable()) + if(audioPacket.hasMetadata() && audioPacket.getMetadata().isRecordable()) { String identifier = audioPacket.getMetadata().getUniqueIdentifier(); - if(mRecorders.containsKey(identifier)) + 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(); + } } } } @@ -150,4 +173,31 @@ public class RecorderManager implements Listener return new ComplexBufferWaveRecorder(AUDIO_SAMPLE_RATE, sb.toString()); } + + /** + * Monitors currently running recorders and removes/closes any recorders that are idle more than 6 seconds + */ + public class RecorderMonitor 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(); + } + } + } + } + } } diff --git a/src/record/wave/RealBufferWaveRecorder.java b/src/record/wave/RealBufferWaveRecorder.java index ac17988a..763c9ded 100644 --- a/src/record/wave/RealBufferWaveRecorder.java +++ b/src/record/wave/RealBufferWaveRecorder.java @@ -1,22 +1,33 @@ /******************************************************************************* - * SDR Trunk - * Copyright (C) 2014,2015 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 + * sdrtrunk + * Copyright (C) 2014-2017 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 + * ******************************************************************************/ package record.wave; +import module.Module; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sample.ConversionUtils; +import sample.Listener; +import sample.real.IFilteredRealBufferListener; +import sample.real.RealBuffer; +import util.TimeStamp; + +import javax.sound.sampled.AudioFormat; import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; @@ -26,206 +37,175 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import javax.sound.sampled.AudioFormat; - -import module.Module; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import sample.ConversionUtils; -import sample.Listener; -import sample.real.IFilteredRealBufferListener; -import sample.real.RealBuffer; -import util.TimeStamp; - /** * WAVE audio recorder module for recording real sample buffers to a wave file */ -public class RealBufferWaveRecorder extends Module - implements IFilteredRealBufferListener, Listener +public class RealBufferWaveRecorder extends Module + implements IFilteredRealBufferListener, Listener { - private final static Logger mLog = - LoggerFactory.getLogger( RealBufferWaveRecorder.class ); + private final static Logger mLog = LoggerFactory.getLogger(RealBufferWaveRecorder.class); private WaveWriter mWriter; private String mFilePrefix; private Path mFile; - private AudioFormat mAudioFormat; - - private BufferProcessor mBufferProcessor; - private ScheduledFuture mProcessorHandle; - private LinkedBlockingQueue mBuffers = - new LinkedBlockingQueue<>( 500 ); - private long mLastBufferReceived; - - private AtomicBoolean mRunning = new AtomicBoolean(); - - public RealBufferWaveRecorder( int sampleRate, String filePrefix ) - { - mAudioFormat = new AudioFormat( sampleRate, //SampleRate - 16, //Sample Size - 1, //Channels - true, //Signed - false ); //Little Endian + private AudioFormat mAudioFormat; - mFilePrefix = filePrefix; - } + private BufferProcessor mBufferProcessor; + private ScheduledFuture mProcessorHandle; + private LinkedBlockingQueue mBuffers = new LinkedBlockingQueue<>(500); + private long mLastBufferReceived; - /** - * Timestamp of when the latest buffer was received by this recorder - */ - public long getLastBufferReceived() - { - return mLastBufferReceived; - } + private AtomicBoolean mRunning = new AtomicBoolean(); - public Path getFile() - { - return mFile; - } - - public void start( ScheduledExecutorService executor ) - { - if( mRunning.compareAndSet( false, true ) ) - { - if( mBufferProcessor == null ) - { - mBufferProcessor = new BufferProcessor(); - } + public RealBufferWaveRecorder(int sampleRate, String filePrefix) + { + mAudioFormat = new AudioFormat(sampleRate, //SampleRate + 16, //Sample Size + 1, //Channels + true, //Signed + false); //Little Endian - try - { - StringBuilder sb = new StringBuilder(); - sb.append( mFilePrefix ); - sb.append( "_" ); - sb.append( TimeStamp.getLongTimeStamp( "_" ) ); - sb.append( ".wav" ); - - mFile = Paths.get( sb.toString() ); + mFilePrefix = filePrefix; + } - mWriter = new WaveWriter( mAudioFormat, mFile ); + /** + * Indicates if the recorder is currently running. + */ + public boolean isRunning() + { + return mRunning.get(); + } + + /** + * Timestamp of when the latest buffer was received by this recorder + */ + public long getLastBufferReceived() + { + return mLastBufferReceived; + } + + public Path getFile() + { + return mFile; + } + + public void start(ScheduledExecutorService executor) + { + if(mRunning.compareAndSet(false, true)) + { + if(mBufferProcessor == null) + { + mBufferProcessor = new BufferProcessor(); + } + + try + { + StringBuilder sb = new StringBuilder(); + sb.append(mFilePrefix); + sb.append("_"); + sb.append(TimeStamp.getLongTimeStamp("_")); + sb.append(".wav"); + + mFile = Paths.get(sb.toString()); + + mWriter = new WaveWriter(mAudioFormat, mFile); /* Schedule the processor to run every 500 milliseconds */ - mProcessorHandle = executor.scheduleAtFixedRate( - mBufferProcessor, 0, 500, TimeUnit.MILLISECONDS ); - } - catch( IOException io ) - { - mLog.error( "Error starting real buffer recorder", io ); - } - } - } - - public void stop() - { - /* Signal the buffer processor to end with a poison pill */ - receive( new PoisonPill() ); - - mRunning.set( false ); - } - - @Override - public void receive( RealBuffer buffer ) - { - if( mRunning.get() ) - { - boolean success = mBuffers.offer( buffer ); - - if( !success ) - { - mLog.error( "recorder buffer overflow - purging [" + - mFile.toFile().getAbsolutePath() + "]" ); - - mBuffers.clear(); - } - - mLastBufferReceived = System.currentTimeMillis(); - } - } - - @Override - public Listener getFilteredRealBufferListener() - { - return this; - } - - @Override - public void dispose() - { - stop(); - } - - @Override - public void reset() - { - } - - public class BufferProcessor implements Runnable - { - public void run() - { - try - { - RealBuffer buffer = mBuffers.poll(); - - while( buffer != null ) - { - if( buffer instanceof PoisonPill ) - { - buffer = null; - - mBuffers.clear(); - - if( mWriter != null ) - { - try - { - mWriter.close(); - } - catch( IOException io ) - { - mLog.error( "Error stopping complex wave recorder [" + - getFile() + "]", io ); - } - } - - if( mProcessorHandle != null ) - { - mProcessorHandle.cancel( true ); - } - - mProcessorHandle = null; - mWriter = null; - mFile = null; - } - else - { - mWriter.write( ConversionUtils.convertToSigned16BitSamples( buffer ) ); - buffer = mBuffers.poll(); - } - } + mProcessorHandle = executor.scheduleAtFixedRate(mBufferProcessor, 0, 500, TimeUnit.MILLISECONDS); } - catch ( IOException ioe ) - { - /* Stop this module if/when we get an IO exception */ - mBuffers.clear(); - stop(); - - mLog.error( "IOException while trying to write to the wave " - + "writer", ioe ); - } - } + catch(IOException io) + { + mLog.error("Error starting real buffer recorder", io); + } + } + } + + public void stop() + { + if(mRunning.compareAndSet(true, false)) + { + try + { + write(); + + if(mWriter != null) + { + mWriter.close(); + mWriter = null; + } + } + catch(IOException ioe) + { + mLog.error("Error writing final audio buffers to recording during shutdown", ioe); + } + } + } + + @Override + public void receive(RealBuffer buffer) + { + if(mRunning.get()) + { + boolean success = mBuffers.offer(buffer); + + if(!success) + { + mLog.error("recorder buffer overflow - purging [" + mFile.toFile().getAbsolutePath() + "]"); + mBuffers.clear(); + } + + mLastBufferReceived = System.currentTimeMillis(); + } + } + + @Override + public Listener getFilteredRealBufferListener() + { + return this; + } + + @Override + public void dispose() + { + stop(); + } + + @Override + public void reset() + { + } + + /** + * Writes all audio currently in the queue to the file + * @throws IOException if there are any errors writing the audio + */ + private void write() throws IOException + { + RealBuffer buffer = mBuffers.poll(); + + while(buffer != null) + { + mWriter.write(ConversionUtils.convertToSigned16BitSamples(buffer)); + buffer = mBuffers.poll(); + } + } + + public class BufferProcessor implements Runnable + { + public void run() + { + try + { + write(); + } + catch(IOException ioe) + { + /* Stop this module if/when we get an IO exception */ + mBuffers.clear(); + stop(); + + mLog.error("IO Exception while trying to write to the wave writer", ioe); + } + } } - - /** - * This is used as a sentinel value to signal the buffer processor to end - */ - public class PoisonPill extends RealBuffer - { - public PoisonPill() - { - super( new float[ 1 ] ); - } - } } diff --git a/src/record/wave/WaveWriter.java b/src/record/wave/WaveWriter.java index a7733e69..645cbf5c 100644 --- a/src/record/wave/WaveWriter.java +++ b/src/record/wave/WaveWriter.java @@ -1,19 +1,20 @@ /******************************************************************************* - * SDR Trunk - * Copyright (C) 2014 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 + * sdrtrunk + * Copyright (C) 2014-2017 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 + * ******************************************************************************/ package record.wave; @@ -34,10 +35,7 @@ import javax.sound.sampled.AudioFormat; public class WaveWriter implements AutoCloseable { -// private final static Logger mLog = LoggerFactory.getLogger( WaveWriter.class ); - - private static final Pattern FILENAME_PATTERN = - Pattern.compile( "(.*_)(\\d+)(\\.wav)" ); + private static final Pattern FILENAME_PATTERN = Pattern.compile( "(.*_)(\\d+)(\\.wav)" ); public static final long MAX_WAVE_SIZE = 2l * (long)Integer.MAX_VALUE; private AudioFormat mAudioFormat; @@ -58,8 +56,7 @@ public class WaveWriter implements AutoCloseable * @param maxSize - maximum file size ( range: 1 - 4,294,967,294 bytes ) * @throws IOException - if there are any IO issues */ - public WaveWriter( AudioFormat format, Path file, long maxSize ) - throws IOException + public WaveWriter( AudioFormat format, Path file, long maxSize ) throws IOException { Validate.isTrue(format != null); Validate.isTrue( file != null ); @@ -102,15 +99,11 @@ public class WaveWriter implements AutoCloseable while( Files.exists( mFile ) ) { - mFile = Paths.get( mFile.toFile().getAbsolutePath() - .replace( ".wav", "_" + version + ".wav" ) ); - + mFile = Paths.get( mFile.toFile().getAbsolutePath().replace( ".wav", "_" + version + ".wav" ) ); version++; } - Files.createFile( mFile ); - - mFileChannel = (FileChannel.open( mFile, StandardOpenOption.WRITE ) ); + mFileChannel = (FileChannel.open( mFile, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW ) ); ByteBuffer header = WaveUtils.getWaveHeader( mAudioFormat ); @@ -129,9 +122,18 @@ public class WaveWriter implements AutoCloseable { mFileChannel.force( true ); mFileChannel.close(); + mFileChannel = null; } - - /** + + @Override + protected void finalize() throws IOException + { + mFileChannel.force( true ); + mFileChannel.close(); + mFileChannel = null; + } + + /** * Writes the buffer contents to the file. Assumes that the buffer is full * and the first byte of data is at position 0. */