From e395d737f8e3f70fa32d75f4b4c2433e39af124d Mon Sep 17 00:00:00 2001 From: Denny Date: Sun, 22 Apr 2018 04:44:35 -0400 Subject: [PATCH] [WIP]-39 updates to resolve issue with stopping channel streams. --- .../filter/channelizer/ChannelCalculator.java | 2 +- .../ComplexPolyphaseChannelizerM2.java | 23 ++++++++ .../ContinuousBufferProcessor.java | 7 ++- .../channelizer/PolyphaseChannelManager.java | 39 +++++++++++-- .../channelizer/PolyphaseChannelSource.java | 47 ++++++++++------ .../output/ChannelOutputProcessor.java | 14 ++++- .../IPolyphaseChannelOutputProcessor.java | 3 + .../dsheirer/module/ProcessingChain.java | 2 - .../sample/OverflowableTransferQueue.java | 7 +++ .../sample/buffer/AbstractReusableBuffer.java | 55 ++++++++++++++----- .../buffer/AbstractReusableBufferQueue.java | 27 ++++++++- .../buffer/ReusableBufferAssembler.java | 4 +- .../buffer/ReusableChannelResultsBuffer.java | 6 +- .../buffer/ReusableComplexBufferQueue.java | 7 +++ .../tuner/channel/TunerChannelSource.java | 27 ++++----- 15 files changed, 206 insertions(+), 64 deletions(-) diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java index a33b2f2b..2fb541f0 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java @@ -33,7 +33,7 @@ public class ChannelCalculator * from an index where the frequency may exist either at the overlapping boundary of two indexes, or in the middle * of the wrap-around (N/2) index. */ - private enum IndexBoundaryPolicy + public enum IndexBoundaryPolicy { ADJUST_POSITIVE, ADJUST_NEGATIVE diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java index a6be055a..d727d72c 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java @@ -31,6 +31,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Arrays; +import java.util.Collection; import java.util.List; public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChannelizer @@ -132,11 +133,17 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann init(filterTaps); } + /** + * Starts sample processing + */ public void start() { mIFFTProcessor.start(); } + /** + * Stops sample processing. + */ public void stop() { mIFFTProcessor.stop(); @@ -504,6 +511,22 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann } }); } + + /** + * Clears any buffers from the dispatch/processing queue. Overrides the parent method so that we can set + * the user count to 0 to allow the buffer to be reclaimed. + */ + protected Collection clearQueue() + { + Collection buffersToDispose = super.clearQueue(); + + for(ReusableChannelResultsBuffer buffer: buffersToDispose) + { + buffer.clearUserCount(); + } + + return buffersToDispose; + } } public static void main(String[] args) diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java index 9c3a1791..1df13430 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java @@ -23,6 +23,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -138,9 +139,11 @@ public class ContinuousBufferProcessor implements Listener /** * Clears any buffers from the queue */ - protected void clearQueue() + protected Collection clearQueue() { - mQueue.clear(); + Collection clearedElements = new ArrayList<>(); + mQueue.drainTo(clearedElements); + return clearedElements; } /** 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 dd11711f..660be28d 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 @@ -26,6 +26,7 @@ import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.buffer.IReusableBufferProvider; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; 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; @@ -134,21 +135,34 @@ public class PolyphaseChannelManager implements ISourceEventProcessor */ public TunerChannelSource getChannel(TunerChannel tunerChannel) { + mLog.debug("Getting channel: " + tunerChannel.toString()); PolyphaseChannelSource channelSource = null; try { List polyphaseIndexes = mChannelCalculator.getChannelIndexes(tunerChannel); + StringBuilder sb = new StringBuilder(); + sb.append("Channel Indexes - "); + for(int index: polyphaseIndexes) + { + sb.append(index).append(":").append((long)mChannelCalculator.getIndexCenterFrequency(index, + ChannelCalculator.IndexBoundaryPolicy.ADJUST_POSITIVE)).append(" "); + } + mLog.debug(sb.toString()); + IPolyphaseChannelOutputProcessor outputProcessor = getOutputProcessor(polyphaseIndexes); + mLog.debug("Channel Ouput Processor: " + outputProcessor.getClass()); + if(outputProcessor != null) { + long centerFrequency = mChannelCalculator.getCenterFrequencyForIndexes(polyphaseIndexes); + channelSource = new PolyphaseChannelSource(tunerChannel, outputProcessor, mChannelSourceEventListener, - mChannelCalculator.getChannelSampleRate()); + mChannelCalculator.getChannelSampleRate(), centerFrequency); mChannelSources.add(channelSource); - mSourceEventBroadcaster.broadcast(SourceEvent.channelCountChange(getTunerChannelCount())); } } catch(IllegalArgumentException iae) @@ -176,8 +190,10 @@ public class PolyphaseChannelManager implements ISourceEventProcessor if(outputProcessor != null) { + long centerFrequency = mChannelCalculator.getCenterFrequencyForIndexes(polyphaseIndexes); + channelSource = new PolyphaseChannelSource(tunerChannel, outputProcessor, mChannelSourceEventListener, - mChannelCalculator.getChannelSampleRate()); + mChannelCalculator.getChannelSampleRate(), centerFrequency); } } catch(IllegalArgumentException iae) @@ -227,8 +243,12 @@ public class PolyphaseChannelManager implements ISourceEventProcessor { synchronized(mBufferProcessor) { - mPolyphaseChannelizer.addChannel(channelSource); + //Note: the polyphase channel source has already been added to the mChannelSources in getChannel() method + mPolyphaseChannelizer.addChannel(channelSource); + mSourceEventBroadcaster.broadcast(SourceEvent.channelCountChange(getTunerChannelCount())); + + //If this is the first channel, register to start the sample buffers flowing if(mPolyphaseChannelizer.getRegisteredChannelCount() == 1) { mReusableBufferProvider.addBufferListener(mBufferProcessor); @@ -249,8 +269,10 @@ public class PolyphaseChannelManager implements ISourceEventProcessor synchronized(mBufferProcessor) { mChannelSources.remove(channelSource); + mPolyphaseChannelizer.removeChannel(channelSource); mSourceEventBroadcaster.broadcast(SourceEvent.channelCountChange(getTunerChannelCount())); + //If this is the last/only channel, deregister to stop the sample buffers if(mPolyphaseChannelizer.getRegisteredChannelCount() == 0) { mReusableBufferProvider.removeBufferListener(mBufferProcessor); @@ -532,9 +554,18 @@ 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; default: mLog.error("Received unrecognized source event from polyphase channel source [" + sourceEvent.getEvent() + "]"); + break; } } } 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 29a04521..19a09b36 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 @@ -33,8 +33,8 @@ public class PolyphaseChannelSource extends TunerChannelSource private ReusableBufferAssembler mReusableBufferAssembler; private IPolyphaseChannelOutputProcessor mPolyphaseChannelOutputProcessor; private double mChannelSampleRate; - private long mCenterFrequency; - private long mFrequencyCorrection; + private long mIndexCenterFrequency; + private long mChannelFrequencyCorrection; /** * Polyphase channelizer tuner channel source implementation. Adapts the channel array output samples from the @@ -45,14 +45,17 @@ public class PolyphaseChannelSource extends TunerChannelSource * @param outputProcessor - to process polyphase channelizer channel results into a channel stream * @param producerSourceEventListener to receive source event requests (e.g. start/stop sample stream) * @param channelSampleRate for the downstream sample output + * @param centerFrequency of the incoming polyphase channel(s) */ public PolyphaseChannelSource(TunerChannel tunerChannel, IPolyphaseChannelOutputProcessor outputProcessor, - Listener producerSourceEventListener, double channelSampleRate) + Listener producerSourceEventListener, double channelSampleRate, + long centerFrequency) { super(producerSourceEventListener, tunerChannel); mPolyphaseChannelOutputProcessor = outputProcessor; mChannelSampleRate = channelSampleRate; mReusableBufferAssembler = new ReusableBufferAssembler(2500, mChannelSampleRate); + setFrequency(centerFrequency); } /** @@ -114,10 +117,10 @@ public class PolyphaseChannelSource extends TunerChannelSource //Finally, setup the frequency offset for the output processor and reset the frequency correction value //to allow consumers to adjust and calculate a new frequency correction value, if needed. - mCenterFrequency = frequency; - mFrequencyCorrection = 0; + mIndexCenterFrequency = frequency; + mChannelFrequencyCorrection = 0; mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); - broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mFrequencyCorrection)); + broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mChannelFrequencyCorrection)); } } @@ -158,13 +161,22 @@ public class PolyphaseChannelSource extends TunerChannelSource @Override public void dispose() { - mLog.info("Polyphase channel source received request to dispose"); + if(mReusableBufferAssembler != null) + { + mReusableBufferAssembler.dispose(); + } + + if(mPolyphaseChannelOutputProcessor != null) + { + mPolyphaseChannelOutputProcessor.dispose(); + mPolyphaseChannelOutputProcessor = null; + } } @Override - public long getFrequencyCorrection() + public long getChannelFrequencyCorrection() { - return mFrequencyCorrection; + return mChannelFrequencyCorrection; } /** @@ -173,26 +185,26 @@ public class PolyphaseChannelSource extends TunerChannelSource * * @param value to apply for frequency correction in hertz */ - protected void setFrequencyCorrection(long value) + protected void setChannelFrequencyCorrection(long value) { - mFrequencyCorrection = value; + mChannelFrequencyCorrection = value; updateFrequencyOffset(); - broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mFrequencyCorrection)); + broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mChannelFrequencyCorrection)); } /** - * Sets the center frequency for the incoming sample stream channel results and resets frequency correction to zero. + * Sets the center frequency for this channel.frequency for the incoming sample stream channel results and resets frequency correction to zero. * @param frequency in hertz */ @Override protected void setFrequency(long frequency) { - mLog.debug("Center Frequency set to: " + frequency + " for channel " + getTunerChannel().toString()); - mCenterFrequency = frequency; + mLog.debug("Index Center Frequency set to: " + frequency + " for channel " + getTunerChannel().toString()); + mIndexCenterFrequency = frequency; //Set frequency correction to zero to trigger an update to the mixer and allow downstream monitors to //recalculate the frequency error correction again - setFrequencyCorrection(0); + setChannelFrequencyCorrection(0); } @Override @@ -210,7 +222,7 @@ public class PolyphaseChannelSource extends TunerChannelSource */ private long getFrequencyOffset() { - return getTunerChannel().getFrequency() - mCenterFrequency + mFrequencyCorrection; + return getTunerChannel().getFrequency() - mIndexCenterFrequency + mChannelFrequencyCorrection; } /** @@ -222,6 +234,7 @@ public class PolyphaseChannelSource extends TunerChannelSource { synchronized(mPolyphaseChannelOutputProcessor) { + mLog.debug("Setting output processor frequency offset to: " + getFrequencyOffset()); mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); } } 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 69e7ec6f..e27d199a 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 @@ -40,7 +40,6 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP private int mMaxResultsToProcess; private int mInputChannelCount; -//TODO: swap this out and use the LowPhaseNoiseOscillator private IOscillator mFrequencyCorrectionMixer; private boolean mFrequencyCorrectionEnabled; @@ -55,12 +54,23 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP public ChannelOutputProcessor(int inputChannelCount, double sampleRate) { mInputChannelCount = inputChannelCount; + +//TODO: swap this out and use the LowPhaseNoiseOscillator mFrequencyCorrectionMixer = new Oscillator(0, sampleRate); mMaxResultsToProcess = (int)(sampleRate / 10) * 2; //process at 100 millis interval, twice the expected inflow rate mChannelResultsQueue = new OverflowableTransferQueue<>((int)(sampleRate * 3), (int)(sampleRate * 0.5)); } + public void dispose() + { + if(mChannelResultsQueue != null) + { + mChannelResultsQueue.dispose(); + mChannelResultsQueue = null; + } + } + /** * Sets the frequency offset to apply to the incoming samples to mix the desired signal to baseband. * @param frequencyOffset in hertz @@ -68,7 +78,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP @Override public void setFrequencyOffset(long frequencyOffset) { - mLog.debug("Setting frequency offset to: " + frequencyOffset); + mLog.debug("Setting frequency offset to: " + frequencyOffset + " for class:" + this.getClass()); mFrequencyCorrectionMixer.setFrequency(frequencyOffset); mFrequencyCorrectionEnabled = (frequencyOffset != 0); } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java index c9856f25..95c45055 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java @@ -70,4 +70,7 @@ public interface IPolyphaseChannelOutputProcessor * @param source to receive overflow notifications */ void setSourceOverflowListener(Source source); + + + void dispose(); } diff --git a/src/main/java/io/github/dsheirer/module/ProcessingChain.java b/src/main/java/io/github/dsheirer/module/ProcessingChain.java index 34b81cdb..5beafede 100644 --- a/src/main/java/io/github/dsheirer/module/ProcessingChain.java +++ b/src/main/java/io/github/dsheirer/module/ProcessingChain.java @@ -641,8 +641,6 @@ public class ProcessingChain implements IChannelEventListener "chain"); } - /* Release the source */ - mSource.dispose(); mSource = null; } } diff --git a/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java b/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java index d3ad1541..c342ac3f 100644 --- a/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java @@ -56,6 +56,13 @@ public class OverflowableTransferQueue mResetThreshold = resetThreshold; } + public void dispose() + { + clear(); + mOverflowListener = null; + mSourceOverflowListener = null; + } + /** * Adds the element to the queue if able to do so without exceeding maximum queue size. Otherwise, ignores * the element. diff --git a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java index d374e76a..77fbaf2e 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java @@ -15,10 +15,15 @@ ******************************************************************************/ package io.github.dsheirer.sample.buffer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.concurrent.atomic.AtomicInteger; public abstract class AbstractReusableBuffer { + private final static Logger mLog = LoggerFactory.getLogger(AbstractReusableBuffer.class); + protected IReusableBufferDisposedListener mBufferDisposedListener; protected long mTimestamp; protected AtomicInteger mUserCount = new AtomicInteger(); @@ -35,6 +40,11 @@ public abstract class AbstractReusableBuffer mBufferDisposedListener = bufferDisposedListener; } + public void dispose() + { + mBufferDisposedListener = null; + } + public void setDebugName(String debugName) { mDebugName = debugName; @@ -65,15 +75,12 @@ public abstract class AbstractReusableBuffer } /** - * Sets the number of users that will receive this buffer. Each user is expected to invoke the - * decrementUserCount() method to indicate that they are finished using this buffer so that this buffer can be - * reused once the user count is zero. - * - * @param userCount that will receive this reusable buffer + * Sets the user count to 0 and recycles this buffer */ - public void setUserCount(int userCount) + public void clearUserCount() { - mUserCount.set(userCount); + mUserCount.set(0); + recycle(); } /** @@ -88,14 +95,25 @@ public abstract class AbstractReusableBuffer */ public void decrementUserCount() { - int currentCount = mUserCount.decrementAndGet(); - - if(currentCount == 0) + if(mUserCount.decrementAndGet() <= 0) { recycle(); + } + } + + /** + * Sends this buffer back to the owning buffer queue for reuse + */ + private void recycle() + { + prepareForRecycle(); + + if(mBufferDisposedListener != null) + { mBufferDisposedListener.disposed(this); } - else if(currentCount < 0) + + if(mUserCount.get() < 0) { throw new IllegalStateException("User count is below zero. This indicates that this buffer's decrement" + " user count was invoked by more than the expected user count"); @@ -103,10 +121,10 @@ public abstract class AbstractReusableBuffer } /** - * Invoked just prior to notifying the owner that this buffer is ready for recycle. This method - * is intended for sub-class implementations to perform any recycle cleanup actions. + * Invoked just prior to notifying the owner that this buffer is ready for prepareForRecycle. This method + * is intended for sub-class implementations to perform any prepareForRecycle cleanup actions. */ - protected void recycle() + protected void prepareForRecycle() { //No-op in this abstract class } @@ -138,7 +156,14 @@ public abstract class AbstractReusableBuffer */ public void incrementUserCount(int additionalUserCount) { - mUserCount.addAndGet(additionalUserCount); + if(additionalUserCount >= 0) + { + mUserCount.addAndGet(additionalUserCount); + } + else + { + mLog.debug("Request to add negative user count [" + additionalUserCount + "] ignored"); + } } /** diff --git a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java index 18b618e0..e777b1eb 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java @@ -30,18 +30,37 @@ public abstract class AbstractReusableBufferQueue mProducerSourceEventListener; private Listener mConsumerSourceEventListener; - private BufferProcessor mBufferProcessor = new BufferProcessor(); + private ScheduledIntervalProcessor mScheduledIntervalProcessor = new ScheduledIntervalProcessor(); /** * Tuner Channel Source is a Digital Drop Channel (DDC) abstract class that defines the minimum functionality @@ -78,12 +78,12 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc * Sets the frequency correction for the outbound sample stream to the consumer * @param correction in hertz */ - protected abstract void setFrequencyCorrection(long correction); + protected abstract void setChannelFrequencyCorrection(long correction); /** * Frequency correction value currently being applied to the outbound sample stream */ - public abstract long getFrequencyCorrection(); + public abstract long getChannelFrequencyCorrection(); /** * Sets the listener to receive the complex buffer sample output from this channel @@ -125,7 +125,7 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc mProducerSourceEventListener.receive(SourceEvent.startSampleStreamRequest(this)); } - mBufferProcessor.start(); + mScheduledIntervalProcessor.start(); } /** @@ -140,7 +140,7 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc mProducerSourceEventListener.receive(SourceEvent.sourceDisposeRequest(this)); } - mBufferProcessor.stop(); + mScheduledIntervalProcessor.stop(); } @Override @@ -172,22 +172,19 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc { switch(sourceEvent.getEvent()) { - //Notification events from the producer case NOTIFICATION_FREQUENCY_CHANGE: - mLog.debug("Processing Source Event - setting frequency to " + sourceEvent.getValue().longValue()); - setFrequency(sourceEvent.getValue().longValue()); + case NOTIFICATION_FREQUENCY_AND_SAMPLE_RATE_LOCKED: + case NOTIFICATION_FREQUENCY_AND_SAMPLE_RATE_UNLOCKED: + case NOTIFICATION_FREQUENCY_CORRECTION_CHANGE: + case NOTIFICATION_STOP_SAMPLE_STREAM: + //no-op break; case NOTIFICATION_SAMPLE_RATE_CHANGE: setSampleRate(sourceEvent.getValue().doubleValue()); break; - case NOTIFICATION_FREQUENCY_AND_SAMPLE_RATE_LOCKED: - case NOTIFICATION_FREQUENCY_AND_SAMPLE_RATE_UNLOCKED: - case NOTIFICATION_FREQUENCY_CORRECTION_CHANGE: - //no-op - break; //Request events from the consumer case REQUEST_CHANNEL_FREQUENCY_CORRECTION_CHANGE: - setFrequencyCorrection(sourceEvent.getValue().longValue()); + setChannelFrequencyCorrection(sourceEvent.getValue().longValue()); break; default: mLog.error("Ignoring unrecognized source event: " + sourceEvent.getEvent() + " from [" + @@ -255,7 +252,7 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc * 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 BufferProcessor implements Runnable + public class ScheduledIntervalProcessor implements Runnable { private ScheduledFuture mScheduledFuture; private boolean mStopped = false;