diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableBufferDisposedListener.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableBufferDisposedListener.java new file mode 100644 index 00000000..fd261e23 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableBufferDisposedListener.java @@ -0,0 +1,25 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +public interface IReusableBufferDisposedListener +{ + /** + * Notifies the listener that the timestamped buffer has been disposed and can now be reused. + * @param reusableComplexBuffer that has been disposed + */ + void disposed(ReusableComplexBuffer reusableComplexBuffer); +} diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferListener.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferListener.java new file mode 100644 index 00000000..2f7a7306 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferListener.java @@ -0,0 +1,23 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +import io.github.dsheirer.sample.Listener; + +public interface IReusableComplexBufferListener +{ + Listener getReusableComplexBufferListener(); +} diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferProvider.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferProvider.java new file mode 100644 index 00000000..96ad66dd --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/IReusableComplexBufferProvider.java @@ -0,0 +1,36 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +import io.github.dsheirer.sample.Listener; + +public interface IReusableComplexBufferProvider +{ + /** + * Adds the listener to receive complex buffer samples + */ + void addBufferListener(Listener listener); + + /** + * Removes the listener from receiving complex buffer samples + */ + void removeBufferListener(Listener listener); + + /** + * Indicates if there are any complex buffer listeners registered + */ + boolean hasBufferListeners(); +} diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferAssembler.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferAssembler.java new file mode 100644 index 00000000..a58b62f3 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferAssembler.java @@ -0,0 +1,136 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +import io.github.dsheirer.dsp.filter.channelizer.SampleTimestampManager; +import io.github.dsheirer.sample.Listener; +import org.apache.commons.lang3.Validate; + +import java.nio.FloatBuffer; + +/** + * Assembles complex float samples into a ComplexBuffer containing an array of floats. Monitors incoming sample count + * to accurately assign a relative timestamp to each assembled buffer. + */ +public class ReusableBufferAssembler +{ + private ReusableBufferQueue mReusableBufferQueue = new ReusableBufferQueue(); + private SampleTimestampManager mTimestampManager; + private FloatBuffer mBuffer; + private int mBufferSize; + private long mCurrentBufferTimestamp; + private Listener mListener; + + public ReusableBufferAssembler(int bufferSize, double sampleRate) + { + Validate.isTrue(bufferSize % 2 == 0); + mBufferSize = bufferSize; + mBuffer = FloatBuffer.allocate(mBufferSize); + mTimestampManager = new SampleTimestampManager(sampleRate); + } + + /** + * Updates the reference timestamp to use for all subsequent samples that are assembled. This timestamp should + * identify the relative timestamp of the first/next sample that is received by this assembler. + * @param timestamp to use relative to the incoming sample arrays. + */ + public void updateTimestamp(long timestamp) + { + mTimestampManager.setReferenceTimestamp(timestamp); + + if(mBuffer.position() == 0) + { + mCurrentBufferTimestamp = timestamp; + } + } + + public void dispose() + { + mListener = null; + mBuffer = null; + } + + /** + * Adds the samples to this assembler. As each buffer is assembled, it will be dispatched to the registered + * listener. + * + * @param samples to assemble + */ + public void receive(float[] samples) + { + if(mBuffer.remaining() >= samples.length) + { + mBuffer.put(samples); + mTimestampManager.increment(samples.length / 2); + + if(!mBuffer.hasRemaining()) + { + flush(); + } + } + else + { + int offset = 0; + + while(offset < samples.length) + { + int toCopy = mBuffer.remaining(); + + if((samples.length - offset) < toCopy) + { + toCopy = samples.length - offset; + } + + mBuffer.put(samples, offset, toCopy); + + offset += toCopy; + mTimestampManager.increment(toCopy / 2); + + if(!mBuffer.hasRemaining()) + { + flush(); + } + } + } + } + + /** + * Flushes the current buffer contents to the registered listener + */ + public void flush() + { + if(mListener != null && mBuffer.position() > 0) + { + mBuffer.rewind(); + ReusableComplexBuffer reusableComplexBuffer = mReusableBufferQueue.getBuffer(mBufferSize); + reusableComplexBuffer.reloadFrom(mBuffer, mCurrentBufferTimestamp); + reusableComplexBuffer.setUserCount(1); + mListener.receive(reusableComplexBuffer); + } + + mBuffer.rewind(); + mCurrentBufferTimestamp = mTimestampManager.getCurrentTimestamp(); + } + + /** + * Sets the listener to receive the output from this buffer assembler. Note: the listener is responsible for + * managing the user count for any buffers received from this buffer assembler. + */ + public void setListener(Listener listener) + { + mListener = listener; + } +} diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferBroadcaster.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferBroadcaster.java new file mode 100644 index 00000000..e2ac0f80 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferBroadcaster.java @@ -0,0 +1,67 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +import io.github.dsheirer.sample.Broadcaster; +import io.github.dsheirer.sample.Listener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ReusableBufferBroadcaster extends Broadcaster +{ + private final static Logger mLog = LoggerFactory.getLogger(ReusableBufferBroadcaster.class); + + /** + * Increments the user count for the reusable complex buffer and then broadcasts the buffer to all registered + * listeners. + * + * The total user count is established and applied to the buffer prior to dispatching. If we were to simply + * increment the user count prior to sending to each consumer, there is a possibility that the consumer could + * immediately decrement the user count and prematurely signal that the buffer is ready for disposal before we + * send the buffer to all consumers. + */ + @Override + public void broadcast(ReusableComplexBuffer reusableComplexBuffer) + { + //There is a minuscule possibility that the listener list will change from the time that we get the size until + //the for/each iterator is created below, therefore we have to accurately track the user count. If a listener + //is added in the time between getting the list size and creating the iterator, that listener will NOT receive + //a copy of the buffer, but will instead get the next buffer. Likewise, if a listener is removed, we have to + //decrement the user count on the buffer after iterating the listener list. + int listenerCount = mListeners.size(); + + reusableComplexBuffer.incrementUserCount(listenerCount); + + for(Listener listener : mListeners) + { + if(listenerCount > 0) + { + listener.receive(reusableComplexBuffer); + } + + listenerCount--; + } + + while(listenerCount > 0) + { + reusableComplexBuffer.decrementUserCount(); + listenerCount--; + } + + //Decrement user counter for this broadcaster + reusableComplexBuffer.decrementUserCount(); + } +} diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferQueue.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferQueue.java new file mode 100644 index 00000000..655c6967 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableBufferQueue.java @@ -0,0 +1,74 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Queue; +import java.util.concurrent.LinkedTransferQueue; + +public class ReusableBufferQueue implements IReusableBufferDisposedListener +{ + private final static Logger mLog = LoggerFactory.getLogger(ReusableBufferQueue.class); + private Queue mReusableBufferQueue = new LinkedTransferQueue<>(); + private int mBufferCount = 0; + private String mDebugName; + + public ReusableBufferQueue(String debugName) + { + mDebugName = debugName; + } + + public ReusableBufferQueue() + { + + } + + /** + * Implements buffer disposed listener interface. Disposed (ie user count = 0) buffers will automatically callback + * to this method to indicate when they are disposed. + * + * @param reusableComplexBuffer that has been disposed + */ + @Override + public void disposed(ReusableComplexBuffer reusableComplexBuffer) + { + mReusableBufferQueue.offer(reusableComplexBuffer); + } + + /** + * Returns a reusable buffer from an internal recycling queue, or creates a new buffer if there are currently no + * buffers available for reuse. + * + * @param size of the samples that will be loaded into the buffer + * @return a reusable buffer + */ + public ReusableComplexBuffer getBuffer(int size) + { + ReusableComplexBuffer buffer = mReusableBufferQueue.poll(); + + //Create a new buffer if we currently don't have any buffers to reuse + if(buffer == null) + { + buffer = new ReusableComplexBuffer(this, new float[size]); + mBufferCount++; + mLog.debug("Buffer Created - Count:" + mBufferCount + (mDebugName != null ? " [" + mDebugName + "]" : "")); + } + + return buffer; + } +} diff --git a/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableComplexBuffer.java b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableComplexBuffer.java new file mode 100644 index 00000000..00345429 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/complex/reusable/ReusableComplexBuffer.java @@ -0,0 +1,211 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.sample.complex.reusable; + +import io.github.dsheirer.sample.complex.ComplexBuffer; +import org.apache.commons.lang3.Validate; + +import java.nio.FloatBuffer; +import java.util.concurrent.atomic.AtomicInteger; + +public class ReusableComplexBuffer extends ComplexBuffer +{ + private IReusableBufferDisposedListener mDisposalListener; + private long mTimestamp; + private AtomicInteger mUserCount = new AtomicInteger(); + + /** + * Creates a reusable, timestamped complex buffer using the specified time in milliseconds. + * + * @param disposalListener to be notified when all consumers/users are finished using the buffer + * @param samples of data + * @param timestamp in millis for the buffer + */ + public ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples, long timestamp) + { + super(samples); + + Validate.notNull(disposalListener, "Timestamped Buffer Listener cannot be null"); + mDisposalListener = disposalListener; + mTimestamp = timestamp; + } + + /** + * Constructs a timestamped complex buffer using the current system time in milliseconds. + * + * @param disposalListener to be notified when all consumers are finished using the buffer + * @param samples of data + */ + public ReusableComplexBuffer(IReusableBufferDisposedListener disposalListener, float[] samples) + { + this(disposalListener, samples, System.currentTimeMillis()); + } + + /** + * Creates a copy of the samples from this buffer + */ + public float[] getSamplesCopy() + { + float[] samples = getSamples(); + float[] copy = new float[samples.length]; + System.arraycopy(samples, 0, copy, 0, samples.length); + return copy; + } + + /** + * Number of complex samples contained in this buffer + */ + public int getSampleLength() + { + return getSamples().length / 2; + } + + /** + * Reference timestamp for this complex buffer. Timestamp is relative to the first sample in the buffer. + * @return timestamp as milliseconds since epoch + */ + public long getTimestamp() + { + return mTimestamp; + } + + /** + * Reloads this buffer by copying the sample array into this buffer. Internal buffer length will be automatically + * resized to fit the new sample array length. + * + * This is convenient if the source of the sample array originates via a native buffer. + * @param samples to copy into this buffer + * @param timestamp for the samples in millis + */ + public void reloadFrom(float[] samples, long timestamp) + { + if(mUserCount.get() > 0) + { + throw new IllegalStateException("New data cannot be loaded into this reusable buffer while the user count " + + "is above zero - user count:" + mUserCount.get()); + } + + mUserCount.set(0); + + if(mSamples.length != samples.length) + { + mSamples = new float[samples.length]; + } + + System.arraycopy(samples, 0, mSamples, 0, samples.length); + + mTimestamp = timestamp; + } + + /** + * Reloads this buffer by filling this buffer from the float buffer (ie copying the buffer contents). Internal + * buffer length will be automatically resized to accommodate the float buffer's capacity/size. + * + * @param floatBuffer to copy into this buffer + * @param timestamp for the samples in millis + */ + public void reloadFrom(FloatBuffer floatBuffer, long timestamp) + { + if(mUserCount.get() > 0) + { + throw new IllegalStateException("New data cannot be loaded into this reusable buffer while the user count " + + "is above zero - user count:" + mUserCount.get()); + } + + mUserCount.set(0); + + if(mSamples.length != floatBuffer.capacity()) + { + mSamples = new float[floatBuffer.capacity()]; + } + + floatBuffer.rewind(); + floatBuffer.get(mSamples); + + mTimestamp = timestamp; + } + + /** + * 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 + */ + public void setUserCount(int userCount) + { + mUserCount.set(userCount); + } + + /** + * Decrements the user count for this buffer to indicate that the user/receiver of this buffer is finished using + * the buffer. + * + * User count is established prior to sending this buffer to each listener and each listener invokes this method to + * indicate that they have finished processing the buffer so that when the user count reaches zero, this buffer + * can be reused. + * + * This method is thread-safe. + */ + public ReusableComplexBuffer decrementUserCount() + { + int currentCount = mUserCount.decrementAndGet(); + + if(currentCount == 0) + { + mDisposalListener.disposed(this); + } + else if(currentCount < 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"); + } + + return this; + } + + /** + * Increments the user count to indicate that this buffer will be sent to another user. + * + * Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected + * user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is + * finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse. + * + * This method is thread-safe. + */ + public ReusableComplexBuffer incrementUserCount() + { + mUserCount.incrementAndGet(); + return this; + } + + /** + * Increments the user count by the specified count to indicate that this buffer will be sent to additional users. + * + * Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected + * user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is + * finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse. + * + * This method is thread-safe. + * + * @param additionalUserCount to add to the current user count + */ + public ReusableComplexBuffer incrementUserCount(int additionalUserCount) + { + mUserCount.addAndGet(additionalUserCount); + return this; + } +}