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 197f8513..77b23ec6 100644 --- a/src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java +++ b/src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java @@ -62,6 +62,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; /** * Channel processing manager handles all starting and stopping of channel decoding. A processing chain is created @@ -74,7 +76,9 @@ public class ChannelProcessingManager implements Listener private final static Logger mLog = LoggerFactory.getLogger(ChannelProcessingManager.class); private static final String TUNER_UNAVAILABLE_DESCRIPTION = "TUNER UNAVAILABLE"; private Map mProcessingChains = new ConcurrentHashMap<>(); + private Lock mLock = new ReentrantLock(); + private ChannelSourceEventErrorListener mSourceErrorListener = new ChannelSourceEventErrorListener(); private List> mAudioSegmentListeners = new CopyOnWriteArrayList<>(); private List> mDecodeEventListeners = new CopyOnWriteArrayList<>(); private Broadcaster mChannelEventBroadcaster = new Broadcaster(); @@ -170,9 +174,8 @@ public class ChannelProcessingManager implements Listener * * @param event that requests either enable/start or disable/stop a channel. */ - @Deprecated //TODO: convert all usage to start(ChannelStartProcessingRequest request) method @Override - public synchronized void receive(ChannelEvent event) + public void receive(ChannelEvent event) { Channel channel = event.getChannel(); @@ -350,38 +353,9 @@ public class ChannelProcessingManager implements Listener processingChain.addDecodeEventListener(listener); } - //Add a listener to detect source error state that indicates the channel should be shutdown - processingChain.addSourceEventListener(sourceEvent -> - { - if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE && sourceEvent.getSource() != null) - { - Channel toShutdown = null; - - for(Map.Entry entry: mProcessingChains.entrySet()) - { - if(entry.getValue().hasSource(sourceEvent.getSource())) - { - toShutdown = entry.getKey(); - break; - } - } - - if(toShutdown != null) - { - mLog.warn("Channel source error detected - stopping channel [" + toShutdown.getName() + "]"); - - try - { - stopProcessing(toShutdown); - } - catch(ChannelException ce) - { - mLog.error("Error stopping channel [" + channel.getName() + "] with source error - " + - ce.getMessage()); - } - } - } - }); + //Add a listener to detect source error state that indicates the channel should be shutdown. + //Note: processing chain will only add this once. + processingChain.addSourceEventListener(mSourceErrorListener); //Register this manager to receive channel events from traffic channel manager modules within //the processing chain @@ -459,18 +433,90 @@ public class ChannelProcessingManager implements Listener processingChain.getChannelState().updateChannelStateIdentifiers(notification); } } - } - mProcessingChains.put(channel, processingChain); - processingChain.start(); + if(addProcessingChain(channel, processingChain)) + { + processingChain.start(); - //This has to be done on the FX event thread when the playlist editor is constructed - Platform.runLater(() -> channel.setProcessing(true)); + //This has to be done on the FX event thread when the playlist editor is constructed + Platform.runLater(() -> channel.setProcessing(true)); - getChannelMetadataModel().add(new ChannelAndMetadata(channel, processingChain.getChannelState().getChannelMetadata())); + mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_START)); + } + else + { + mLog.warn("Channel [" + channel.getName() + "] processing chain not added because it already exists"); + processingChain.removeEventLoggingModules(); + processingChain.removeRecordingModules(); + processingChain.removeFrequencyChangeListener(channel); + channel.resetFrequencyCorrection(); + mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP)); + mChannelEventBroadcaster.removeListener(processingChain); + processingChain.getEventBus().unregister(ChannelProcessingManager.this); + processingChain.dispose(); + } + } - mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_START)); + /** + * Thread-safe add processing chain and add channel metadata to channel metadata model. + * @param channel for the processing chain + * @param processingChain to add + * @return true if processing chain was added or false if it was not added due to there already + * being a processing chain registered for that channel. + */ + private boolean addProcessingChain(Channel channel, ProcessingChain processingChain) + { + boolean added = false; + + mLock.lock(); + + try + { + if(!mProcessingChains.containsKey(channel)) + { + added = true; + mProcessingChains.put(channel, processingChain); + getChannelMetadataModel().add(new ChannelAndMetadata(channel, processingChain.getChannelState().getChannelMetadata())); + } + } + finally + { + mLock.unlock(); + } + + return added; + } + + /** + * Thread-safe remove processing chain. + * @param channel for identifying the processing chain + * @return the removed processing chain or null + */ + private ProcessingChain removeProcessingChain(Channel channel) + { + ProcessingChain removed = null; + + mLock.lock(); + + try + { + removed = mProcessingChains.remove(channel); + + if(removed != null) + { + for(ChannelMetadata channelMetadata: removed.getChannelState().getChannelMetadata()) + { + getChannelMetadataModel().remove(channelMetadata); + } + } + } + finally + { + mLock.unlock(); + } + + return removed; } /** @@ -483,30 +529,34 @@ public class ChannelProcessingManager implements Listener //This has to be done on the FX event thread when the playlist editor is constructed Platform.runLater(() -> channel.setProcessing(false)); - if(mProcessingChains.containsKey(channel)) + ProcessingChain processingChain = removeProcessingChain(channel); + + if(processingChain != null) { - ProcessingChain processingChain = mProcessingChains.remove(channel); + //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(); - for(ChannelMetadata channelMetadata: processingChain.getChannelState().getChannelMetadata()) - { - getChannelMetadataModel().remove(channelMetadata); - } + //Deregister channel from receive frequency correction events to show in the spectral display (hack!) + processingChain.removeFrequencyChangeListener(channel); + channel.resetFrequencyCorrection(); - processingChain.stop(); + mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP)); + mChannelEventBroadcaster.removeListener(processingChain); - processingChain.removeEventLoggingModules(); - processingChain.removeRecordingModules(); - - //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); - - //Unregister for event bus requests and notifications - processingChain.getEventBus().unregister(ChannelProcessingManager.this); - processingChain.dispose(); + //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); + } + }); } else { @@ -653,4 +703,52 @@ public class ChannelProcessingManager implements Listener } } } + + /** + * Monitors all channels for an error in the source event that would require the + * channel's processing chain to be stopped + */ + private class ChannelSourceEventErrorListener implements Listener + { + @Override public void receive(SourceEvent sourceEvent) + { + if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE && sourceEvent.getSource() != null) + { + Channel toShutdown = null; + + mLock.lock(); + + try + { + for(Map.Entry entry: mProcessingChains.entrySet()) + { + if(entry.getValue().hasSource(sourceEvent.getSource())) + { + toShutdown = entry.getKey(); + break; + } + } + } + finally + { + mLock.unlock(); + } + + if(toShutdown != null) + { + mLog.warn("Channel source error detected - stopping channel [" + toShutdown.getName() + "]"); + + try + { + stopProcessing(toShutdown); + } + catch(ChannelException ce) + { + mLog.error("Error stopping channel [" + (toShutdown != null ? toShutdown.getName() : "unknown") + + "] with source error - " + ce.getMessage()); + } + } + } + } + } } 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 8c3b77d1..a8e56f8a 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 @@ -23,25 +23,29 @@ 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.List; +/** + * Polyphase Channelizer's Tuner Channel Source implementation. Wraps a ChannelOutputProcessor instance and + * provides tuner channel source functionality. Supports dynamic swapout of the underlying channel output processor + * as the upstream polyphase channelizer reconfigures during tuned center frequency changes. + */ public class PolyphaseChannelSource extends TunerChannelSource implements Listener { private IPolyphaseChannelOutputProcessor mPolyphaseChannelOutputProcessor; private IPolyphaseChannelOutputProcessor mReplacementPolyphaseChannelOutputProcessor; + private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; private long mReplacementFrequency; private double mChannelSampleRate; private long mIndexCenterFrequency; private long mChannelFrequencyCorrection; - private Listener mComplexSamplesListener; /** - * Polyphase channelizer tuner channel source implementation. Adapts the channel array output samples from the - * polyphase channelizer into a single channel, or a channel synthesized from two adjacent channels that is - * frequency translated and decimated to a single channel. + * Constructs an instance * * @param tunerChannel describing the desired channel frequency and bandwidth/minimum sample rate * @param outputProcessor - to process polyphase channelizer channel results into a channel stream @@ -58,7 +62,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen mPolyphaseChannelOutputProcessor = outputProcessor; mPolyphaseChannelOutputProcessor.setListener(this); mChannelSampleRate = channelSampleRate; - + mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS); setFrequency(centerFrequency); } @@ -67,6 +71,8 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { super.start(); + mStreamHeartbeatProcessor.start(); + if(mPolyphaseChannelOutputProcessor != null) { mPolyphaseChannelOutputProcessor.start(); @@ -82,6 +88,8 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { mPolyphaseChannelOutputProcessor.stop(); } + + mStreamHeartbeatProcessor.stop(); } /** @@ -90,19 +98,14 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen @Override public void setListener(final Listener listener) { - mComplexSamplesListener = listener; + mStreamHeartbeatProcessor.setListener(listener); mPolyphaseChannelOutputProcessor.setListener(this); } @Override public void receive(ComplexSamples complexSamples) { - getHeartbeatManager().broadcast(); - - if(mComplexSamplesListener != null) - { - mComplexSamplesListener.receive(complexSamples); - } + mStreamHeartbeatProcessor.receive(complexSamples); } /** diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/CICTunerChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/CICTunerChannelSource.java deleted file mode 100644 index 3861b143..00000000 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/CICTunerChannelSource.java +++ /dev/null @@ -1,247 +0,0 @@ -/* - * ***************************************************************************** - * Copyright (C) 2014-2022 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.buffer.INativeBuffer; -import io.github.dsheirer.buffer.NativeBufferPoisonPill; -import io.github.dsheirer.dsp.filter.cic.PrimeCicDecimationFilter; -import io.github.dsheirer.dsp.filter.design.FilterDesignException; -import io.github.dsheirer.dsp.mixer.ComplexMixer; -import io.github.dsheirer.dsp.mixer.ComplexMixerFactory; -import io.github.dsheirer.sample.Listener; -import io.github.dsheirer.sample.complex.ComplexSamples; -import io.github.dsheirer.sample.complex.ComplexSamplesRepackager; -import io.github.dsheirer.source.SourceEvent; -import io.github.dsheirer.util.Dispatcher; - -import java.util.Iterator; -import java.util.List; - -/** - * Complex sample source that provides a frequency-translated and decimated sample buffer stream. Uses a - * CIC decimation filter that requires the decimation rate to be an integer multiple. Sample buffer processing - * occurs on a scheduled runnable thread. - */ -public class CICTunerChannelSource extends TunerChannelSource implements Listener -{ - //Maximum number of filled buffers for the blocking queue - private static final int BUFFER_MAX_CAPACITY = 300; - - //Threshold for resetting buffer overflow condition - private static final int BUFFER_OVERFLOW_RESET_THRESHOLD = 100; - - //Samples repackager is set to 2048 (a power of 2) to align with channel processing chain requirements. - private final ComplexSamplesRepackager mSamplesRepackager = new ComplexSamplesRepackager(1024); - private Dispatcher mBufferDispatcher; - private ComplexMixer mFrequencyCorrectionMixer; - private PrimeCicDecimationFilter mIDecimationFilter; - private PrimeCicDecimationFilter mQDecimationFilter; - private double mChannelSampleRate; - private long mChannelFrequencyCorrection = 0; - private long mTunerFrequency; - private Listener mListener; - - /** - * Constructs a frequency translating and CIC decimating channel source. - * - * @param producerSourceEventListener to receive sample stream start/stop requests - * @param tunerChannel that details the desired channel frequency and bandwidth - * @param sampleRate of the incoming sample stream - * @param channelSpecification for the requested channel. - * @throws FilterDesignException if a final cleanup filter cannot be designed using the remez filter - * designer and the filter parameters. - */ - public CICTunerChannelSource(Listener producerSourceEventListener, TunerChannel tunerChannel, - double sampleRate, ChannelSpecification channelSpecification) throws FilterDesignException - { - super(producerSourceEventListener, tunerChannel); - - int decimation = (int)(sampleRate / channelSpecification.getMinimumSampleRate()); - - mIDecimationFilter = new PrimeCicDecimationFilter(sampleRate, decimation, channelSpecification.getPassFrequency(), - channelSpecification.getStopFrequency()); - mQDecimationFilter = new PrimeCicDecimationFilter(sampleRate, decimation, channelSpecification.getPassFrequency(), - channelSpecification.getStopFrequency()); - - mBufferDispatcher = new Dispatcher(BUFFER_MAX_CAPACITY, "sdrtrunk cic channel " + - tunerChannel.getFrequency(), new NativeBufferPoisonPill()); - mBufferDispatcher.setListener(new NativeBufferProcessor()); - - //Setup the frequency mixer to the current source frequency - mChannelSampleRate = sampleRate / (double)decimation; - mTunerFrequency = tunerChannel.getFrequency(); - long frequencyOffset = mTunerFrequency - getTunerChannel().getFrequency(); - - mFrequencyCorrectionMixer = ComplexMixerFactory.getMixer(frequencyOffset, sampleRate); - } - - @Override - public void start() - { - super.start(); - mBufferDispatcher.start(); - } - - @Override - public void stop() - { - super.stop(); - mBufferDispatcher.stop(); - } - - @Override - public void dispose() - { - // - } - - /** - * Primary interface for receiving incoming complex sample buffers to be frequency translated and decimated. - */ - @Override - public void receive(T complexSamples) - { - mBufferDispatcher.receive(complexSamples); - } - - /** - * Sets/updates the center frequency for the sample streaming being sent from the producer. - * - * @param frequency in hertz - */ - @Override - public void setFrequency(long frequency) - { - mTunerFrequency = frequency; - updateMixerFrequencyOffset(); - } - - /** - * Current frequency correction being applied to the channel sample stream - * - * @return correction in hertz - */ - public long getFrequencyCorrection() - { - return mChannelFrequencyCorrection; - } - - /** - * Changes the frequency correction value and broadcasts the change to the registered downstream listener. - * - * @param correction current frequency correction value. - */ - public void setFrequencyCorrection(long correction) - { - mChannelFrequencyCorrection = correction; - updateMixerFrequencyOffset(); - broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mChannelFrequencyCorrection)); - } - - /** - * Calculates the local mixer frequency offset from the tuned frequency, - * channel's requested frequency, and channel frequency correction. - */ - private void updateMixerFrequencyOffset() - { - long offset = mTunerFrequency - getTunerChannel().getFrequency() - mChannelFrequencyCorrection; - mFrequencyCorrectionMixer.setFrequency(offset); - } - - /** - * Sets the sample rate of the incoming sample stream from the producer - * - * @param sampleRate in hertz - */ - @Override - protected void setSampleRate(double sampleRate) - { - //Not implemented. Sample rate changes are not permitted once sample stream starts - } - - /** - * Sets the frequency correction for the outbound sample stream to the consumer - * - * @param correction in hertz - */ - @Override - protected void setChannelFrequencyCorrection(long correction) - { - mChannelFrequencyCorrection = correction; - broadcastConsumerSourceEvent(SourceEvent.channelFrequencyCorrectionChange(correction)); - updateMixerFrequencyOffset(); - } - - /** - * Frequency correction value currently being applied to the outbound sample stream - */ - @Override - public long getChannelFrequencyCorrection() - { - return mChannelFrequencyCorrection; - } - - /** - * Sets the listener to receive the complex buffer sample output from this channel - * - * @param listener to receive complex buffers - */ - @Override - public void setListener(Listener listener) - { - mListener = listener; - } - - @Override - public double getSampleRate() - { - return mChannelSampleRate; - } - - /** - * Processes native buffers received from the dispatcher and sends to the registered listener - */ - public class NativeBufferProcessor implements Listener - { - @Override - public void receive(T nativeBuffer) - { - getHeartbeatManager().broadcast(); - - if(mListener != null) - { - Iterator iterator = nativeBuffer.iterator(); - - while(iterator.hasNext()) - { - ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next()); - float[] i = mIDecimationFilter.decimate(basebanded.i()); - float[] q = mQDecimationFilter.decimate(basebanded.q()); - - List repackaged = mSamplesRepackager.process(i, q); - - for(ComplexSamples samples: repackaged) - { - mListener.receive(samples); - } - } - } - } - } -} 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 f3ab04fe..23e02c88 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 @@ -54,6 +54,7 @@ public class HalfBandTunerChannelSource extends TunerCh private long mChannelFrequencyCorrection = 0; private long mTunerFrequency; private Listener mListener; + private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; /** * Constructs a frequency translating and CIC decimating channel source. @@ -70,6 +71,8 @@ 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); @@ -91,6 +94,7 @@ public class HalfBandTunerChannelSource extends TunerCh public void start() { super.start(); + mStreamHeartbeatProcessor.start(); mBufferDispatcher.start(); } @@ -99,6 +103,7 @@ public class HalfBandTunerChannelSource extends TunerCh { super.stop(); mBufferDispatcher.stop(); + mStreamHeartbeatProcessor.stop(); } @Override @@ -200,13 +205,7 @@ public class HalfBandTunerChannelSource extends TunerCh @Override public void setListener(Listener listener) { - if(listener == null) - { - //Wait for the native buffer processor thread to die before we nullify the listener - stop(); - } - - mListener = listener; + mStreamHeartbeatProcessor.setListener(listener); } @Override @@ -223,21 +222,14 @@ public class HalfBandTunerChannelSource extends TunerCh @Override public void receive(T nativeBuffer) { - getHeartbeatManager().broadcast(); + Iterator iterator = nativeBuffer.iterator(); - Listener listener = mListener; - - if(listener != null) + while(iterator.hasNext()) { - 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()); - listener.receive(new ComplexSamples(i, q)); - } + ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next()); + float[] i = mIDecimationFilter.decimateReal(basebanded.i()); + float[] q = mQDecimationFilter.decimateReal(basebanded.q()); + mStreamHeartbeatProcessor.receive(new ComplexSamples(i, q)); } } } 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 064a2b12..0b25176d 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 @@ -42,7 +42,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou private final static Logger mLog = LoggerFactory.getLogger(PassThroughChannelSource.class); private TunerController mTunerController; private Dispatcher mBufferDispatcher; - private Listener mComplexSamplesListener; + private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; /** * Constructs an instance @@ -59,12 +59,14 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou mBufferDispatcher = new Dispatcher<>(500, "sdrtrunk pass-through channel " + tunerChannel.getFrequency(), new NativeBufferPoisonPill()); mBufferDispatcher.setListener(new BufferProcessor()); + mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS); } @Override public void start() { super.start(); + mStreamHeartbeatProcessor.stop(); mBufferDispatcher.start(); } @@ -73,6 +75,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou { super.stop(); mBufferDispatcher.stop(); + mStreamHeartbeatProcessor.stop(); } @Override @@ -100,9 +103,9 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou } @Override - public void setListener(Listener complexSamplesListener) + public void setListener(Listener listener) { - mComplexSamplesListener = complexSamplesListener; + mStreamHeartbeatProcessor.setListener(listener); } @Override @@ -122,16 +125,11 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou @Override public void receive(INativeBuffer nativeBuffer) { - getHeartbeatManager().broadcast(); + Iterator iterator = nativeBuffer.iterator(); - if(mComplexSamplesListener != null) + while(iterator.hasNext()) { - Iterator iterator = nativeBuffer.iterator(); - - while(iterator.hasNext()) - { - mComplexSamplesListener.receive(iterator.next()); - } + mStreamHeartbeatProcessor.receive(iterator.next()); } } } 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 new file mode 100644 index 00000000..262402bb --- /dev/null +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java @@ -0,0 +1,135 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2022 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 java.util.concurrent.locks.ReentrantLock; + +/** + * 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 ReentrantLock mLock = new ReentrantLock(); + 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() + { + //Acquire lock only if it's not already locked, otherwise ignore heartbeat for this timer interval. + if(mLock.tryLock()) + { + try + { + mHeartbeatManager.broadcast(); + } + finally + { + mLock.unlock(); + } + } + } + + @Override + public void receive(T data) + { + mLock.lock(); + + try + { + mHeartbeatManager.broadcast(); + + Listener listener = mListener; + + if(listener != null) + { + listener.receive(data); + } + } + finally + { + mLock.unlock(); + } + } + + /** + * 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 5e5d303b..0f121b61 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 @@ -31,12 +31,11 @@ import org.slf4j.LoggerFactory; public abstract class TunerChannelSource extends ComplexSource implements ISourceEventProcessor { private final static Logger mLog = LoggerFactory.getLogger(TunerChannelSource.class); - private static final long BUFFER_PROCESSOR_RUN_INTERVAL_MILLISECONDS = 50; + protected static final long HEARTBEAT_INTERVAL_MS = 100; private SourceEventListenerToProcessorAdapter mConsumerSourceEventListenerAdapter; protected TunerChannel mTunerChannel; private Listener mProducerSourceEventListener; private Listener mConsumerSourceEventListener; -// private ScheduledIntervalProcessor mScheduledIntervalProcessor = new ScheduledIntervalProcessor(); /** * Tuner Channel Source is a Digital Drop Channel (DDC) abstract class that defines the minimum functionality @@ -245,95 +244,4 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc { mConsumerSourceEventListener = null; } - -// /** -// * Processor to periodically invoke buffer sample processing on an interval timer. At each interval this processor -// * sends a heartbeat to the registered consumer and then commands the sub-class implementation to process any -// * queued buffers and distribute complex buffer sample(s) to the registered consumer. -// */ -// public class ScheduledIntervalProcessor implements Runnable -// { -// private ScheduledFuture mScheduledFuture; -// private boolean mStopped = false; -// -// /** -// * Commands this processor to do a shutdown at the end of this or the next iteration. Once successfully -// * shutdown, it will invoke the performDisposal() method to cleanup this instance. -// */ -// public void stop() -// { -// mStopped = true; -// } -// -// /** -// * Starts buffer processing on a fixed interval using a scheduled thread pool -// */ -// public void start() -// { -// if(mScheduledFuture == null) -// { -// mScheduledFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(this, 0, -// BUFFER_PROCESSOR_RUN_INTERVAL_MILLISECONDS, TimeUnit.MILLISECONDS); -// } -// } -// -// /** -// * Implementation of the Runnable interface to periodically send a heartbeat and then process buffer samples. -// */ -// @Override -// public void run() -// { -// try -// { -// if(!mStopped) -// { -// try -// { -// getHeartbeatManager().broadcast(); -// } -// catch(Throwable t) -// { -// mLog.error("Error while sending heartbeat", t); -// } -// } -// -// if(!mStopped) -// { -// try -// { -// processSamples(); -// } -// catch(Throwable t) -// { -// mLog.error("Error while processing samples", t); -// } -// } -// -// if(mStopped) -// { -// if(mScheduledFuture != null) -// { -// //Set may-interrupt to false so that we can complete this iteration -// mScheduledFuture.cancel(false); -// } -// -// mScheduledFuture = null; -// -// try -// { -// getHeartbeatManager().broadcast(); -// performDisposal(); -// } -// catch(Throwable t) -// { -// mLog.error("Error during final shutdown processing of samples", t); -// } -// } -// } -// catch(Throwable t) -// { -// mLog.error("Error during heartbeat processing", t); -// } -// } -// } }