GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-12 08:37:25 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

#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.

Browse Source 
Note: there is a bug where temporary recordings are not being consistently deleted that I'm still working on.
This commit is contained in:
Denny
2017-01-03 09:27:17 -05:00
parent 4578e45f60
commit bc430b255f
5 changed files with 184 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
src/audio/broadcast/AudioBroadcaster.java
View File

@@ -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));
}

33
src/audio/broadcast/StreamableAudioRecording.java → src/audio/broadcast/AudioRecording.java
View File

@@ -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;
}
}

125
src/audio/broadcast/BroadcastModel.java
View File

@@ -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);
}
}
}
}

23
src/audio/broadcast/StreamManager.java
View File

@@ -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());

6
src/record/AudioRecorder.java
View File

@@ -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);
}
}