From 372d2ff870ac29a343ff1e0befe50b575c62f282 Mon Sep 17 00:00:00 2001 From: Denny Sheirer Date: Tue, 11 Apr 2023 04:07:41 -0400 Subject: [PATCH] #1506 Resolves multiple performance issues: tuner freezing after a period of run time, buffer overflows, degraded audio playback. (#1513) -Reduces message and event histories per channel from 500 to 200. -Resolves issue of tuner freezing due to cancelled USB transfer buffers not being resubmitted. -Reworks the audio playback system to remove in-progress call preemption and remove thread locks which can cause glitchy behavior during audio playback. -Updates audio outputs to also use single thread pool for processing audio segments. -Updates MBECallSequenceConverter to batch convert MBE to WAV audio. -Unifies recording filename prefixes across baseband, mbe and binary recordings. -Resolves issue with registering heartbeat manager to dispatcher. -Uses single-thread thread pools for Dispatcher to support batch & process to overcome context switching consistency issue in Windows. Co-authored-by: Dennis Sheirer --- .../dsheirer/audio/AbstractAudioModule.java | 11 +- .../github/dsheirer/audio/AudioSegment.java | 53 ++- .../audio/codec/mbe/JmbeAudioModule.java | 2 +- .../codec/mbe/MBECallSequenceConverter.java | 252 ++++++---- .../dsheirer/audio/playback/AudioOutput.java | 444 ++++++++---------- .../audio/playback/AudioPlaybackManager.java | 66 ++- .../channel/ChannelProcessingManager.java | 41 +- .../AbstractComplexPolyphaseChannelizer.java | 12 +- .../ComplexPolyphaseChannelizerM2.java | 34 +- .../channelizer/PolyphaseChannelManager.java | 26 +- .../channelizer/PolyphaseChannelSource.java | 26 +- .../output/ChannelOutputProcessor.java | 24 +- .../output/OneChannelOutputProcessor.java | 7 +- .../output/TwoChannelOutputProcessor.java | 7 +- .../playback/PlaybackPreferenceEditor.java | 140 +----- .../module/decode/event/DecodeEventModel.java | 4 +- .../decode/event/MessageActivityModel.java | 35 +- .../p25/audio/P25P1CallSequenceRecorder.java | 34 +- .../dsheirer/preference/TimestampFormat.java | 12 +- .../playback/PlaybackPreference.java | 97 +--- .../dsheirer/record/RecorderFactory.java | 74 ++- .../record/binary/BinaryRecorder.java | 19 +- .../wave/ComplexSamplesWaveRecorder.java | 9 +- .../record/wave/NativeBufferWaveRecorder.java | 14 +- .../dsheirer/record/wave/WaveWriter.java | 24 +- .../github/dsheirer/source/SourceEvent.java | 10 - .../channel/HalfBandTunerChannelSource.java | 43 +- .../channel/PassThroughChannelSource.java | 30 +- .../channel/StreamProcessorWithHeartbeat.java | 129 ----- .../tuner/channel/TunerChannelSource.java | 1 - .../HeterodyneChannelSourceManager.java | 17 +- .../tuner/rtl/r820t/R820TTunerEditor.java | 1 - .../source/tuner/usb/USBTunerController.java | 10 +- .../io/github/dsheirer/util/Dispatcher.java | 206 ++++---- 34 files changed, 802 insertions(+), 1112 deletions(-) delete mode 100644 src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java diff --git a/src/main/java/io/github/dsheirer/audio/AbstractAudioModule.java b/src/main/java/io/github/dsheirer/audio/AbstractAudioModule.java index 770c4d17..beae5d6e 100644 --- a/src/main/java/io/github/dsheirer/audio/AbstractAudioModule.java +++ b/src/main/java/io/github/dsheirer/audio/AbstractAudioModule.java @@ -1,7 +1,6 @@ /* - * ****************************************************************************** - * sdrtrunk - * Copyright (C) 2014-2018 Dennis Sheirer + * ***************************************************************************** + * Copyright (C) 2014-2023 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 @@ -15,7 +14,7 @@ * * You should have received a copy of the GNU General Public License * along with this program. If not, see - * ***************************************************************************** + * **************************************************************************** */ package io.github.dsheirer.audio; @@ -106,7 +105,7 @@ public abstract class AbstractAudioModule extends Module implements IAudioSegmen * Gets the current audio segment, or creates a new audio segment as necessary and broadcasts it to any registered * listener(s). */ - protected AudioSegment getAudioSegment() + public AudioSegment getAudioSegment() { synchronized(this) { @@ -135,7 +134,7 @@ public abstract class AbstractAudioModule extends Module implements IAudioSegmen } } - protected void addAudio(float[] audioBuffer) + public void addAudio(float[] audioBuffer) { AudioSegment audioSegment = getAudioSegment(); diff --git a/src/main/java/io/github/dsheirer/audio/AudioSegment.java b/src/main/java/io/github/dsheirer/audio/AudioSegment.java index 1f40ecfa..b16d6369 100644 --- a/src/main/java/io/github/dsheirer/audio/AudioSegment.java +++ b/src/main/java/io/github/dsheirer/audio/AudioSegment.java @@ -1,18 +1,21 @@ -/******************************************************************************* - * sdr-trunk - * Copyright (C) 2014-2020 Dennis Sheirer +/* + * ***************************************************************************** + * Copyright (C) 2014-2023 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 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. + * 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 - * - ******************************************************************************/ + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ package io.github.dsheirer.audio; @@ -26,15 +29,6 @@ import io.github.dsheirer.identifier.IdentifierUpdateNotification; import io.github.dsheirer.identifier.MutableIdentifierCollection; import io.github.dsheirer.sample.Broadcaster; import io.github.dsheirer.sample.Listener; -import javafx.beans.property.BooleanProperty; -import javafx.beans.property.IntegerProperty; -import javafx.beans.property.SimpleBooleanProperty; -import javafx.beans.property.SimpleIntegerProperty; -import javafx.collections.FXCollections; -import javafx.collections.ObservableSet; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.util.Collection; import java.util.Collections; import java.util.Date; @@ -43,6 +37,14 @@ import java.util.List; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicInteger; +import javafx.beans.property.BooleanProperty; +import javafx.beans.property.IntegerProperty; +import javafx.beans.property.SimpleBooleanProperty; +import javafx.beans.property.SimpleIntegerProperty; +import javafx.collections.FXCollections; +import javafx.collections.ObservableSet; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Audio segment containing all related metadata and a dynamic collection of audio packets. An audio segment can be @@ -138,6 +140,15 @@ public class AudioSegment implements Listener return mComplete; } + /** + * Indicates if this audio segment is completed. + * @return true if completed. + */ + public boolean isComplete() + { + return mComplete.get(); + } + /** * Duplicate call audio property. This flag is set to true whenever a duplicate call detection function detects * an audio segment is a duplicate. diff --git a/src/main/java/io/github/dsheirer/audio/codec/mbe/JmbeAudioModule.java b/src/main/java/io/github/dsheirer/audio/codec/mbe/JmbeAudioModule.java index 56c19f01..2eae1166 100644 --- a/src/main/java/io/github/dsheirer/audio/codec/mbe/JmbeAudioModule.java +++ b/src/main/java/io/github/dsheirer/audio/codec/mbe/JmbeAudioModule.java @@ -67,7 +67,7 @@ public abstract class JmbeAudioModule extends AbstractAudioModule implements Lis mAudioCodec = null; } - protected IAudioCodec getAudioCodec() + public IAudioCodec getAudioCodec() { return mAudioCodec; } diff --git a/src/main/java/io/github/dsheirer/audio/codec/mbe/MBECallSequenceConverter.java b/src/main/java/io/github/dsheirer/audio/codec/mbe/MBECallSequenceConverter.java index 3dea871f..e99fc779 100644 --- a/src/main/java/io/github/dsheirer/audio/codec/mbe/MBECallSequenceConverter.java +++ b/src/main/java/io/github/dsheirer/audio/codec/mbe/MBECallSequenceConverter.java @@ -1,27 +1,46 @@ /* + * ***************************************************************************** + * Copyright (C) 2014-2023 Dennis Sheirer * - * * ****************************************************************************** - * * Copyright (C) 2014-2019 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 - * * ***************************************************************************** + * 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 io.github.dsheirer.audio.codec.mbe; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.github.dsheirer.alias.AliasList; +import io.github.dsheirer.audio.AudioSegment; +import io.github.dsheirer.identifier.IdentifierUpdateNotification; +import io.github.dsheirer.module.decode.p25.audio.P25P1AudioModule; +import io.github.dsheirer.module.decode.p25.audio.P25P1CallSequenceRecorder; +import io.github.dsheirer.module.decode.p25.audio.VoiceFrame; +import io.github.dsheirer.module.decode.p25.identifier.radio.APCO25RadioIdentifier; +import io.github.dsheirer.module.decode.p25.identifier.talkgroup.APCO25Talkgroup; +import io.github.dsheirer.preference.UserPreferences; +import io.github.dsheirer.record.AudioSegmentRecorder; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collection; +import jmbe.iface.IAudioCodec; +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.filefilter.DirectoryFileFilter; +import org.apache.commons.io.filefilter.SuffixFileFilter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,79 +51,130 @@ public class MBECallSequenceConverter { private final static Logger mLog = LoggerFactory.getLogger(MBECallSequenceConverter.class); -// public static void convert(Path input, Path output) throws IOException -// { -// InputStream inputStream = Files.newInputStream(input); -// ObjectMapper mapper = new ObjectMapper(); -// MBECallSequence sequence = mapper.readValue(inputStream, MBECallSequence.class); -// convert(sequence, output); -// } -// -// public static void convert(MBECallSequence callSequence, Path outputPath) -// { -// if(callSequence == null || callSequence.isEncrypted()) -// { -// throw new IllegalArgumentException("Cannot decode null or encrypted call sequence"); -// } -// -// if(callSequence != null && !callSequence.isEncrypted()) -// { -// ThumbDv.AudioProtocol protocol = ThumbDv.AudioProtocol.P25_PHASE2; -// -// AudioPacketWaveRecorder recorder = new AudioPacketWaveRecorder(outputPath); -// recorder.start(); -// -// long delayMillis = 0; -// -// try(ThumbDv thumbDv = new ThumbDv(protocol, recorder)) -// { -// thumbDv.start(); -// for(VoiceFrame voiceFrame: callSequence.getVoiceFrames()) -// { -// mLog.debug("Frame [" + voiceFrame.getFrame() + "] + Hex [" + AmbeResponse.toHex(voiceFrame.getFrameBytes()) + "]"); -// thumbDv.decode(voiceFrame.getFrameBytes()); -// delayMillis += 30; -// } -// -// if(delayMillis > 0) -// { -// delayMillis += 1000; -// try -// { -// Thread.sleep(delayMillis); -// } -// catch(InterruptedException ie) -// { -// -// } -// } -// } -// catch(IOException ioe) -// { -// mLog.error("Error", ioe); -// } -// -// recorder.stop(Paths.get(outputPath.toString().replace(".tmp", ".wav")), new WaveMetadata()); -// } -// } -// -// public static void main(String[] args) -// { -// String mbe = "/home/denny/SDRTrunk/recordings/20190331085324_154250000_3_TS0_65084_6591007.mbe"; -// String mbe = "/home/denny/SDRTrunk/recordings/20190331085324_154250000_2_TS1_65035.mbe"; -// -// Path input = Paths.get(mbe); -// Path output = Paths.get(mbe.replace(".mbe", ".tmp")); -// -// mLog.info("Converting: " + mbe); -// -// try -// { -// MBECallSequenceConverter.convert(input, output); -// } -// catch(IOException ioe) -// { -// mLog.error("Error", ioe); -// } -// } + /** + * Converts the input MBE file to PCM audio and writes to the output wave file. + * @param input path to the MBE file. + * @param output path to write the WAVE recording. + * @throws IOException if there is an error. + */ + public static void convert(Path input, Path output) throws IOException + { + InputStream inputStream = Files.newInputStream(input); + ObjectMapper mapper = new ObjectMapper(); + MBECallSequence sequence = mapper.readValue(inputStream, MBECallSequence.class); + convert(sequence, output); + } + + public static void convert(MBECallSequence callSequence, Path outputPath) + { + if(callSequence == null || callSequence.isEncrypted()) + { + throw new IllegalArgumentException("Cannot decode null or encrypted call sequence"); + } + + if(callSequence != null) + { + if(callSequence.getProtocol().equals(P25P1CallSequenceRecorder.PROTOCOL)) + { + P25P1AudioModule audioModule = new P25P1AudioModule(new UserPreferences(), new AliasList("mbe generator")); + audioModule.setRecordAudio(true); + audioModule.start(); + + if(callSequence.getFromIdentifier() != null) + { + int from = 0; + + try + { + from = Integer.parseInt(callSequence.getFromIdentifier()); + audioModule.getIdentifierUpdateListener().receive(new IdentifierUpdateNotification(APCO25RadioIdentifier.createFrom(from), + IdentifierUpdateNotification.Operation.ADD, 0)); + } + catch(Exception e) + { + mLog.error("Error parsing from identifier from value [" + callSequence.getFromIdentifier()); + } + } + + if(callSequence.getToIdentifier() != null) + { + int to = 0; + + try + { + to = Integer.parseInt(callSequence.getToIdentifier()); + audioModule.getIdentifierUpdateListener().receive(new IdentifierUpdateNotification(APCO25Talkgroup.create(to), + IdentifierUpdateNotification.Operation.ADD, 0)); + } + catch(Exception e) + { + mLog.error("Error parsing from identifier from value [" + callSequence.getFromIdentifier()); + } + } + + IAudioCodec codec = audioModule.getAudioCodec(); + + for(VoiceFrame voiceFrame: callSequence.getVoiceFrames()) + { + byte[] frameBytes = voiceFrame.getFrameBytes(); + float[] audio = codec.getAudio(frameBytes); + audioModule.addAudio(audio); + } + + AudioSegment audioSegment = audioModule.getAudioSegment(); + + try + { + AudioSegmentRecorder.recordWAVE(audioSegment, outputPath); + } + catch(IOException ioe) + { + mLog.error("Error writing audio segment, ioe"); + } + + audioModule.stop(); + } + } + } + + public static void main(String[] args) + { + boolean all = true; + + String path = "/home/denny/SDRTrunk/recordings"; + Path input = Paths.get(path); + + if(all) + { + Collection mbeFiles = FileUtils.listFiles(input.toFile(), new SuffixFileFilter(".mbe"), DirectoryFileFilter.DIRECTORY); + + for(File inputFile: mbeFiles) + { + Path output = Paths.get(inputFile.getAbsolutePath().replace(".mbe", ".wav")); + mLog.info("Converting: " + inputFile); + try + { + MBECallSequenceConverter.convert(inputFile.toPath(), output); + } + catch(IOException ioe) + { + mLog.error("Error", ioe); + } + } + } + else + { + Path output = Paths.get(path.replace(".mbe", ".wav")); + mLog.info("Converting: " + path); + + try + { + MBECallSequenceConverter.convert(input, output); + } + catch(IOException ioe) + { + mLog.error("Error", ioe); + } + } + } } diff --git a/src/main/java/io/github/dsheirer/audio/playback/AudioOutput.java b/src/main/java/io/github/dsheirer/audio/playback/AudioOutput.java index 7738927f..9834467c 100644 --- a/src/main/java/io/github/dsheirer/audio/playback/AudioOutput.java +++ b/src/main/java/io/github/dsheirer/audio/playback/AudioOutput.java @@ -1,30 +1,27 @@ /* + * ***************************************************************************** + * Copyright (C) 2014-2023 Dennis Sheirer * - * * ****************************************************************************** - * * Copyright (C) 2014-2020 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 - * * ***************************************************************************** + * 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 io.github.dsheirer.audio.playback; import com.google.common.eventbus.Subscribe; -import io.github.dsheirer.alias.id.priority.Priority; import io.github.dsheirer.audio.AudioEvent; import io.github.dsheirer.audio.AudioSegment; +import io.github.dsheirer.controller.NamingThreadFactory; import io.github.dsheirer.eventbus.MyEventBus; import io.github.dsheirer.identifier.IdentifierCollection; import io.github.dsheirer.identifier.IdentifierUpdateNotification; @@ -33,11 +30,13 @@ import io.github.dsheirer.preference.UserPreferences; import io.github.dsheirer.sample.Broadcaster; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.source.mixer.MixerChannel; -import io.github.dsheirer.util.ThreadPool; -import javafx.beans.property.BooleanProperty; -import javafx.beans.property.IntegerProperty; -import javafx.beans.property.SimpleBooleanProperty; -import javafx.beans.property.SimpleIntegerProperty; +import java.nio.ByteBuffer; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,10 +50,6 @@ import javax.sound.sampled.LineListener; import javax.sound.sampled.LineUnavailableException; import javax.sound.sampled.Mixer; import javax.sound.sampled.SourceDataLine; -import java.nio.ByteBuffer; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; /** * Audio output/playback channel for a single audio mixer channel. Providers support for playback of audio segments @@ -67,7 +62,6 @@ public abstract class AudioOutput implements LineListener, Listener mIdentifierCollectionListener; private Broadcaster mAudioEventBroadcaster = new Broadcaster<>(); - private ScheduledFuture mProcessorFuture; private SourceDataLine mOutput; private Mixer mMixer; private MixerChannel mMixerChannel; @@ -76,17 +70,17 @@ public abstract class AudioOutput implements LineListener, Listener mAudioSegmentQueue = new LinkedTransferQueue<>(); private AudioSegment mCurrentAudioSegment; - private AudioSegment mNextAudioSegment; - private ReentrantLock mLock = new ReentrantLock(); private int mCurrentBufferIndex = 0; private UserPreferences mUserPreferences; - private BooleanProperty mEmptyProperty = new SimpleBooleanProperty(true); - private IntegerProperty mAudioPriority = new SimpleIntegerProperty(Priority.DEFAULT_PRIORITY); private ByteBuffer mAudioSegmentStartTone; - private ByteBuffer mAudioSegmentPreemptTone; private ByteBuffer mAudioSegmentDropTone; private boolean mRunning = false; + private ScheduledExecutorService mScheduledExecutorService; + private ScheduledFuture mProcessorFuture; + private long mOutputLastTimestamp = 0; + private static final long STALE_PLAYBACK_THRESHOLD_MS = 500; /** * Single audio channel playback with automatic starting and stopping of the @@ -107,6 +101,8 @@ public abstract class AudioOutput implements LineListener, Listener 0) + { + playAudio(mAudioSegmentStartTone); + } + } + + mCurrentBufferIndex = 0; + } + + /** + * Indicates if the audio segment should be thrown away because it is marked as duplicate or do-not-monitor. + * @param audioSegment to evaluate + * @return true if the audio should be thrown away. + */ + private boolean isThrowaway(AudioSegment audioSegment) + { + return audioSegment != null && + (audioSegment.isDoNotMonitor() || + (audioSegment.isDuplicate() && mUserPreferences.getDuplicateCallDetectionPreference() + .isDuplicatePlaybackSuppressionEnabled())); + } + /** * Manage audio segment playback and process audio segment buffers. This method is designed to be called - * by a threaded processor repeatedly to playback the current audio segment and check for and start a newly - * assigned audio segment. It also handles starting and stopping the playback source data line to avoid audio + * by a threaded processor repeatedly to playback the current audio segment and check for and start newly + * added audio segments. It also handles starting and stopping the playback source data line to avoid audio * discontinuities due to buffer underruns. */ private void processAudio() { - if(mNextAudioSegment != null) + if(mCurrentAudioSegment == null) { - mLock.lock(); - - try - { - //Remove an assigned segment that's subsequently flagged as duplicate or do not monitor before playback - if(mNextAudioSegment.isDoNotMonitor() || (mNextAudioSegment.isDuplicate() && - mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled())) - { - mNextAudioSegment = null; - - if(mCurrentAudioSegment == null) - { - mEmptyProperty.set(true); - } - } - //For linked audio segments, allow the linked segment to complete first before assigning the next - else if(mNextAudioSegment.isLinked()) - { - if(mCurrentAudioSegment == null) - { - mCurrentAudioSegment = mNextAudioSegment; - mNextAudioSegment = null; - mCurrentBufferIndex = 0; - - if(mCurrentAudioSegment != null) - { - mAudioPriority.bind(mCurrentAudioSegment.monitorPriorityProperty()); - mCurrentAudioSegment.addIdentifierUpdateNotificationListener(this); - broadcast(mCurrentAudioSegment.getIdentifierCollection()); - } - else - { - mAudioPriority.setValue(Priority.DEFAULT_PRIORITY); - } - } - } - else - { - //Insert audio segment start or audio priority preemption bonk tone - if(mCurrentAudioSegment == null) - { - playAudio(getAudioSegmentStartTone()); - } - else if(mCurrentBufferIndex > 0 && - (!mCurrentAudioSegment.completeProperty().get() || - mCurrentBufferIndex < mCurrentAudioSegment.getAudioBufferCount())) - { - playAudio(getAudioSegmentPreemptionTone()); - } - else - { - playAudio(getAudioSegmentStartTone()); - } - - //Close current audio segment - dispose(mCurrentAudioSegment); - mCurrentAudioSegment = mNextAudioSegment; - mNextAudioSegment = null; - mCurrentBufferIndex = 0; - - if(mCurrentAudioSegment != null) - { - mAudioPriority.bind(mCurrentAudioSegment.monitorPriorityProperty()); - mCurrentAudioSegment.addIdentifierUpdateNotificationListener(this); - broadcast(mCurrentAudioSegment.getIdentifierCollection()); - } - else - { - mAudioPriority.setValue(Priority.DEFAULT_PRIORITY); - } - } - } - finally - { - mLock.unlock(); - } + loadNextAudioSegment(); } - if(mCurrentAudioSegment != null) + //Reevaluate current audio segment to see if the status has changed for duplicate or do-not-monitor. + if(isThrowaway(mCurrentAudioSegment)) { - //Check for completed audio segment or a segment flagged as duplicate or Do Not Monitor - if(mCurrentAudioSegment.isDoNotMonitor() || - (mCurrentAudioSegment.isDuplicate() && - mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled()) || - (mCurrentAudioSegment.completeProperty().get() && mCurrentBufferIndex >= mCurrentAudioSegment.getAudioBufferCount())) + if(mCurrentBufferIndex > 0) { - if(mCurrentAudioSegment.isDoNotMonitor()) - { - playAudio(getAudioSegmentDropTone()); - } - - dispose(mCurrentAudioSegment); - mCurrentAudioSegment = null; - - mLock.lock(); - - try - { - if(mNextAudioSegment == null) - { - mEmptyProperty.set(true); - } - } - finally - { - mLock.unlock(); - } - - return; + playAudio(mAudioSegmentDropTone); } - //Process any new buffers that have been added to the audio segment. If a next audio segment gets assigned - //while processing, exit the loop so that we can evaluate the next for higher priority preempt. If the next - //segment is a linked segment, ignore it so that we can close out the current segment. - while(mCurrentAudioSegment != null && (mNextAudioSegment == null || mNextAudioSegment.isLinked()) && - mCurrentBufferIndex < mCurrentAudioSegment.getAudioBufferCount() && - !mCurrentAudioSegment.isDoNotMonitor() && !(mCurrentAudioSegment.isDuplicate() && - mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled())) + disposeCurrentAudioSegment(); + loadNextAudioSegment(); + } + + while(mCurrentAudioSegment != null && mCurrentBufferIndex < mCurrentAudioSegment.getAudioBufferCount()) + { + try { float[] audioBuffer = mCurrentAudioSegment.getAudioBuffers().get(mCurrentBufferIndex++); if(audioBuffer != null) { ByteBuffer audio = convert(audioBuffer); + //This call blocks until all audio bytes are dumped into the data line. playAudio(audio); } } + catch(Exception e) + { + mLog.error("Error while processing audio for [" + mMixerChannel.name() + "]", e); + } + } + + //Check for completed and fully-played audio segment -- load next audio segment + if(mCurrentAudioSegment != null && + mCurrentAudioSegment.isComplete() && + (mCurrentBufferIndex >= mCurrentAudioSegment.getAudioBufferCount())) + { + disposeCurrentAudioSegment(); } checkStop(); @@ -523,25 +434,7 @@ public abstract class AudioOutput implements LineListener, Listener= mBufferStopThreshold) + if(mOutput.isRunning()) { - mOutput.drain(); - mOutput.stop(); - mRunning = false; + if(mOutput.available() >= mBufferStopThreshold) + { + mOutput.drain(); + mOutput.stop(); + mRunning = false; + } + //Detect when we have a partially filled output that will neither start nor stop and the last write + //timestamp was more than the staleness threshold (500 ms) -- stop the output and discard stale contents + else if(mOutput.available() < mBufferStopThreshold && + mOutput.available() > mBufferStartThreshold && + (System.currentTimeMillis() - mOutputLastTimestamp >= STALE_PLAYBACK_THRESHOLD_MS)) + { + mOutput.stop(); + //Discard the stale buffer contents. + mOutput.flush(); + mRunning = false; + } } //If output playback stopped on its own because the buffer emptied, then cleanup - else if(!mOutput.isRunning()) + else { mRunning = false; } @@ -735,16 +642,33 @@ public abstract class AudioOutput implements LineListener, Listener - * ***************************************************************************** + * **************************************************************************** */ package io.github.dsheirer.audio.playback; @@ -24,6 +23,7 @@ import io.github.dsheirer.audio.AudioEvent; import io.github.dsheirer.audio.AudioException; import io.github.dsheirer.audio.AudioSegment; import io.github.dsheirer.audio.IAudioController; +import io.github.dsheirer.controller.NamingThreadFactory; import io.github.dsheirer.eventbus.MyEventBus; import io.github.dsheirer.preference.PreferenceType; import io.github.dsheirer.preference.UserPreferences; @@ -31,17 +31,18 @@ import io.github.dsheirer.sample.Broadcaster; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.source.mixer.MixerChannel; import io.github.dsheirer.source.mixer.MixerChannelConfiguration; -import io.github.dsheirer.util.ThreadPool; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.util.ArrayList; import java.util.Comparator; import java.util.Iterator; import java.util.List; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Manages scheduling and playback of audio segments to the local users audio system. @@ -50,8 +51,6 @@ public class AudioPlaybackManager implements Listener, IAudioContr { private static final Logger mLog = LoggerFactory.getLogger(AudioPlaybackManager.class); - public static final String AUDIO_CHANNELS_PROPERTY = "audio.manager.channels"; - public static final String AUDIO_MIXER_PROPERTY = "audio.manager.mixer"; public static final AudioEvent CONFIGURATION_CHANGE_STARTED = new AudioEvent(AudioEvent.Type.AUDIO_CONFIGURATION_CHANGE_STARTED, null); public static final AudioEvent CONFIGURATION_CHANGE_COMPLETE = @@ -64,6 +63,8 @@ public class AudioPlaybackManager implements Listener, IAudioContr private List mAudioSegments = new ArrayList<>(); private List mPendingAudioSegments = new ArrayList<>(); private LinkedTransferQueue mNewAudioSegmentQueue = new LinkedTransferQueue<>(); + private ScheduledExecutorService mScheduledExecutorService = + Executors.newSingleThreadScheduledExecutor(new NamingThreadFactory("sdrtrunk audio manager")); /** * Constructs an instance. @@ -202,29 +203,13 @@ public class AudioPlaybackManager implements Listener, IAudioContr //Sort audio segments by playback priority and assign to empty audio outputs if(!mAudioSegments.isEmpty()) { + //TODO: change this to take audio segment start time into account also mAudioSegments.sort(Comparator.comparingInt(o -> o.monitorPriorityProperty().get())); //Assign empty audio outputs first for(AudioOutput audioOutput: mAudioOutputs) { - if(audioOutput.emptyProperty().get()) - { - audioOutput.play(mAudioSegments.remove(0)); - - if(mAudioSegments.isEmpty()) - { - return; - } - } - } - } - - //Preempt ongoing audio segment playback when higher priority audio segments are available - if(!mAudioSegments.isEmpty()) - { - for(AudioOutput audioOutput: mAudioOutputs) - { - if(mAudioSegments.get(0).monitorPriorityProperty().get() < audioOutput.audioPriorityProperty().get()) + if(audioOutput.isEmpty()) { audioOutput.play(mAudioSegments.remove(0)); @@ -284,7 +269,7 @@ public class AudioPlaybackManager implements Listener, IAudioContr } catch(AudioException ae) { - mLog.error("Error changing audio output to [" + configuration.toString() + "]", ae); + mLog.error("Error changing audio output to [" + configuration + "]", ae); } } } @@ -332,7 +317,7 @@ public class AudioPlaybackManager implements Listener, IAudioContr throw new AudioException("Unsupported mixer channel configuration: " + entry.getMixerChannel()); } - mProcessingTask = ThreadPool.SCHEDULED.scheduleAtFixedRate(new AudioSegmentProcessor(), + mProcessingTask = mScheduledExecutorService.scheduleAtFixedRate(new AudioSegmentProcessor(), 0, 100, TimeUnit.MILLISECONDS); mControllerBroadcaster.broadcast(CONFIGURATION_CHANGE_COMPLETE); mMixerChannelConfiguration = entry; @@ -343,7 +328,7 @@ public class AudioPlaybackManager implements Listener, IAudioContr * Current audio playback mixer channel configuration setting. */ @Override - public MixerChannelConfiguration getMixerChannelConfiguration() throws AudioException + public MixerChannelConfiguration getMixerChannelConfiguration() { return mMixerChannelConfiguration; } @@ -384,16 +369,23 @@ public class AudioPlaybackManager implements Listener, IAudioContr */ public class AudioSegmentProcessor implements Runnable { + private AtomicBoolean mProcessing = new AtomicBoolean(); + @Override public void run() { - try + if(mProcessing.compareAndSet(false, true)) { - processAudioSegments(); - } - catch(Throwable t) - { - mLog.error("Encountered error while processing audio segments", t); + try + { + processAudioSegments(); + } + catch(Throwable t) + { + mLog.error("Encountered error while processing audio segments", t); + } + + mProcessing.set(false); } } } diff --git a/src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java b/src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java index 65d6bcbb..449548e6 100644 --- a/src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java +++ b/src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -623,30 +623,27 @@ public class ChannelProcessingManager implements Listener Platform.runLater(() -> channel.setProcessing(false)); } - //Since the processing chain's source will block on stop(), throw it to the thread pool to run - ThreadPool.CACHED.submit(() -> { - try - { - processingChain.stop(); - processingChain.removeEventLoggingModules(); - processingChain.removeRecordingModules(); + try + { + processingChain.stop(); + processingChain.removeEventLoggingModules(); + processingChain.removeRecordingModules(); - //Deregister channel from receive frequency correction events to show in the spectral display (hack!) - processingChain.removeFrequencyChangeListener(channel); - channel.resetFrequencyCorrection(); + //Deregister channel from receive frequency correction events to show in the spectral display (hack!) + processingChain.removeFrequencyChangeListener(channel); + channel.resetFrequencyCorrection(); - mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP)); - mChannelEventBroadcaster.removeListener(processingChain); + mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP)); + mChannelEventBroadcaster.removeListener(processingChain); - //Unregister for event bus requests and notifications - processingChain.getEventBus().unregister(ChannelProcessingManager.this); - processingChain.dispose(); - } - catch(Exception e) - { - mLog.error("Error during shutdown of processing chain for channel [" + channel.getName() + "}", e); - } - }); + //Unregister for event bus requests and notifications + processingChain.getEventBus().unregister(ChannelProcessingManager.this); + processingChain.dispose(); + } + catch(Exception e) + { + mLog.error("Error during shutdown of processing chain for channel [" + channel.getName() + "}", e); + } } } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java index a84e6a79..a21bf25b 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -122,6 +122,11 @@ public abstract class AbstractComplexPolyphaseChannelizer implements Listener> { - public IFFTProcessorDispatcher(int maximumSize) + public IFFTProcessorDispatcher(int batchSize, long interval) { - super(maximumSize, "sdrtrunk polyphase ifft processor", Collections.emptyList()); + super("sdrtrunk polyphase ifft processor", batchSize, interval); //We create a listener interface to receive the batched channel results arrays from the scheduled thread pool //dispatcher thread that is part of this continuous buffer processor. We perform an IFFT on each //channel results array contained in each results buffer and then dispatch the buffer //so that it can be distributed to each channel listener. setListener(list -> { - List processedChannelResults = new ArrayList<>(); - - for(float[] channelResults: list) + try { - //Rotate each of the channels to the correct phase using the IFFT - mFFT.complexInverse(channelResults, true); - processedChannelResults.add(channelResults); - } + List processedChannelResults = new ArrayList<>(); - dispatch(processedChannelResults); + for(float[] channelResults: list) + { + //Rotate each of the channels to the correct phase using the IFFT + mFFT.complexInverse(channelResults, true); + processedChannelResults.add(channelResults); + } + + dispatch(processedChannelResults); + } + catch(Throwable t) + { + mLog.error("Error during IFFT and dispatch of processed channel results", t); + } }); } } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java index 83a72ea2..e4bc0153 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java @@ -20,7 +20,6 @@ package io.github.dsheirer.dsp.filter.channelizer; import io.github.dsheirer.buffer.INativeBuffer; import io.github.dsheirer.buffer.INativeBufferProvider; -import io.github.dsheirer.buffer.NativeBufferPoisonPill; import io.github.dsheirer.controller.channel.event.ChannelStopProcessingRequest; import io.github.dsheirer.dsp.filter.design.FilterDesignException; import io.github.dsheirer.eventbus.MyEventBus; @@ -28,7 +27,6 @@ import io.github.dsheirer.sample.Broadcaster; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.InterleavedComplexSamples; import io.github.dsheirer.source.ISourceEventProcessor; -import io.github.dsheirer.source.Source; import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.source.SourceException; import io.github.dsheirer.source.tuner.TunerController; @@ -110,8 +108,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor } mChannelCalculator = new ChannelCalculator(sampleRate, channelCount, frequency, CHANNEL_OVERSAMPLING); - mBufferDispatcher = new Dispatcher(500, "sdrtrunk polyphase buffer processor", - new NativeBufferPoisonPill()); + mBufferDispatcher = new Dispatcher("sdrtrunk polyphase buffer processor", 50, 10); mBufferDispatcher.setListener(mNativeBufferReceiver); } @@ -177,14 +174,6 @@ public class PolyphaseChannelManager implements ISourceEventProcessor } } - /** - * Current channel sample rate which is (2 * channel bandwidth). - */ - public double getChannelSampleRate() - { - return mChannelCalculator.getChannelSampleRate(); - } - /** * Current channel bandwidth/spacing. */ @@ -439,9 +428,10 @@ public class PolyphaseChannelManager implements ISourceEventProcessor } break; case REQUEST_STOP_SAMPLE_STREAM: - if(sourceEvent.hasSource() && sourceEvent.getSource() instanceof PolyphaseChannelSource) + if(sourceEvent.hasSource() && sourceEvent.getSource() instanceof PolyphaseChannelSource channelSource) { - stopChannelSource((PolyphaseChannelSource)sourceEvent.getSource()); + stopChannelSource(channelSource); + channelSource.dispose(); } else { @@ -449,14 +439,6 @@ public class PolyphaseChannelManager implements ISourceEventProcessor (sourceEvent.hasSource() ? sourceEvent.getSource().getClass() : "null source")); } break; - case REQUEST_SOURCE_DISPOSE: - Source source = sourceEvent.getSource(); - - if(source instanceof PolyphaseChannelSource) - { - source.dispose(); - } - break; case NOTIFICATION_MEASURED_FREQUENCY_ERROR_SYNC_LOCKED: //Rebroadcast so that the tuner source can process this event mSourceEventBroadcaster.broadcast(sourceEvent); diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java index 3ba96a25..e3ce9689 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java @@ -25,7 +25,6 @@ import io.github.dsheirer.dsp.filter.design.FilterDesignException; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.SourceEvent; -import io.github.dsheirer.source.tuner.channel.StreamProcessorWithHeartbeat; import io.github.dsheirer.source.tuner.channel.TunerChannel; import io.github.dsheirer.source.tuner.channel.TunerChannelSource; import java.util.ArrayList; @@ -42,7 +41,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { private Logger mLog = LoggerFactory.getLogger(PolyphaseChannelSource.class); private IPolyphaseChannelOutputProcessor mPolyphaseChannelOutputProcessor; - private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; + private Listener mSamplesListener; private double mChannelSampleRate; private long mIndexCenterFrequency; private List mOutputProcessorIndexes = new ArrayList<>(); @@ -64,7 +63,6 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { super(producerSourceEventListener, tunerChannel); mChannelSampleRate = channelCalculator.getChannelSampleRate(); - mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS); doUpdateOutputProcessor(channelCalculator, filterManager); } @@ -108,8 +106,6 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { super.start(); - mStreamHeartbeatProcessor.start(); - if(mPolyphaseChannelOutputProcessor != null) { mPolyphaseChannelOutputProcessor.start(); @@ -125,8 +121,6 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { mPolyphaseChannelOutputProcessor.stop(); } - - mStreamHeartbeatProcessor.stop(); } /** @@ -135,14 +129,24 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen @Override public void setListener(final Listener listener) { - mStreamHeartbeatProcessor.setListener(listener); + mSamplesListener = listener; mPolyphaseChannelOutputProcessor.setListener(this); } @Override public void receive(ComplexSamples complexSamples) { - mStreamHeartbeatProcessor.receive(complexSamples); + if(mSamplesListener != null) + { + try + { + mSamplesListener.receive(complexSamples); + } + catch(Throwable t) + { + mLog.error("Error dispatching complex samples to listener [" + mSamplesListener.getClass() + "]", t); + } + } } /** @@ -218,7 +222,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { case 1: mPolyphaseChannelOutputProcessor = new OneChannelOutputProcessor(channelCalculator.getChannelSampleRate(), - indexes, channelCalculator.getChannelCount()); + indexes, channelCalculator.getChannelCount(), getHeartbeatManager()); mPolyphaseChannelOutputProcessor.setListener(this); mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); mPolyphaseChannelOutputProcessor.start(); @@ -229,7 +233,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen float[] filter = filterManager.getFilter(channelCalculator.getChannelSampleRate(), channelCalculator.getChannelBandwidth(), 2); mPolyphaseChannelOutputProcessor = new TwoChannelOutputProcessor(channelCalculator.getChannelSampleRate(), - indexes, filter, channelCalculator.getChannelCount()); + indexes, filter, channelCalculator.getChannelCount(), getHeartbeatManager()); mPolyphaseChannelOutputProcessor.setListener(this); mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); mPolyphaseChannelOutputProcessor.start(); diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java index df5a936e..e75cbf45 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -20,8 +20,8 @@ package io.github.dsheirer.dsp.filter.channelizer.output; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.ComplexSamples; +import io.github.dsheirer.source.heartbeat.HeartbeatManager; import io.github.dsheirer.util.Dispatcher; -import java.util.Collections; import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,6 +31,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP private final static Logger mLog = LoggerFactory.getLogger(ChannelOutputProcessor.class); private Dispatcher> mChannelResultsDispatcher; + private HeartbeatManager mHeartbeatManager; protected Listener mComplexSamplesListener; private int mInputChannelCount; private long mCurrentSampleTimestamp = System.currentTimeMillis(); @@ -42,12 +43,25 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP * @param inputChannelCount is the number of input channels for this output processor * @param sampleRate of the output channel. This is used to match the oscillator's sample rate to the output * channel sample rate for frequency translation/correction. + * @param heartbeatManager to receive pings on the dispatcher thread */ - public ChannelOutputProcessor(int inputChannelCount, double sampleRate) + public ChannelOutputProcessor(int inputChannelCount, double sampleRate, HeartbeatManager heartbeatManager) { mInputChannelCount = inputChannelCount; - mChannelResultsDispatcher = new Dispatcher<>((int)sampleRate, "sdrtrunk polyphase channel", Collections.emptyList()); - mChannelResultsDispatcher.setListener(floats -> process(floats)); + //Process 1/10th of the sample rate per second at a rate of 20 times a second (200% of anticipated rate) + mHeartbeatManager = heartbeatManager; + mChannelResultsDispatcher = new Dispatcher("sdrtrunk polyphase channel", (int)(sampleRate / 10), + 50, mHeartbeatManager); + mChannelResultsDispatcher.setListener(floats -> { + try + { + process(floats); + } + catch(Throwable t) + { + mLog.error("Error processing channel results", t); + } + }); } /** diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java index 9eda378a..cf66c6e5 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java @@ -19,6 +19,7 @@ package io.github.dsheirer.dsp.filter.channelizer.output; import io.github.dsheirer.sample.complex.ComplexSamples; +import io.github.dsheirer.source.heartbeat.HeartbeatManager; import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,10 +37,12 @@ public class OneChannelOutputProcessor extends ChannelOutputProcessor * @param sampleRate of the output sample stream. * @param channelIndexes containing a single channel index. * @param gain value to apply. This is typically the same as the channelizer's channel count. + * @param heartbeatManager to receive heartbeats on the dispatch thread */ - public OneChannelOutputProcessor(double sampleRate, List channelIndexes, float gain) + public OneChannelOutputProcessor(double sampleRate, List channelIndexes, float gain, + HeartbeatManager heartbeatManager) { - super(1, sampleRate); + super(1, sampleRate, heartbeatManager); setPolyphaseChannelIndices(channelIndexes); mMixerAssembler = new OneChannelMixerAssembler(gain); mMixerAssembler.getMixer().setSampleRate(sampleRate); diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java index b1ed6cda..7c57241b 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java @@ -19,6 +19,7 @@ package io.github.dsheirer.dsp.filter.channelizer.output; import io.github.dsheirer.sample.complex.ComplexSamples; +import io.github.dsheirer.source.heartbeat.HeartbeatManager; import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,12 +38,14 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor * @param sampleRate of the output sample stream. * @param channelIndexes containing two channel indices. * @param gain to apply to output. Typically this is equal to the channelizer's channel count. + * @param heartbeatManager to be pinged on the dispatcher thread */ - public TwoChannelOutputProcessor(double sampleRate, List channelIndexes, float[] filter, float gain) + public TwoChannelOutputProcessor(double sampleRate, List channelIndexes, float[] filter, float gain, + HeartbeatManager heartbeatManager) { //Set the frequency correction oscillator to 2 x output sample rate since we'll be correcting the frequency //after synthesizing both input channels - super(2, sampleRate); + super(2, sampleRate, heartbeatManager); setPolyphaseChannelIndices(channelIndexes); mMixerAssembler = new TwoChannelMixerAssembler(gain); mMixerAssembler.getMixer().setSampleRate(sampleRate); diff --git a/src/main/java/io/github/dsheirer/gui/preference/playback/PlaybackPreferenceEditor.java b/src/main/java/io/github/dsheirer/gui/preference/playback/PlaybackPreferenceEditor.java index c13051d1..273c957b 100644 --- a/src/main/java/io/github/dsheirer/gui/preference/playback/PlaybackPreferenceEditor.java +++ b/src/main/java/io/github/dsheirer/gui/preference/playback/PlaybackPreferenceEditor.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -24,10 +24,9 @@ import io.github.dsheirer.preference.UserPreferences; import io.github.dsheirer.preference.playback.PlaybackPreference; import io.github.dsheirer.source.mixer.MixerChannelConfiguration; import io.github.dsheirer.source.mixer.MixerManager; -import javafx.beans.value.ChangeListener; -import javafx.beans.value.ObservableValue; -import javafx.event.ActionEvent; -import javafx.event.EventHandler; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.ShortBuffer; import javafx.geometry.HPos; import javafx.geometry.Insets; import javafx.geometry.Orientation; @@ -49,9 +48,6 @@ import org.slf4j.LoggerFactory; import javax.sound.sampled.AudioSystem; import javax.sound.sampled.Clip; import javax.sound.sampled.DataLine; -import java.nio.ByteBuffer; -import java.nio.ByteOrder; -import java.nio.ShortBuffer; /** @@ -60,20 +56,16 @@ import java.nio.ShortBuffer; public class PlaybackPreferenceEditor extends HBox { private final static Logger mLog = LoggerFactory.getLogger(PlaybackPreferenceEditor.class); - private PlaybackPreference mPlaybackPreference; + private final PlaybackPreference mPlaybackPreference; private GridPane mEditorPane; private ComboBox mMixerComboBox; private Button mMixerTestButton; private ToggleSwitch mUseAudioSegmentStartToneSwitch; private Button mTestStartToneButton; - private ToggleSwitch mUseAudioSegmentPreemptToneSwitch; - private Button mTestPreemptToneButton; private ToggleSwitch mUseAudioSegmentDropToneSwitch; private Button mTestDropToneButton; private ComboBox mStartToneFrequencyComboBox; private ComboBox mStartToneVolumeComboBox; - private ComboBox mPreemptToneFrequencyComboBox; - private ComboBox mPreemptToneVolumeComboBox; private ComboBox mDropToneFrequencyComboBox; private ComboBox mDropToneVolumeComboBox; @@ -114,18 +106,6 @@ public class PlaybackPreferenceEditor extends HBox mEditorPane.add(getStartToneVolumeComboBox(), 4, row); mEditorPane.add(getTestStartToneButton(), 5, row); - mEditorPane.add(getUseAudioSegmentPreemptToneSwitch(), 0, ++row); - mEditorPane.add(new Label("Priority Preempt Tone"), 1, row, 3, 1); - Label preemptFrequencyLabel = new Label("Frequency:"); - GridPane.setHalignment(preemptFrequencyLabel, HPos.RIGHT); - mEditorPane.add(preemptFrequencyLabel, 1, ++row); - mEditorPane.add(getPreemptToneFrequencyComboBox(), 2, row); - Label preemptVolumeLabel = new Label("Volume:"); - GridPane.setHalignment(preemptVolumeLabel, HPos.RIGHT); - mEditorPane.add(preemptVolumeLabel, 3, row); - mEditorPane.add(getPreemptToneVolumeComboBox(), 4, row); - mEditorPane.add(getTestPreemptToneButton(), 5, row); - mEditorPane.add(getUseAudioSegmentDropToneSwitch(), 0, ++row); mEditorPane.add(new Label("Drop Tone - Do Not Monitor"), 1, row, 3, 1); Label dropFrequencyLabel = new Label("Frequency:"); @@ -177,52 +157,21 @@ public class PlaybackPreferenceEditor extends HBox mUseAudioSegmentStartToneSwitch = new ToggleSwitch(); mUseAudioSegmentStartToneSwitch.setAlignment(Pos.BASELINE_RIGHT); mUseAudioSegmentStartToneSwitch.setSelected(mPlaybackPreference.getUseAudioSegmentStartTone()); - mUseAudioSegmentStartToneSwitch.selectedProperty().addListener(new ChangeListener() - { - @Override - public void changed(ObservableValue observable, Boolean oldValue, Boolean newValue) - { - mPlaybackPreference.setUseAudioSegmentStartTone(newValue); - } - }); + mUseAudioSegmentStartToneSwitch.selectedProperty().addListener((observable, oldValue, newValue) -> + mPlaybackPreference.setUseAudioSegmentStartTone(newValue)); } return mUseAudioSegmentStartToneSwitch; } - private ToggleSwitch getUseAudioSegmentPreemptToneSwitch() - { - if(mUseAudioSegmentPreemptToneSwitch == null) - { - mUseAudioSegmentPreemptToneSwitch = new ToggleSwitch(); - mUseAudioSegmentPreemptToneSwitch.setSelected(mPlaybackPreference.getUseAudioSegmentPreemptTone()); - mUseAudioSegmentPreemptToneSwitch.selectedProperty().addListener(new ChangeListener() - { - @Override - public void changed(ObservableValue observable, Boolean oldValue, Boolean newValue) - { - mPlaybackPreference.setUseAudioSegmentPreemptTone(newValue); - } - }); - } - - return mUseAudioSegmentPreemptToneSwitch; - } - private ToggleSwitch getUseAudioSegmentDropToneSwitch() { if(mUseAudioSegmentDropToneSwitch == null) { mUseAudioSegmentDropToneSwitch = new ToggleSwitch(); mUseAudioSegmentDropToneSwitch.setSelected(mPlaybackPreference.getUseAudioSegmentDropTone()); - mUseAudioSegmentDropToneSwitch.selectedProperty().addListener(new ChangeListener() - { - @Override - public void changed(ObservableValue observable, Boolean oldValue, Boolean newValue) - { - mPlaybackPreference.setUseAudioSegmentDropTone(newValue); - } - }); + mUseAudioSegmentDropToneSwitch.selectedProperty().addListener((observable, oldValue, newValue) -> + mPlaybackPreference.setUseAudioSegmentDropTone(newValue)); } return mUseAudioSegmentDropToneSwitch; @@ -236,14 +185,7 @@ public class PlaybackPreferenceEditor extends HBox IconNode iconNode = new IconNode(FontAwesome.PLAY); iconNode.setFill(Color.CORNFLOWERBLUE); mTestStartToneButton.setGraphic(iconNode); - mTestStartToneButton.setOnAction(new EventHandler() - { - @Override - public void handle(ActionEvent event) - { - play(mPlaybackPreference.getStartTone()); - } - }); + mTestStartToneButton.setOnAction(event -> play(mPlaybackPreference.getStartTone())); mTestStartToneButton.disableProperty().bind(getUseAudioSegmentStartToneSwitch().selectedProperty().not()); } @@ -251,29 +193,6 @@ public class PlaybackPreferenceEditor extends HBox return mTestStartToneButton; } - public Button getTestPreemptToneButton() - { - if(mTestPreemptToneButton == null) - { - mTestPreemptToneButton = new Button("Test"); - IconNode iconNode = new IconNode(FontAwesome.PLAY); - iconNode.setFill(Color.CORNFLOWERBLUE); - mTestPreemptToneButton.setGraphic(iconNode); - mTestPreemptToneButton.setOnAction(new EventHandler() - { - @Override - public void handle(ActionEvent event) - { - play(mPlaybackPreference.getPreemptTone()); - } - }); - - mTestPreemptToneButton.disableProperty().bind(getUseAudioSegmentPreemptToneSwitch().selectedProperty().not()); - } - - return mTestPreemptToneButton; - } - public Button getTestDropToneButton() { if(mTestDropToneButton == null) @@ -282,14 +201,7 @@ public class PlaybackPreferenceEditor extends HBox IconNode iconNode = new IconNode(FontAwesome.PLAY); iconNode.setFill(Color.CORNFLOWERBLUE); mTestDropToneButton.setGraphic(iconNode); - mTestDropToneButton.setOnAction(new EventHandler() - { - @Override - public void handle(ActionEvent event) - { - play(mPlaybackPreference.getDropTone()); - } - }); + mTestDropToneButton.setOnAction(event -> play(mPlaybackPreference.getDropTone())); mTestDropToneButton.disableProperty().bind(getUseAudioSegmentDropToneSwitch().selectedProperty().not()); } @@ -357,36 +269,6 @@ public class PlaybackPreferenceEditor extends HBox return mStartToneVolumeComboBox; } - public ComboBox getPreemptToneFrequencyComboBox() - { - if(mPreemptToneFrequencyComboBox == null) - { - mPreemptToneFrequencyComboBox = new ComboBox<>(); - mPreemptToneFrequencyComboBox.getItems().addAll(ToneFrequency.values()); - mPreemptToneFrequencyComboBox.getSelectionModel().select(mPlaybackPreference.getPreemptToneFrequency()); - mPreemptToneFrequencyComboBox.getSelectionModel().selectedItemProperty() - .addListener((observable, oldValue, newValue) -> mPlaybackPreference.setPreemptToneFrequency(newValue)); - mPreemptToneFrequencyComboBox.disableProperty().bind(getUseAudioSegmentPreemptToneSwitch().selectedProperty().not()); - } - - return mPreemptToneFrequencyComboBox; - } - - public ComboBox getPreemptToneVolumeComboBox() - { - if(mPreemptToneVolumeComboBox == null) - { - mPreemptToneVolumeComboBox = new ComboBox<>(); - mPreemptToneVolumeComboBox.getItems().addAll(ToneVolume.values()); - mPreemptToneVolumeComboBox.getSelectionModel().select(mPlaybackPreference.getPreemptToneVolume()); - mPreemptToneVolumeComboBox.getSelectionModel().selectedItemProperty() - .addListener((observable, oldValue, newValue) -> mPlaybackPreference.setPreemptToneVolume(newValue)); - mPreemptToneVolumeComboBox.disableProperty().bind(getUseAudioSegmentPreemptToneSwitch().selectedProperty().not()); - } - - return mPreemptToneVolumeComboBox; - } - /** * Plays the audio buffer over the default mono playback device * @param audioSamples with 8 kHz mono PCM samples diff --git a/src/main/java/io/github/dsheirer/module/decode/event/DecodeEventModel.java b/src/main/java/io/github/dsheirer/module/decode/event/DecodeEventModel.java index e4e96619..697391dc 100644 --- a/src/main/java/io/github/dsheirer/module/decode/event/DecodeEventModel.java +++ b/src/main/java/io/github/dsheirer/module/decode/event/DecodeEventModel.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -51,7 +51,7 @@ public class DecodeEventModel extends AbstractTableModel implements Listener mEvents = new ArrayList<>(); protected FilterSet mEventFilterSet = new DecodeEventFilterSet(); diff --git a/src/main/java/io/github/dsheirer/module/decode/event/MessageActivityModel.java b/src/main/java/io/github/dsheirer/module/decode/event/MessageActivityModel.java index 9f89b827..615b3a45 100644 --- a/src/main/java/io/github/dsheirer/module/decode/event/MessageActivityModel.java +++ b/src/main/java/io/github/dsheirer/module/decode/event/MessageActivityModel.java @@ -1,23 +1,20 @@ /* + * ***************************************************************************** + * Copyright (C) 2014-2023 Dennis Sheirer * - * * ****************************************************************************** - * * Copyright (C) 2014-2019 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 - * * ***************************************************************************** + * 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 io.github.dsheirer.module.decode.event; @@ -26,13 +23,13 @@ import io.github.dsheirer.filter.FilterSet; import io.github.dsheirer.message.IMessage; import io.github.dsheirer.message.StuffBitsMessage; import io.github.dsheirer.sample.Listener; - -import javax.swing.table.AbstractTableModel; import java.awt.EventQueue; import java.text.SimpleDateFormat; import java.util.LinkedList; import java.util.List; +import javax.swing.table.AbstractTableModel; + public class MessageActivityModel extends AbstractTableModel implements Listener { private static final long serialVersionUID = 1L; @@ -41,7 +38,7 @@ public class MessageActivityModel extends AbstractTableModel implements Listener private static final int TIMESLOT = 2; private static final int MESSAGE = 3; - protected int mMaxMessages = 500; + protected int mMaxMessages = 200; protected LinkedList mMessageItems = new LinkedList<>(); protected int[] mColumnWidths = {20, 20, 500}; protected String[] mHeaders = new String[]{"Time", "Protocol", "Timeslot", "Message"}; diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/audio/P25P1CallSequenceRecorder.java b/src/main/java/io/github/dsheirer/module/decode/p25/audio/P25P1CallSequenceRecorder.java index 1eda09d0..7584cdf2 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/audio/P25P1CallSequenceRecorder.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/audio/P25P1CallSequenceRecorder.java @@ -1,23 +1,20 @@ /* + * ***************************************************************************** + * Copyright (C) 2014-2023 Dennis Sheirer * - * * ****************************************************************************** - * * Copyright (C) 2014-2019 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 - * * ***************************************************************************** + * 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 io.github.dsheirer.module.decode.p25.audio; @@ -42,11 +39,10 @@ import io.github.dsheirer.module.decode.p25.phase1.message.ldu.LDUMessage; import io.github.dsheirer.module.decode.p25.phase1.message.tdu.TDULinkControlMessage; import io.github.dsheirer.module.decode.p25.phase1.message.tdu.TDUMessage; import io.github.dsheirer.preference.UserPreferences; +import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; - /** * P25 Phase 1 IMBE Frame recorder generates P25 call sequence recordings containing JSON representations of audio * frames, optional encryption and call identifiers. @@ -55,7 +51,7 @@ public class P25P1CallSequenceRecorder extends MBECallSequenceRecorder { private final static Logger mLog = LoggerFactory.getLogger(P25P1CallSequenceRecorder.class); - private static final String PROTOCOL = "APCO25-PHASE1"; + public static final String PROTOCOL = "APCO25-PHASE1"; private MBECallSequence mCallSequence; diff --git a/src/main/java/io/github/dsheirer/preference/TimestampFormat.java b/src/main/java/io/github/dsheirer/preference/TimestampFormat.java index 5950594b..e91c8ab0 100644 --- a/src/main/java/io/github/dsheirer/preference/TimestampFormat.java +++ b/src/main/java/io/github/dsheirer/preference/TimestampFormat.java @@ -1,7 +1,6 @@ /* - * ****************************************************************************** - * sdrtrunk - * Copyright (C) 2014-2018 Dennis Sheirer + * ***************************************************************************** + * Copyright (C) 2014-2023 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 @@ -15,7 +14,7 @@ * * You should have received a copy of the GNU General Public License * along with this program. If not, see - * ***************************************************************************** + * **************************************************************************** */ package io.github.dsheirer.preference; @@ -30,15 +29,14 @@ public enum TimestampFormat TIMESTAMP_DEFAULT("yyyy-MM-dd HH:mm:ss", "Timestamp: yyyy-MM-dd HH:mm:ss"), TIMESTAMP_COLONS("yyyy:MM:dd:HH:mm:ss", "Timestamp Colons: yyyy:MM:dd:HH:mm:ss"), TIMESTAMP_MILLIS("yyyy:MM:dd:HH:mm:ss.SSS", "Timestamp Milliseconds: yyyy:MM:dd:HH:mm:ss.SSS"), - TIMESTAMP_COMPACT("yyyyMMddHHmmss", "Timestamp Compact: yyyyMMddHHmmss"), + TIMESTAMP_COMPACT("yyyyMMdd_HHmmss", "Timestamp Compact: yyyyMMdd_HHmmss"), DATE_DASHES("yyyy-MM-dd", "Date Dashes: YYYY-MM-DD"), DATE_COLONS("yyyy:MM:dd", "Date Colons: YYYY:MM:DD"), DATE_COMPACT("yyyyMMdd", "Date Compact: YYYYMMDD"), TIME_COMPACT("HHmmss", "Time Compact: HHmmss"), TIME_COLONS("HH:mm:ss", "Time Colons: HH:mm:ss"), TIME_MILLIS("HH:mm:ss.SSS", "Time Milliseconds: HH:mm:ss.SSS"), - TIME_SECONDS("ss.SSS", "Time Seconds: ss.SSS"), - ; + TIME_SECONDS("ss.SSS", "Time Seconds: ss.SSS"); private String mFormat; private String mLabel; diff --git a/src/main/java/io/github/dsheirer/preference/playback/PlaybackPreference.java b/src/main/java/io/github/dsheirer/preference/playback/PlaybackPreference.java index aceeae12..b1457e45 100644 --- a/src/main/java/io/github/dsheirer/preference/playback/PlaybackPreference.java +++ b/src/main/java/io/github/dsheirer/preference/playback/PlaybackPreference.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2020 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -25,14 +25,12 @@ import io.github.dsheirer.gui.preference.playback.ToneVolume; import io.github.dsheirer.preference.Preference; import io.github.dsheirer.preference.PreferenceType; import io.github.dsheirer.sample.Listener; -import io.github.dsheirer.source.mixer.MixerChannel; import io.github.dsheirer.source.mixer.MixerChannelConfiguration; import io.github.dsheirer.source.mixer.MixerManager; +import java.util.prefs.Preferences; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.prefs.Preferences; - /** * User preferences for audio playback */ @@ -42,10 +40,6 @@ public class PlaybackPreference extends Preference private static final String PREFERENCE_KEY_DROP_TONE_FREQUENCY = "audio.playback.segment.drop.frequency"; private static final String PREFERENCE_KEY_DROP_TONE_VOLUME = "audio.playback.segment.drop.volume"; - private static final String PREFERENCE_KEY_USE_AUDIO_SEGMENT_PREEMPT_TONE = "audio.playback.segment.preempt.tone"; - private static final String PREFERENCE_KEY_PREEMPT_TONE_FREQUENCY = "audio.playback.segment.preempt.frequency"; - private static final String PREFERENCE_KEY_PREEMPT_TONE_VOLUME = "audio.playback.segment.preempt.volume"; - private static final String PREFERENCE_KEY_USE_AUDIO_SEGMENT_START_TONE = "audio.playback.segment.start.tone"; private static final String PREFERENCE_KEY_START_TONE_FREQUENCY = "audio.playback.segment.start.frequency"; private static final String PREFERENCE_KEY_START_TONE_VOLUME = "audio.playback.segment.start.volume"; @@ -56,12 +50,9 @@ public class PlaybackPreference extends Preference private final static Logger mLog = LoggerFactory.getLogger(PlaybackPreference.class); private Preferences mPreferences = Preferences.userNodeForPackage(PlaybackPreference.class); private Boolean mUseAudioSegmentStartTone; - private Boolean mUseAudioSegmentPreemptTone; private Boolean mUseAudioSegmentDropTone; private ToneFrequency mStartToneFrequency; private ToneVolume mStartToneVolume; - private ToneFrequency mPreemptToneFrequency; - private ToneVolume mPreemptToneVolume; private ToneFrequency mDropToneFrequency; private ToneVolume mDropToneVolume; private MixerChannelConfiguration mMixerChannelConfiguration; @@ -127,29 +118,6 @@ public class PlaybackPreference extends Preference notifyPreferenceUpdated(); } - /** - * Indicates if an audio segment preempt tone should be used. - */ - public boolean getUseAudioSegmentPreemptTone() - { - if(mUseAudioSegmentPreemptTone == null) - { - mUseAudioSegmentPreemptTone = mPreferences.getBoolean(PREFERENCE_KEY_USE_AUDIO_SEGMENT_PREEMPT_TONE, true); - } - - return mUseAudioSegmentPreemptTone; - } - - /** - * Sets the preference for using an audio segment preempt tone - */ - public void setUseAudioSegmentPreemptTone(boolean use) - { - mUseAudioSegmentPreemptTone = use; - mPreferences.putBoolean(PREFERENCE_KEY_USE_AUDIO_SEGMENT_PREEMPT_TONE, use); - notifyPreferenceUpdated(); - } - /** * Frequency for the drop tone */ @@ -198,30 +166,6 @@ public class PlaybackPreference extends Preference notifyPreferenceUpdated(); } - /** - * Frequency for the preempt tone - */ - public ToneFrequency getPreemptToneFrequency() - { - if(mPreemptToneFrequency == null) - { - int frequency = mPreferences.getInt(PREFERENCE_KEY_PREEMPT_TONE_FREQUENCY, ToneFrequency.F400.getValue()); - mPreemptToneFrequency = ToneFrequency.fromValue(frequency); - } - - return mPreemptToneFrequency; - } - - /** - * Sets the frequency for the preempt tone - */ - public void setPreemptToneFrequency(ToneFrequency toneFrequency) - { - mPreemptToneFrequency = toneFrequency; - mPreferences.putInt(PREFERENCE_KEY_PREEMPT_TONE_FREQUENCY, toneFrequency.getValue()); - notifyPreferenceUpdated(); - } - /** * Drop tone volume */ @@ -270,30 +214,6 @@ public class PlaybackPreference extends Preference notifyPreferenceUpdated(); } - /** - * Preempt tone volume - */ - public ToneVolume getPreemptToneVolume() - { - if(mPreemptToneVolume == null) - { - int volume = mPreferences.getInt(PREFERENCE_KEY_PREEMPT_TONE_VOLUME, ToneVolume.V5.getValue()); - mPreemptToneVolume = ToneVolume.fromValue(volume); - } - - return mPreemptToneVolume; - } - - /** - * Sets the preempt tone volume - */ - public void setPreemptToneVolume(ToneVolume toneVolume) - { - mPreemptToneVolume = toneVolume; - mPreferences.putInt(PREFERENCE_KEY_PREEMPT_TONE_VOLUME, toneVolume.getValue()); - notifyPreferenceUpdated(); - } - /** * Buffer with samples for the audio segment start tone */ @@ -307,19 +227,6 @@ public class PlaybackPreference extends Preference return null; } - /** - * Buffer with samples for the audio segment preempt tone - */ - public float[] getPreemptTone() - { - if(getUseAudioSegmentPreemptTone()) - { - return ToneUtil.getTone(getPreemptToneFrequency(), getPreemptToneVolume(), TONE_LENGTH_SAMPLES); - } - - return null; - } - /** * Buffer with samples for the audio segment drop tone */ diff --git a/src/main/java/io/github/dsheirer/record/RecorderFactory.java b/src/main/java/io/github/dsheirer/record/RecorderFactory.java index a546e86e..e91f4228 100644 --- a/src/main/java/io/github/dsheirer/record/RecorderFactory.java +++ b/src/main/java/io/github/dsheirer/record/RecorderFactory.java @@ -31,8 +31,8 @@ import io.github.dsheirer.record.wave.IRecordingStatusListener; import io.github.dsheirer.record.wave.NativeBufferWaveRecorder; import io.github.dsheirer.source.config.SourceConfigTuner; import io.github.dsheirer.source.config.SourceConfigTunerMultipleFrequency; -import io.github.dsheirer.source.config.SourceConfiguration; import io.github.dsheirer.util.StringUtils; +import io.github.dsheirer.util.TimeStamp; import java.io.File; import java.nio.file.Path; import java.util.ArrayList; @@ -57,47 +57,43 @@ public class RecorderFactory for(RecorderType recorderType: channel.getRecordConfiguration().getRecorders()) { + long frequency = getFrequency(channel); + switch(recorderType) { case BASEBAND: if(channel.isStandardChannel()) { - recorderModules.add(getBasebandRecorder(channel.toString(), userPreferences)); + recorderModules.add(getBasebandRecorder(channel.toString(), frequency, userPreferences)); } break; case DEMODULATED_BIT_STREAM: if(channel.isStandardChannel() && channel.getDecodeConfiguration().getDecoderType().providesBitstream()) { recorderModules.add(new BinaryRecorder(getRecordingBasePath(userPreferences), - getRecordingNameFromChannel(channel), channel.getDecodeConfiguration().getDecoderType().getProtocol())); + StringUtils.replaceIllegalCharacters(channel.toString()), + channel.getDecodeConfiguration().getDecoderType().getProtocol(), + frequency)); } break; case TRAFFIC_BASEBAND: if(channel.isTrafficChannel()) { - recorderModules.add(getBasebandRecorder(channel.toString(), userPreferences)); + recorderModules.add(getBasebandRecorder(channel.toString(), frequency, userPreferences)); } break; case TRAFFIC_DEMODULATED_BIT_STREAM: if(channel.isTrafficChannel() && channel.getDecodeConfiguration().getDecoderType().providesBitstream()) { recorderModules.add(new BinaryRecorder(getRecordingBasePath(userPreferences), - getRecordingNameFromChannel(channel), channel.getDecodeConfiguration().getDecoderType().getProtocol())); + StringUtils.replaceIllegalCharacters(channel.toString()), + channel.getDecodeConfiguration().getDecoderType().getProtocol(), + frequency)); } break; case MBE_CALL_SEQUENCE: if(channel.isStandardChannel() && channel.getSourceConfiguration() instanceof SourceConfigTuner) { - long frequency = 0; - - if(channel.getSourceConfiguration() instanceof SourceConfigTuner) - { - frequency = ((SourceConfigTuner)channel.getSourceConfiguration()).getFrequency(); - } - else if(channel.getSourceConfiguration() instanceof SourceConfigTunerMultipleFrequency) - { - frequency = ((SourceConfigTunerMultipleFrequency)channel.getSourceConfiguration()).getFrequencies().get(0); - } switch(channel.getDecodeConfiguration().getDecoderType()) { @@ -119,19 +115,12 @@ public class RecorderFactory case TRAFFIC_MBE_CALL_SEQUENCE: if(channel.isTrafficChannel() && channel.getSourceConfiguration() instanceof SourceConfigTuner) { - long frequency = 0; - - if(channel.getSourceConfiguration() instanceof SourceConfigTuner) - { - frequency = ((SourceConfigTuner)channel.getSourceConfiguration()).getFrequency(); - } - else if(channel.getSourceConfiguration() instanceof SourceConfigTunerMultipleFrequency) - { - frequency = ((SourceConfigTunerMultipleFrequency)channel.getSourceConfiguration()).getFrequencies().get(0); - } - switch(channel.getDecodeConfiguration().getDecoderType()) { + case DMR: + recorderModules.add(new DMRCallSequenceRecorder(userPreferences, frequency, + channel.getSystem(), channel.getSite())); + break; case P25_PHASE1: recorderModules.add(new P25P1CallSequenceRecorder(userPreferences, frequency, channel.getSystem(), channel.getSite())); @@ -150,24 +139,22 @@ public class RecorderFactory } /** - * Creates a recording file name from the channel frequency and the channel configuration. - * @param channel to process - * @return recording file name. + * Extract the frequency from the channel configuration + * @param channel with a source configuration + * @return frequency */ - private static String getRecordingNameFromChannel(Channel channel) + private static long getFrequency(Channel channel) { - if(channel.getSourceConfiguration() != null) + if(channel.getSourceConfiguration() instanceof SourceConfigTuner) { - SourceConfiguration sourceConfiguration = channel.getSourceConfiguration(); - - if(sourceConfiguration instanceof SourceConfigTuner sourceConfigTuner) - { - long frequency = sourceConfigTuner.getFrequency(); - return frequency + "_" + channel.toString(); - } + return ((SourceConfigTuner)channel.getSourceConfiguration()).getFrequency(); + } + else if(channel.getSourceConfiguration() instanceof SourceConfigTunerMultipleFrequency) + { + return ((SourceConfigTunerMultipleFrequency)channel.getSourceConfiguration()).getFrequencies().get(0); } - return StringUtils.replaceIllegalCharacters(channel.toString()); + return 0; } /** @@ -181,11 +168,16 @@ public class RecorderFactory /** * Constructs a baseband recorder for use in a processing chain. */ - public static ComplexSamplesWaveRecorder getBasebandRecorder(String channelName, UserPreferences userPreferences) + public static ComplexSamplesWaveRecorder getBasebandRecorder(String channelName, long frequency, UserPreferences userPreferences) { StringBuilder sb = new StringBuilder(); sb.append(getRecordingBasePath(userPreferences)); - sb.append(File.separator).append(StringUtils.replaceIllegalCharacters(channelName)).append("_baseband"); + sb.append(File.separator); + sb.append(TimeStamp.getTimeStamp("_")); + sb.append("_"); + sb.append(frequency); + sb.append("_"); + sb.append(StringUtils.replaceIllegalCharacters(channelName)).append("_baseband"); return new ComplexSamplesWaveRecorder(BASEBAND_SAMPLE_RATE, sb.toString()); } diff --git a/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java b/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java index 3a1c8489..e0dbf000 100644 --- a/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java +++ b/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -25,9 +25,6 @@ import io.github.dsheirer.sample.buffer.IByteBufferListener; import io.github.dsheirer.util.Dispatcher; import io.github.dsheirer.util.StringUtils; import io.github.dsheirer.util.TimeStamp; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; @@ -36,6 +33,8 @@ import java.nio.file.Path; import java.nio.file.StandardOpenOption; import java.util.EnumSet; import java.util.concurrent.atomic.AtomicBoolean; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Binary recorder module for recording demodulated bit/byte streams @@ -50,14 +49,14 @@ public class BinaryRecorder extends Module implements IByteBufferListener private final static Logger mLog = LoggerFactory.getLogger(BinaryRecorder.class); private static final int MAX_RECORDING_BYTE_SIZE = 524288; //500 kB - private Dispatcher mBufferProcessor = new Dispatcher<>(500, - "sdrtrunk binary recorder", ByteBuffer.allocate(0)); + private Dispatcher mBufferProcessor = new Dispatcher<>("sdrtrunk binary recorder", 200, 250); private AtomicBoolean mRunning = new AtomicBoolean(); private Path mBaseRecordingPath; private String mRecordingIdentifier; private BinaryWriter mBinaryWriter = new BinaryWriter(); private int mBytesRecordedCounter; private Protocol mProtocol; + private long mFrequency; /** * Constructs a binary recorder. @@ -65,13 +64,15 @@ public class BinaryRecorder extends Module implements IByteBufferListener * @param baseRecordingPath where the recording should be created * @param recordingIdentifier to include in the recording file name. * @param protocol to include as a values in the recording file name + * @param frequency in hertz */ - public BinaryRecorder(Path baseRecordingPath, String recordingIdentifier, Protocol protocol) + public BinaryRecorder(Path baseRecordingPath, String recordingIdentifier, Protocol protocol, long frequency) { mBaseRecordingPath = baseRecordingPath; mRecordingIdentifier = recordingIdentifier; mBufferProcessor.setListener(mBinaryWriter); mProtocol = protocol; + mFrequency = frequency; } public void start() @@ -96,7 +97,7 @@ public class BinaryRecorder extends Module implements IByteBufferListener { if(mBufferProcessor != null) { - mBufferProcessor.flushAndStop(); + mBufferProcessor.stop(); mBufferProcessor.setListener(null); try @@ -122,6 +123,8 @@ public class BinaryRecorder extends Module implements IByteBufferListener StringBuilder sb = new StringBuilder(); sb.append(TimeStamp.getTimeStamp("_")); sb.append("_"); + sb.append(mFrequency); + sb.append("_"); sb.append(mProtocol.getBitRate()).append("BPS_"); sb.append(mProtocol.getFileNameLabel()).append("_"); sb.append(mRecordingIdentifier.trim()); diff --git a/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java b/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java index c2e921e0..b3340442 100644 --- a/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java +++ b/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -27,7 +27,6 @@ import io.github.dsheirer.source.ISourceEventListener; import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.util.Dispatcher; import io.github.dsheirer.util.ThreadPool; -import io.github.dsheirer.util.TimeStamp; import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; @@ -45,9 +44,7 @@ public class ComplexSamplesWaveRecorder extends Module implements IComplexSample { private final static Logger mLog = LoggerFactory.getLogger(ComplexSamplesWaveRecorder.class); - private Dispatcher mBufferProcessor = new Dispatcher<>(500, - "sdrtrunk complex wave recorder", new ComplexSamples(new float[0], new float[0], System.currentTimeMillis())); - + private Dispatcher mBufferProcessor = new Dispatcher<>("sdrtrunk complex wave recorder", 100, 250); private AtomicBoolean mRunning = new AtomicBoolean(); private BufferWaveWriter mWriter; private String mFilePrefix; @@ -87,8 +84,6 @@ public class ComplexSamplesWaveRecorder extends Module implements IComplexSample { StringBuilder sb = new StringBuilder(); sb.append(mFilePrefix); - sb.append("_"); - sb.append(TimeStamp.getTimeStamp("_")); sb.append(".wav"); mFile = Paths.get(sb.toString()); diff --git a/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java b/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java index 997a4805..49e987f4 100644 --- a/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java +++ b/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -19,7 +19,6 @@ package io.github.dsheirer.record.wave; import io.github.dsheirer.buffer.INativeBuffer; -import io.github.dsheirer.buffer.NativeBufferPoisonPill; import io.github.dsheirer.module.Module; import io.github.dsheirer.sample.ConversionUtils; import io.github.dsheirer.sample.Listener; @@ -29,16 +28,16 @@ import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.util.Dispatcher; import io.github.dsheirer.util.ThreadPool; import io.github.dsheirer.util.TimeStamp; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.sound.sampled.AudioFormat; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.file.Path; import java.nio.file.Paths; import java.util.Iterator; import java.util.concurrent.atomic.AtomicBoolean; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.sound.sampled.AudioFormat; /** * WAVE audio recorder module for recording complex (I&Q) samples to a wave file @@ -48,8 +47,7 @@ public class NativeBufferWaveRecorder extends Module implements Listener mBufferProcessor = new Dispatcher<>(500, - "sdrtrunk native buffer wave recorder", new NativeBufferPoisonPill()); + private Dispatcher mBufferProcessor = new Dispatcher<>("sdrtrunk native buffer wave recorder", 100, 250); private AtomicBoolean mRunning = new AtomicBoolean(); private NativeBufferWaveWriter mWriter; diff --git a/src/main/java/io/github/dsheirer/record/wave/WaveWriter.java b/src/main/java/io/github/dsheirer/record/wave/WaveWriter.java index 573f992a..f05a69e0 100644 --- a/src/main/java/io/github/dsheirer/record/wave/WaveWriter.java +++ b/src/main/java/io/github/dsheirer/record/wave/WaveWriter.java @@ -1,7 +1,6 @@ /* - * ****************************************************************************** - * sdrtrunk - * Copyright (C) 2014-2019 Dennis Sheirer + * ***************************************************************************** + * Copyright (C) 2014-2023 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 @@ -15,15 +14,10 @@ * * You should have received a copy of the GNU General Public License * along with this program. If not, see - * ***************************************************************************** + * **************************************************************************** */ package io.github.dsheirer.record.wave; -import org.apache.commons.lang3.Validate; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.sound.sampled.AudioFormat; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.ByteOrder; @@ -35,6 +29,11 @@ import java.nio.file.StandardOpenOption; import java.util.Arrays; import java.util.regex.Matcher; import java.util.regex.Pattern; +import org.apache.commons.lang3.Validate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.sound.sampled.AudioFormat; public class WaveWriter implements AutoCloseable { @@ -115,12 +114,17 @@ public class WaveWriter implements AutoCloseable { int version = 2; - while(Files.exists(mFile)) + while(Files.exists(mFile) && version < 20) { mFile = Paths.get(mFile.toFile().getAbsolutePath().replace(".tmp", "_" + version + ".tmp")); version++; } + if(version >= 20) + { + throw new IOException("Unable to create a unique file name for recording - exceeded 20 versioning attempts"); + } + mFileChannel = (FileChannel.open(mFile, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW)); ByteBuffer header = getWaveHeader(mAudioFormat); diff --git a/src/main/java/io/github/dsheirer/source/SourceEvent.java b/src/main/java/io/github/dsheirer/source/SourceEvent.java index 2d6ed305..96c686b9 100644 --- a/src/main/java/io/github/dsheirer/source/SourceEvent.java +++ b/src/main/java/io/github/dsheirer/source/SourceEvent.java @@ -52,7 +52,6 @@ public class SourceEvent REQUEST_FREQUENCY_CHANGE, REQUEST_FREQUENCY_ROTATION, REQUEST_FREQUENCY_SELECTION, - REQUEST_SOURCE_DISPOSE, REQUEST_CHANGE_SQUELCH_THRESHOLD, REQUEST_CURRENT_SQUELCH_THRESHOLD, REQUEST_START_SAMPLE_STREAM, @@ -368,15 +367,6 @@ public class SourceEvent return new SourceEvent(Event.NOTIFICATION_STOP_SAMPLE_STREAM, source); } - /** - * Creates a request to dispose of the source. This is normally used for TunerChannelSource to notify the - * parent source manager that the channel is no longer needed - */ - public static SourceEvent sourceDisposeRequest(Source source) - { - return new SourceEvent(Event.REQUEST_SOURCE_DISPOSE, source); - } - /** * Creates a request to cycle to the next source frequency. This is normally used for decoders to request the * next frequency in a list when a multiple-frequency source configuration is defined. diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java index ab3e344e..2e60df46 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java @@ -19,7 +19,6 @@ package io.github.dsheirer.source.tuner.channel; import io.github.dsheirer.buffer.INativeBuffer; -import io.github.dsheirer.buffer.NativeBufferPoisonPill; import io.github.dsheirer.dsp.filter.decimate.DecimationFilterFactory; import io.github.dsheirer.dsp.filter.decimate.IRealDecimationFilter; import io.github.dsheirer.dsp.filter.design.FilterDesignException; @@ -49,9 +48,9 @@ public class HalfBandTunerChannelSource extends TunerCh private ComplexMixer mFrequencyCorrectionMixer; private IRealDecimationFilter mIDecimationFilter; private IRealDecimationFilter mQDecimationFilter; + private Listener mSamplesListener; private double mChannelSampleRate; private long mTunerFrequency; - private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; /** * Constructs a frequency translating and CIC decimating channel source. @@ -68,16 +67,15 @@ public class HalfBandTunerChannelSource extends TunerCh { super(producerSourceEventListener, tunerChannel); - mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS); - int desiredDecimation = (int)(sampleRate / channelSpecification.getMinimumSampleRate()); int decimation = DecimationFilterFactory.getDecimationRate(desiredDecimation); mIDecimationFilter = DecimationFilterFactory.getRealDecimationFilter(decimation); mQDecimationFilter = DecimationFilterFactory.getRealDecimationFilter(decimation); - mBufferDispatcher = new Dispatcher(BUFFER_MAX_CAPACITY, "sdrtrunk heterodyne channel " + - tunerChannel.getFrequency(), new NativeBufferPoisonPill()); + //Set dispatcher to process 1/10 of estimated sample arrival rate, 20 times per second (up to 200% per interval) + mBufferDispatcher = new Dispatcher("sdrtrunk heterodyne channel " + tunerChannel.getFrequency(), + (int)(sampleRate / 10), 50, getHeartbeatManager()); mBufferDispatcher.setListener(new NativeBufferProcessor()); //Setup the frequency mixer to the current source frequency @@ -91,7 +89,6 @@ public class HalfBandTunerChannelSource extends TunerCh public void start() { super.start(); - mStreamHeartbeatProcessor.start(); mBufferDispatcher.start(); } @@ -100,7 +97,6 @@ public class HalfBandTunerChannelSource extends TunerCh { super.stop(); mBufferDispatcher.stop(); - mStreamHeartbeatProcessor.stop(); } @Override @@ -167,7 +163,7 @@ public class HalfBandTunerChannelSource extends TunerCh @Override public void setListener(Listener listener) { - mStreamHeartbeatProcessor.setListener(listener); + mSamplesListener = listener; } @Override @@ -184,14 +180,29 @@ public class HalfBandTunerChannelSource extends TunerCh @Override public void receive(T nativeBuffer) { - Iterator iterator = nativeBuffer.iterator(); - - while(iterator.hasNext()) + if(mSamplesListener != null) { - ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next()); - float[] i = mIDecimationFilter.decimateReal(basebanded.i()); - float[] q = mQDecimationFilter.decimateReal(basebanded.q()); - mStreamHeartbeatProcessor.receive(new ComplexSamples(i, q, basebanded.timestamp())); + Iterator iterator = nativeBuffer.iterator(); + + while(iterator.hasNext()) + { + ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next()); + float[] i = mIDecimationFilter.decimateReal(basebanded.i()); + float[] q = mQDecimationFilter.decimateReal(basebanded.q()); + + try + { + mSamplesListener.receive(new ComplexSamples(i, q, basebanded.timestamp())); + } + catch(Throwable t) + { + //The listener can be made null and cause the error - only log if we have a non-null listener + if(mSamplesListener != null) + { + mLog.error("Error dispatching complex samples to listener [" + mSamplesListener + "]"); + } + } + } } } } diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java index 1491211a..0979df02 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java @@ -20,7 +20,6 @@ package io.github.dsheirer.source.tuner.channel; import io.github.dsheirer.buffer.INativeBuffer; -import io.github.dsheirer.buffer.NativeBufferPoisonPill; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.ISourceEventListener; @@ -41,7 +40,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou private final static Logger mLog = LoggerFactory.getLogger(PassThroughChannelSource.class); private TunerController mTunerController; private Dispatcher mBufferDispatcher; - private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; + private Listener mBufferListener; /** * Constructs an instance @@ -55,17 +54,15 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou { super(listener, tunerChannel); mTunerController = tunerController; - mBufferDispatcher = new Dispatcher<>(500, "sdrtrunk pass-through channel " + - tunerChannel.getFrequency(), new NativeBufferPoisonPill()); + mBufferDispatcher = new Dispatcher<>("sdrtrunk pass-through channel " + tunerChannel.getFrequency(), + 250, 50, getHeartbeatManager()); mBufferDispatcher.setListener(new BufferProcessor()); - mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS); } @Override public void start() { super.start(); - mStreamHeartbeatProcessor.stop(); mBufferDispatcher.start(); } @@ -74,7 +71,6 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou { super.stop(); mBufferDispatcher.stop(); - mStreamHeartbeatProcessor.stop(); } @Override @@ -92,7 +88,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou @Override public void setListener(Listener listener) { - mStreamHeartbeatProcessor.setListener(listener); + mBufferListener = listener; } @Override @@ -112,11 +108,21 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou @Override public void receive(INativeBuffer nativeBuffer) { - Iterator iterator = nativeBuffer.iterator(); - - while(iterator.hasNext()) + if(mBufferListener != null) { - mStreamHeartbeatProcessor.receive(iterator.next()); + Iterator iterator = nativeBuffer.iterator(); + + while(iterator.hasNext()) + { + try + { + mBufferListener.receive(iterator.next()); + } + catch(Throwable t) + { + mLog.error("Error dispatching complex sample buffers to listener [" + mBufferListener + "]", t); + } + } } } } diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java b/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java deleted file mode 100644 index 2f15836f..00000000 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * ***************************************************************************** - * Copyright (C) 2014-2023 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 io.github.dsheirer.source.tuner.channel; - -import io.github.dsheirer.sample.Listener; -import io.github.dsheirer.source.heartbeat.HeartbeatManager; -import io.github.dsheirer.util.ThreadPool; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Thread-safe implementation that coordinates the flow of data/buffer stream processing with a parallel, periodic - * heartbeat using an internal locking mechanism to coordinate the two activities. This class is intended to be used - * by TunerChannelSource implementations to coordinate the flow of sample buffers produced on a dispatch thread with a - * scheduled timer driven heartbeat mechanism, to ensure the downstream listener receives either samples or a - * heartbeat, but not both at the same time. - * - * Samples are processed as they arrive and are immediately followed by a heartbeat. A separate scheduled timer runs - * on a periodic schedule to optionally fire a heartbeat, independent of the sample stream rate. The scheduled - * heartbeat is optional and will skip a heartbeat interval if this processor is currently dispatching a sample buffer. - * Sample stream processing will block until a currently running scheduled heartbeat timer event completes, when these - * two events occur at the same time. - */ -public class StreamProcessorWithHeartbeat implements Listener -{ - private static final Logger mLog = LoggerFactory.getLogger(StreamProcessorWithHeartbeat.class); - private HeartbeatManager mHeartbeatManager; - private long mHeartbeatInterval; - private Listener mListener; - private ScheduledFuture mHeartbeatTimerFuture; - - /** - * Constructs an instance - * @param heartbeatManager that will be invoked by an internal scheduled timer - * @param heartbeatInterval in milliseconds - */ - public StreamProcessorWithHeartbeat(HeartbeatManager heartbeatManager, long heartbeatInterval) - { - mHeartbeatManager = heartbeatManager; - mHeartbeatInterval = heartbeatInterval; - } - - /** - * Starts the heartbeat timer - */ - public void start() - { - if(mHeartbeatTimerFuture == null) - { - mHeartbeatTimerFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(() -> fireOptionalHeartbeat(), - mHeartbeatInterval, mHeartbeatInterval, TimeUnit.MILLISECONDS); - } - } - - /** - * Stops the heartbeat timer - */ - public void stop() - { - if(mHeartbeatTimerFuture != null) - { - mHeartbeatTimerFuture.cancel(true); - mHeartbeatTimerFuture = null; - } - } - - /** - * Optionally fires a heartbeat if we're not currently processing a native buffer - */ - private void fireOptionalHeartbeat() - { - try - { - mHeartbeatManager.broadcast(); - } - catch(Exception e) - { - mLog.error("Error firing optional heartbeat", e); - } - } - - @Override - public void receive(T data) - { - try - { - mHeartbeatManager.broadcast(); - - Listener listener = mListener; - - if(listener != null) - { - listener.receive(data); - } - } - catch(Exception e) - { - mLog.error("Error distributing data to listeners following heartbeat", e); - } - } - - /** - * Registers the listener to receive data/buffers for the defined generic type that are distributed to this processor. - * @param listener to register or null to deregister an existing listener - */ - public void setListener(Listener listener) - { - mListener = listener; - } -} diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/TunerChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/TunerChannelSource.java index 242859ca..ee2fd464 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/TunerChannelSource.java +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/TunerChannelSource.java @@ -109,7 +109,6 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc public void stop() { broadcastProducerSourceEvent(SourceEvent.stopSampleStreamRequest(this)); - broadcastProducerSourceEvent(SourceEvent.sourceDisposeRequest(this)); } @Override diff --git a/src/main/java/io/github/dsheirer/source/tuner/manager/HeterodyneChannelSourceManager.java b/src/main/java/io/github/dsheirer/source/tuner/manager/HeterodyneChannelSourceManager.java index d6c0de8c..1f98427d 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/manager/HeterodyneChannelSourceManager.java +++ b/src/main/java/io/github/dsheirer/source/tuner/manager/HeterodyneChannelSourceManager.java @@ -291,6 +291,7 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager { switch(sourceEvent.getEvent()) { +//TODO: protect start/stop processing with a reentrant lock case REQUEST_START_SAMPLE_STREAM: if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource) { @@ -304,19 +305,13 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager } break; case REQUEST_STOP_SAMPLE_STREAM: - if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource) + if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource halfBandSource) { - mSampleDelayBuffer.removeListener((HalfBandTunerChannelSource)sourceEvent.getSource()); + mSampleDelayBuffer.removeListener(halfBandSource); stopDelayBuffer(); - } - break; - case REQUEST_SOURCE_DISPOSE: - if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource) - { - HalfBandTunerChannelSource channelSource = (HalfBandTunerChannelSource) sourceEvent.getSource(); - mChannelSources.remove(channelSource); - mTunerChannels.remove(channelSource.getTunerChannel()); - channelSource.dispose(); + mChannelSources.remove(halfBandSource); + mTunerChannels.remove(halfBandSource.getTunerChannel()); + halfBandSource.dispose(); //Unlock the tuner controller if there are no more channels if(getTunerChannelCount() == 0) diff --git a/src/main/java/io/github/dsheirer/source/tuner/rtl/r820t/R820TTunerEditor.java b/src/main/java/io/github/dsheirer/source/tuner/rtl/r820t/R820TTunerEditor.java index 9e608209..bc9aa49e 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/rtl/r820t/R820TTunerEditor.java +++ b/src/main/java/io/github/dsheirer/source/tuner/rtl/r820t/R820TTunerEditor.java @@ -123,7 +123,6 @@ public class R820TTunerEditor extends TunerEditor 0) { dispatchTransfer(transfer); - transfer.buffer().rewind(); } + transfer.buffer().rewind(); + if(mAutoResubmitTransfers) { submitTransfer(transfer); } break; - case LibUsb.TRANSFER_CANCELLED: - //Reset the transfer but don't do anything else since we're shutting down - transfer.buffer().rewind(); - break; default: //Unexpected transfer error - shutdown the tuner transfer.buffer().rewind(); diff --git a/src/main/java/io/github/dsheirer/util/Dispatcher.java b/src/main/java/io/github/dsheirer/util/Dispatcher.java index 7f0bb52e..428cbdc8 100644 --- a/src/main/java/io/github/dsheirer/util/Dispatcher.java +++ b/src/main/java/io/github/dsheirer/util/Dispatcher.java @@ -18,52 +18,70 @@ */ package io.github.dsheirer.util; +import io.github.dsheirer.controller.NamingThreadFactory; import io.github.dsheirer.sample.Listener; -import java.util.concurrent.LinkedBlockingQueue; +import io.github.dsheirer.source.heartbeat.HeartbeatManager; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Threaded processor for receiving elements from a separate producer thread and forwarding those buffers to a - * registered listener on this consumer/dispatcher thread. + * Threaded scheduled processor for receiving elements from a separate producer thread and forwarding those buffers to a + * registered listener on this consumer/dispatcher thread. Internally uses a single-thread thread pool to effect a + * timer-based interval for processing to avoid excessive context switching inherent in a blocking queue. Sizes the + * thread pool to a single thread to ensure Garbage Collector can efficiently clean objects created on the thread. */ public class Dispatcher implements Listener { private final static Logger mLog = LoggerFactory.getLogger(Dispatcher.class); private static final long OVERFLOW_LOG_EVENT_WAIT_PERIOD = TimeUnit.SECONDS.toMillis(10); - private LinkedBlockingQueue mQueue; + private LinkedTransferQueue mQueue = new LinkedTransferQueue<>(); private Listener mListener; private AtomicBoolean mRunning = new AtomicBoolean(); private String mThreadName; - private Thread mThread; - private E mPoisonPill; - private long mLastOverflowLogEvent; + private ScheduledExecutorService mExecutorService; + private ScheduledFuture mScheduledFuture; + private int mBatchSize; + private long mInterval; + private HeartbeatManager mHeartbeatManager; /** - * Constructs an instance - * @param maxSize of the internal queue + * Constructs an instance of a Dispatcher with integrated heartbeat support. * @param threadName to name the dispatcher thread - * @param poisonPill of type E, used to kill the thread + * @param batchSize is maximum number of elements to process from the queue per interval. This should be sized to + * the anticipated number of elements per second, divided by the number of intervals per second and increased by + * a factor of two. This serves to constrain how many elements are processed per interval when the queue starts + * to back up, so that the thread doesn't dominate CPU resources. + * @param interval for processing each batch in milliseconds. + * @param heartbeatManager to receive a heartbeat command at each processing interval. */ - public Dispatcher(int maxSize, String threadName, E poisonPill) + public Dispatcher(String threadName, int batchSize, long interval, HeartbeatManager heartbeatManager) { - if(poisonPill == null) - { - throw new IllegalArgumentException("Poison pill must be non-null"); - } - mQueue = new LinkedBlockingQueue<>(maxSize); - mThreadName = threadName; - mPoisonPill = poisonPill; + this(threadName, batchSize, interval); + mHeartbeatManager = heartbeatManager; } /** - * Listener to receive the queued buffers each time this processor runs. + * Constructs an instance + * @param threadName to name the dispatcher thread + * @param batchSize is maximum number of elements to process from the queue per interval. This should be sized to + * the anticipated number of elements per second, divided by the number of intervals per second and increased by + * a factor of two. This serves to constrain how many elements are processed per interval when the queue starts + * to back up, so that the thread doesn't dominate CPU resources. + * @param interval for processing each batch in milliseconds. */ - protected Listener getListener() + public Dispatcher(String threadName, int batchSize, long interval) { - return mListener; + mThreadName = threadName; + mBatchSize = batchSize; + mInterval = interval; } /** @@ -86,16 +104,7 @@ public class Dispatcher implements Listener { if(mRunning.get()) { - if(!mQueue.offer(e)) - { - if(System.currentTimeMillis() > (mLastOverflowLogEvent + OVERFLOW_LOG_EVENT_WAIT_PERIOD)) - { - mLastOverflowLogEvent = System.currentTimeMillis(); - mLog.warn("Dispatcher - temporary buffer overflow for thread [" + mThreadName + "] - throwing away samples - " + - " processor flag:" + (mRunning.get() ? "running" : "stopped") + - " thread:" + (mThread != null ? (mThread.isAlive() ? mThread.getState() : "dead") : "null" )); - } - } + mQueue.add(e); } } @@ -106,10 +115,22 @@ public class Dispatcher implements Listener { if(mRunning.compareAndSet(false, true)) { - mThread = new Thread(new Processor()); - mThread.setName(mThreadName); - mThread.setPriority(Thread.MAX_PRIORITY); - mThread.start(); + if(mScheduledFuture != null) + { + mScheduledFuture.cancel(true); + } + + if(mExecutorService != null) + { + mExecutorService.shutdown(); + mExecutorService = null; + } + + mQueue.clear(); + mExecutorService = Executors.newSingleThreadScheduledExecutor(new NamingThreadFactory(mThreadName)); + + Runnable r = (mHeartbeatManager != null ? new ProcessorWithHeartbeat() : new Processor()); + mScheduledFuture = mExecutorService.scheduleAtFixedRate(r, 0, mInterval, TimeUnit.MILLISECONDS); } } @@ -120,29 +141,21 @@ public class Dispatcher implements Listener { if(mRunning.compareAndSet(true, false)) { - mQueue.offer(mPoisonPill); - - try + if(mScheduledFuture != null) { - mThread.interrupt(); - mThread.join(); - mThread = null; + mScheduledFuture.cancel(true); + mScheduledFuture = null; + mQueue.clear(); } - catch(Exception e) + + if(mExecutorService != null) { - mLog.error("Timeout while waiting to join terminating buffer processor thread"); + mExecutorService.shutdown(); + mExecutorService = null; } } } - /** - * Finishes processing any queued buffers and then stops the processing thread. - */ - public void flushAndStop() - { - mQueue.offer(mPoisonPill); - } - /** * Indicates if this processor is currently running */ @@ -151,56 +164,75 @@ public class Dispatcher implements Listener return mRunning.get(); } + /** + * Processes elements from the queue. Note: this should only be invoked on the Processor thread. + */ + private void process() + { + List elements = new ArrayList<>(); + + mQueue.drainTo(elements, mBatchSize); + + for(E element: elements) + { + if(mRunning.get() && mListener != null) + { + try + { + mListener.receive(element); + } + catch(Throwable t) + { + mLog.error("Error while dispatching element [" + element.getClass() + "] to listener [" + + mListener.getClass() + "]", t); + } + } + } + } + /** * Processor to service the buffer queue and distribute the buffers to the registered listener */ class Processor implements Runnable { + private AtomicBoolean mRunning = new AtomicBoolean(); + @Override public void run() { - try + if(mRunning.compareAndSet(false, true)) { - mQueue.clear(); + process(); + mRunning.set(false); + } + } + } - E element; + /** + * Processor to service the buffer queue and distribute the buffers to the registered listener. Includes a + * support for commanding a heart beat with each processing interval. + */ + class ProcessorWithHeartbeat implements Runnable + { + private AtomicBoolean mRunning = new AtomicBoolean(); - while(mRunning.get()) + @Override + public void run() + { + if(mRunning.compareAndSet(false, true)) + { + process(); + + try { - try - { - element = mQueue.take(); - - if(mPoisonPill == element) - { - mRunning.set(false); - } - else if(element != null) - { - if(mListener == null) - { - throw new IllegalStateException("Listener for [" + mThreadName + "] is null"); - } - mListener.receive(element); - } - } - catch(InterruptedException e) - { - //Normal shutdown is by interrupt - mRunning.set(false); - } - catch(Exception e) - { - mLog.error("Error while processing element", e); - } + mHeartbeatManager.broadcast(); + } + catch(Throwable t) + { + mLog.error("Error broadcasting heartbeat during Dispatcher processing interval", t); } - //Shutting down - clear the queue - mQueue.clear(); - } - catch(Throwable t) - { - mLog.error("Unexpected error thrown from the Dispatcher thread", t); + mRunning.set(false); } } }