diff --git a/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java b/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java index 1438104f..7c044567 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java @@ -792,7 +792,7 @@ public class FilterFactory tapsPerChannel, windowType,logResults); //Decrease the overall gain of the filter to an objective of 1.0 divided by the number of channels - return normalize(filter, 1.0f);// / (float)channels); + return normalize(filter, 1.0f); } /** 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 8a7175bf..102f145b 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 @@ -34,7 +34,7 @@ import java.util.concurrent.CopyOnWriteArrayList; public abstract class AbstractComplexPolyphaseChannelizer implements Listener, ISourceEventListener { private final static Logger mLog = LoggerFactory.getLogger(AbstractComplexPolyphaseChannelizer.class); - private ReusableChannelResultsBufferQueue mBufferQueue = new ReusableChannelResultsBufferQueue(); + private ReusableChannelResultsBufferQueue mBufferQueue = new ReusableChannelResultsBufferQueue("AbstractComplexPolyphaseChannelizer"); private Broadcaster mSourceChangeBroadcaster = new Broadcaster(); private List mChannels = new CopyOnWriteArrayList<>(); private double mSampleRate; 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 2e3af921..4045673b 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 @@ -50,7 +50,8 @@ public class PolyphaseChannelManager implements ISourceEventProcessor private final static Logger mLog = LoggerFactory.getLogger(PolyphaseChannelManager.class); private static final double MINIMUM_CHANNEL_BANDWIDTH = 12500.0; private static final double CHANNEL_OVERSAMPLING = 2.0; - private static final int POLYPHASE_FILTER_TAPS_PER_CHANNEL = 8; + private static final int POLYPHASE_CHANNELIZER_TAPS_PER_CHANNEL = 8; + private static final int POLYPHASE_SYNTHESIZER_TAPS_PER_CHANNEL = 20; private Broadcaster mSourceEventBroadcaster = new Broadcaster<>(); private IReusableBufferProvider mReusableBufferProvider; @@ -266,6 +267,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor */ private void stopChannelSource(PolyphaseChannelSource channelSource) { + mLog.debug("Stopping channel source ... count:" + mChannelSources.size()); synchronized(mBufferProcessor) { mChannelSources.remove(channelSource); @@ -290,6 +292,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor { //Do nothing } + mLog.debug("Stopped channel source ... count:" + mChannelSources.size()); } /** @@ -349,7 +352,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor if(mPolyphaseChannelizer == null) { mPolyphaseChannelizer = new ComplexPolyphaseChannelizerM2(sampleRate, - POLYPHASE_FILTER_TAPS_PER_CHANNEL); + POLYPHASE_CHANNELIZER_TAPS_PER_CHANNEL); } else { @@ -514,7 +517,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor if(taps == null) { taps = FilterFactory.getSincM2Synthesizer(mChannelCalculator.getChannelBandwidth(), channels, - POLYPHASE_FILTER_TAPS_PER_CHANNEL + 20, Window.WindowType.BLACKMAN_HARRIS_7, true); + POLYPHASE_SYNTHESIZER_TAPS_PER_CHANNEL, Window.WindowType.BLACKMAN_HARRIS_7, true); mOutputProcessorFilters.put(channels, taps); } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2.java index 49955846..02a142cc 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2.java @@ -15,27 +15,47 @@ ******************************************************************************/ package io.github.dsheirer.dsp.filter.channelizer; -import io.github.dsheirer.dsp.filter.FilterFactory; -import io.github.dsheirer.dsp.filter.Window; -import io.github.dsheirer.dsp.mixer.IOscillator; -import io.github.dsheirer.dsp.mixer.LowPhaseNoiseOscillator; +import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; +import io.github.dsheirer.sample.buffer.ReusableComplexBufferQueue; import org.jtransforms.fft.FloatFFT_1D; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Arrays; +/** + * Implements a two-channel polyphase filer synthesizer. This class is intended to be used with an M2 polyphase + * channelizer in order to recover a signal that overlaps the boundary between two channels. This synthesizer will + * rejoin the two M2 neighboring channels, mix the signal of interest to baseband, and filter the the output stream to + * half of the sample rate, producing an overall M2 output sample rate. + * + * The synthesizer concept is modeled on the structure described by Fred Harris et al, in 'Interleaving Different + * Bandwidth Narrowband Channels in Perfect Reconstruction Cascade Polyphase Filter Banks for Efficient Flexible + * Variable Bandwidth Filters in Wideband Digital Transceivers'. + * + * The data buffers and sub-filter kernels bear no resemblance to the structures defined in the paper. The data and + * filter structures in this class are organized to take advantage of Java's ability to leverage SIMD processor + * intrinsics for optimal efficiency in processing. The paper specifies splitting the filter kernel into N polyphase + * partitions and creating N data buffers. This class organizes the data buffer as a contiguous sample + * array and creates two copies of the filter kernel, one arranged for normal filtering and the second arranged for + * phase-shifted filtering. Instead of calculating the dot-product for each sub-filter, we calculate the product of the + * full data array against the filter and then accumulate each sub-filter. This allows java to use SIMD for the array + * product and then normal processing for the accumulation. Since this is a two-channel processor and the results of + * each filter accumulation are added, we use a single accumulator across both filters. + * + */ public class TwoChannelSynthesizerM2 { private final static Logger mLog = LoggerFactory.getLogger(TwoChannelSynthesizerM2.class); - private float[] mData1i; - private float[] mData1q; - private float[] mData2i; - private float[] mData2q; - private float[] mFilter1; - private float[] mFilter2; - private float mAccumulator; + private ReusableComplexBufferQueue mReusableComplexBufferQueue = new ReusableComplexBufferQueue("Two Channel Synthesizer M2"); + private float[] mDataBuffer; + private float[] mNormalFilter; + private float[] mPhaseShiftedFilter; + private float[] mFilterProduct; + private float[] mIFFTBuffer = new float[4]; + private float mIAccumulator; + private float mQAccumulator; private FloatFFT_1D mFFT = new FloatFFT_1D(2); private boolean mFlag = true; @@ -62,100 +82,152 @@ public class TwoChannelSynthesizerM2 */ private void init(float[] filter) { - int length = (filter.length / 2) + (filter.length % 2); + int tapsPerChannel = (int)Math.ceil(filter.length / 2); - mFilter1 = new float[length]; - mFilter2 = new float[length]; - - mData1i = new float[length]; - mData1q = new float[length]; - mData2i = new float[length]; - mData2q = new float[length]; - - for(int x = 0; x < filter.length; x += 2) - { - //Load the filter in reverse to facilitate convolution - int index = length - (x / 2) - 1; - - mFilter1[index] = filter[x]; - - if(x + 1 < filter.length) - { - mFilter2[index] = filter[x + 1]; - } - } + mNormalFilter = getNormalFilter(filter, tapsPerChannel); + mPhaseShiftedFilter = getPhaseShiftedFilter(filter, tapsPerChannel); + mDataBuffer = new float[mNormalFilter.length]; + mFilterProduct = new float[mNormalFilter.length]; } /** * Synthesizes a new channel from the channel 1 and channel 2 sample arrays that are arranged as * i0, q0, i1, q1 ... iN-1, qN-1 * - * @param channel1 input channel - * @param channel2 input channel + * @param channelBuffer1 input channel + * @param channelBuffer2 input channel * @return synthesized channel results of the same length as both channel 1/2 input arrays. */ - public float[] process(float[] channel1, float[] channel2) + public ReusableComplexBuffer process(ReusableComplexBuffer channelBuffer1, ReusableComplexBuffer channelBuffer2) { - if(channel1.length != channel2.length) + if(channelBuffer1.getSamples().length != channelBuffer1.getSamples().length) { throw new IllegalArgumentException("Channel 1 and 2 array length must be equal"); } - //TODO: make this a reusable buffer - float[] output = new float[channel1.length]; + float[] channel1 = channelBuffer1.getSamples(); + float[] channel2 = channelBuffer2.getSamples(); - float[] buffer = new float[4]; + ReusableComplexBuffer synthesizedComplexBuffer = mReusableComplexBufferQueue.getBuffer(channel1.length); + + float[] output = synthesizedComplexBuffer.getSamples(); for(int x = 0; x < channel1.length; x += 2) { - buffer[0] = channel1[x]; //i - buffer[1] = channel1[x + 1]; //q - buffer[2] = channel2[x]; //i - buffer[3] = channel2[x + 1]; //q + mIFFTBuffer[0] = channel1[x]; //i + mIFFTBuffer[1] = channel1[x + 1]; //q + mIFFTBuffer[2] = channel2[x]; //i + mIFFTBuffer[3] = channel2[x + 1]; //q - mFFT.complexInverse(buffer, true); + mFFT.complexInverse(mIFFTBuffer, true); - System.arraycopy(mData1i, 0, mData1i, 1, mData1i.length - 1); - System.arraycopy(mData1q, 0, mData1q, 1, mData1q.length - 1); - System.arraycopy(mData2i, 0, mData2i, 1, mData2i.length - 1); - System.arraycopy(mData2q, 0, mData2q, 1, mData2q.length - 1); - - mData1i[0] = buffer[0]; - mData1q[0] = buffer[1]; - mData2i[0] = buffer[2]; - mData2q[0] = buffer[3]; + System.arraycopy(mDataBuffer, 0, mDataBuffer, 4, mDataBuffer.length - 4); + System.arraycopy(mIFFTBuffer, 0, mDataBuffer, 0, mIFFTBuffer.length); if(mFlag) { -// output[x] = filter(mFilter1, mData1i); -// output[x + 1] = filter(mFilter1, mData1q); - output[x] = filter(mFilter1, mData1i) + filter(mFilter2, mData2i); - output[x + 1] = filter(mFilter1, mData1q) + filter(mFilter2, mData2q); + for(int y = 0; y < mDataBuffer.length; y++) + { + mFilterProduct[y] = mDataBuffer[y] * mNormalFilter[y]; + } } else { -// output[x] = filter(mFilter1, mData2i); -// output[x + 1] = filter(mFilter1, mData2q); - output[x] = filter(mFilter1, mData2i) + filter(mFilter2, mData1i); - output[x + 1] = filter(mFilter1, mData2q) + filter(mFilter2, mData1q); + for(int y = 0; y < mDataBuffer.length; y++) + { + mFilterProduct[y] = mDataBuffer[y] * mPhaseShiftedFilter[y]; + } } + mIAccumulator = 0.0f; + mQAccumulator = 0.0f; + + for(int y = 0; y < mFilterProduct.length; y += 2) + { + mIAccumulator += mFilterProduct[y]; + mQAccumulator += mFilterProduct[y + 1]; + } + + output[x] = mIAccumulator; + output[x + 1] = mQAccumulator; + mFlag = !mFlag; } - return output; + channelBuffer1.decrementUserCount(); + channelBuffer2.decrementUserCount(); + + synthesizedComplexBuffer.incrementUserCount(); + + return synthesizedComplexBuffer; } - private float filter(float[] coefficients, float[] data) + private static float[] getNormalFilter(float[] coefficients, int tapsPerChannel) { - mAccumulator = 0.0f; + float[] filter = new float[4 * tapsPerChannel]; - for(int x = 0; x < coefficients.length; x++) + int idealLength = 2 * tapsPerChannel; + + int offset = idealLength - 2; + + int index; + + for(int x = 0; x < idealLength; x += 2) { - mAccumulator += coefficients[x] * data[x]; + index = offset - x; + + if(index < coefficients.length) + { + //Filter 1 + filter[2 * x + 0] = coefficients[index]; + filter[2 * x + 1] = coefficients[index]; + } + + index = offset - x + 1; + + if(index < coefficients.length) + { + //Filter 2 + filter[2 * x + 2] = coefficients[index]; + filter[2 * x + 3] = coefficients[index]; + } } - return mAccumulator; + return filter; + } + + private static float[] getPhaseShiftedFilter(float[] coefficients, int tapsPerChannel) + { + float[] filter = new float[4 * tapsPerChannel]; + + int idealLength = 2 * tapsPerChannel; + + int offset = idealLength - 2; + + int index; + + for(int x = 0; x < idealLength; x += 2) + { + index = offset - x + 1; + + if(index < coefficients.length) + { + //Filter 2 + filter[2 * x + 0] = coefficients[index]; + filter[2 * x + 1] = coefficients[index]; + } + + index = offset - x; + + if(index < coefficients.length) + { + //Filter 1 + filter[2 * x + 2] = coefficients[index]; + filter[2 * x + 3] = coefficients[index]; + } + } + + return filter; } public static void main(String[] args) @@ -166,19 +238,38 @@ public class TwoChannelSynthesizerM2 try { - float[] taps = FilterFactory.getSincM2Synthesizer(12500.0, 2, 8, Window.WindowType.BLACKMAN_HARRIS_7, true); +// float[] taps = FilterFactory.getSincM2Synthesizer(12500.0, 2, 20, Window.WindowType.BLACKMAN_HARRIS_7, true); - TwoChannelSynthesizerM2 synthesizer = new TwoChannelSynthesizerM2(taps); + float[] taps = new float[15]; + for(int x = 0; x < taps.length; x++) + { + taps[x] = x; + } - IOscillator oscillator = new LowPhaseNoiseOscillator(3000.0, 25000.0); + float[] normalFilter = getPhaseShiftedFilter(taps, 8); - float[] channel1 = oscillator.generateComplex(sampleCount); - float[] channel2 = new float[sampleCount * 2]; + mLog.debug("Filter: " + Arrays.toString(normalFilter)); - float[] synthesized = synthesizer.process(channel1, channel2); - - mLog.debug("1:" + Arrays.toString(channel1)); - mLog.debug("X:" + Arrays.toString(synthesized)); +// TwoChannelSynthesizerM2 synthesizer = new TwoChannelSynthesizerM2(taps); +// +// IOscillator oscillator = new LowPhaseNoiseOscillator(3000.0, 25000.0); +// +// float[] channel1 = oscillator.generateComplex(sampleCount); +// float[] channel2 = new float[sampleCount * 2]; +// +// float[] channel1 = new float[16]; +// float[] channel2 = new float[16]; +// for(int x = 1; x <= 16; x++) +// { +// channel1[x - 1] = x; +// channel2[x - 1] = x + 16; +// } +// +// +// float[] synthesized = synthesizer.process(channel1, channel2); +// +// mLog.debug("1:" + Arrays.toString(channel1)); +// mLog.debug("X:" + Arrays.toString(synthesized)); } catch(Exception fde) { 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 e27d199a..872700bb 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 @@ -20,7 +20,7 @@ package io.github.dsheirer.dsp.filter.channelizer.output; import io.github.dsheirer.dsp.mixer.IOscillator; import io.github.dsheirer.dsp.mixer.Oscillator; -import io.github.dsheirer.sample.OverflowableTransferQueue; +import io.github.dsheirer.sample.buffer.OverflowableReusableBufferTransferQueue; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import io.github.dsheirer.sample.real.IOverflowListener; @@ -35,7 +35,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP { private final static Logger mLog = LoggerFactory.getLogger(ChannelOutputProcessor.class); - private OverflowableTransferQueue mChannelResultsQueue; + private OverflowableReusableBufferTransferQueue mChannelResultsQueue; private List mChannelResultsToProcess = new ArrayList<>(); private int mMaxResultsToProcess; @@ -59,7 +59,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP 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)); + mChannelResultsQueue = new OverflowableReusableBufferTransferQueue<>((int)(sampleRate * 3), (int)(sampleRate * 0.5)); } public void dispose() 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 9684efa6..d06f1e44 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 @@ -20,6 +20,7 @@ package io.github.dsheirer.dsp.filter.channelizer.output; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; +import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -83,14 +84,14 @@ public class OneChannelOutputProcessor extends ChannelOutputProcessor { try { - float[] samples = channelResultsBuffer.getChannel(mChannelOffset); + ReusableComplexBuffer channelBuffer = channelResultsBuffer.getChannel(mChannelOffset); if(hasFrequencyCorrection()) { - getFrequencyCorrectionMixer().mixComplex(samples); + getFrequencyCorrectionMixer().mixComplex(channelBuffer.getSamples()); } - reusableBufferAssembler.receive(samples); + reusableBufferAssembler.receive(channelBuffer); } catch(IllegalArgumentException iae) { 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 f7c3a635..42871dac 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 @@ -23,6 +23,7 @@ import io.github.dsheirer.dsp.filter.fir.remez.RemezFIRFilterDesigner; import io.github.dsheirer.dsp.mixer.FS4DownConverter; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; +import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -139,21 +140,21 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor { for(ReusableChannelResultsBuffer buffer: channelResultsBuffers) { - float[] channel1 = buffer.getChannel(mChannelOffset1); - float[] channel2 = buffer.getChannel(mChannelOffset2); + ReusableComplexBuffer channel1 = buffer.getChannel(mChannelOffset1); + ReusableComplexBuffer channel2 = buffer.getChannel(mChannelOffset2); //Join the two channels using the synthesizer - float[] synthesized = mSynthesizer.process(channel1, channel2); + ReusableComplexBuffer synthesized = mSynthesizer.process(channel1, channel2); //The synthesized channels are centered at +FS/4 ... downconvert to center the spectrum - mFS4DownConverter.mixComplex(synthesized); + mFS4DownConverter.mixComplex(synthesized.getSamples()); //Apply offset and frequency correction to center the signal of interest within the synthesized channel - getFrequencyCorrectionMixer().mixComplex(synthesized); + getFrequencyCorrectionMixer().mixComplex(synthesized.getSamples()); - mLowPassFilter.filter(synthesized); + ReusableComplexBuffer lowPassFiltered = mLowPassFilter.filter(synthesized); - reusableBufferAssembler.receive(synthesized); + reusableBufferAssembler.receive(lowPassFiltered); buffer.decrementUserCount(); } diff --git a/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java b/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java index 2997a6ec..05997803 100644 --- a/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java +++ b/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java @@ -301,15 +301,17 @@ public class SynthesizerViewer extends JFrame ReusableComplexBuffer channel2Buffer = mReusableComplexBufferQueue.getBuffer(mSamplesPerCycle); getChannel2ControlPanel().getOscillator().generateComplex(channel2Buffer); - float[] synthesized = mSynthesizer.process(channel1Buffer.getSamples(), channel2Buffer.getSamples()); - ReusableComplexBuffer synthesizedBuffer = mReusableComplexBufferQueue.getBuffer(synthesized.length); - System.arraycopy(synthesized, 0, synthesizedBuffer.getSamples(), 0, synthesized.length); - channel1Buffer.incrementUserCount(); - getChannel1Panel().receive(channel1Buffer); + channel1Buffer.incrementUserCount(); channel2Buffer.incrementUserCount(); + channel2Buffer.incrementUserCount(); + + ReusableComplexBuffer synthesizedBuffer = mSynthesizer.process(channel1Buffer, channel2Buffer); + + getChannel1Panel().receive(channel1Buffer); + getChannel2Panel().receive(channel2Buffer); - synthesizedBuffer.incrementUserCount(); + getSpectrumPanel().receive(synthesizedBuffer); } } diff --git a/src/main/java/io/github/dsheirer/module/ProcessingChain.java b/src/main/java/io/github/dsheirer/module/ProcessingChain.java index 25197f1b..1a92cfc9 100644 --- a/src/main/java/io/github/dsheirer/module/ProcessingChain.java +++ b/src/main/java/io/github/dsheirer/module/ProcessingChain.java @@ -626,6 +626,8 @@ public class ProcessingChain implements IChannelEventListener { removeModule(mSource); + mSource.stop(); + mSource.setOverflowListener(null); switch(mSource.getSampleType()) diff --git a/src/main/java/io/github/dsheirer/sample/buffer/OverflowableReusableBufferTransferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/OverflowableReusableBufferTransferQueue.java index 914833bb..874f1a9b 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/OverflowableReusableBufferTransferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/OverflowableReusableBufferTransferQueue.java @@ -17,7 +17,7 @@ package io.github.dsheirer.sample.buffer; import io.github.dsheirer.sample.OverflowableTransferQueue; -public class OverflowableReusableBufferTransferQueue extends OverflowableTransferQueue +public class OverflowableReusableBufferTransferQueue extends OverflowableTransferQueue { /** * Concurrent transfer queue that couples a higher-throughput linked transfer queue with an atomic integer for diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java index de1dd829..5575d016 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java @@ -102,10 +102,7 @@ public class ReusableBuffer extends AbstractReusableBuffer mUserCount.set(0); - if(mSamples.length != samples.length) - { - mSamples = new float[samples.length]; - } + resize(samples.length); System.arraycopy(samples, 0, mSamples, 0, samples.length); @@ -129,15 +126,23 @@ public class ReusableBuffer extends AbstractReusableBuffer mUserCount.set(0); - if(mSamples.length != floatBuffer.capacity()) - { - mSamples = new float[floatBuffer.capacity()]; - } - + resize(floatBuffer.capacity()); floatBuffer.rewind(); floatBuffer.get(mSamples); mTimestamp = timestamp; } + /** + * Resizes the internal array to the size argument + * @param size for the internal array + */ + protected void resize(int size) + { + if(mSamples.length != size) + { + mSamples = new float[size]; + } + } + } diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferAssembler.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferAssembler.java index 73c43c38..111c33cb 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferAssembler.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferAssembler.java @@ -107,6 +107,16 @@ public class ReusableBufferAssembler } } + /** + * Adds the samples to this assembler. As each buffer is assembled, it will be dispatched to the registered + * listener. + */ + public void receive(ReusableComplexBuffer samplesBuffer) + { + receive(samplesBuffer.getSamples()); + samplesBuffer.decrementUserCount(); + } + /** * Flushes the current buffer contents to the registered listener */ diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java index 292e1033..bb96adea 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java @@ -51,6 +51,8 @@ public class ReusableBufferQueue extends AbstractReusableBufferQueue mEmptyBuffers = new LinkedList<>(); private LinkedList mFilledBuffers = new LinkedList<>(); private Integer mLength; @@ -103,7 +109,7 @@ public class ReusableChannelResultsBuffer extends AbstractReusableBuffer * @throws IllegalArgumentException if this buffer doesn't contain any samples or if the requested channel is * not within the channel range of the contained channel results. */ - public float[] getChannel(int iChannelIndex) + public ReusableComplexBuffer getChannel(int iChannelIndex) { if(mFilledBuffers.isEmpty()) { @@ -116,18 +122,35 @@ public class ReusableChannelResultsBuffer extends AbstractReusableBuffer "results -- max channel is " + mLength); } - float[] samples = new float[mFilledBuffers.size() * 2]; + ReusableComplexBuffer channelBuffer = mReusableComplexBufferQueue.getBuffer(mFilledBuffers.size() * 2); + + float[] samples = channelBuffer.getSamples(); + int pointer = 0; int qChannelIndex = iChannelIndex + 1; for(float[] channelResults: mFilledBuffers) { - samples[pointer++] = channelResults[iChannelIndex]; - samples[pointer++] = channelResults[qChannelIndex]; + try + { + samples[pointer] = channelResults[iChannelIndex]; + pointer++; + samples[pointer] = channelResults[qChannelIndex]; + pointer++; + } + catch(Exception e) + { + mLog.error("Error accessing channel results - iIndex:" + iChannelIndex + + " Results Size: " + mFilledBuffers.size() + + " Samples Length:" + samples.length + + " Pointer:" + pointer, e); + } } - return samples; + channelBuffer.incrementUserCount(); + + return channelBuffer; } /** diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java index c8084129..c57b7139 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java @@ -58,6 +58,8 @@ public class ReusableComplexBufferQueue extends AbstractReusableBufferQueue