From 49a1ea04f8c238d1a94753084af03a05cebdf8fc Mon Sep 17 00:00:00 2001 From: Denny Date: Sun, 8 Jan 2017 04:38:10 -0500 Subject: [PATCH 1/2] #66 wip --- .../channel/ChannelProcessingManager.java | 62 +- src/module/ProcessingChain.java | 1304 ++++++++++------- 2 files changed, 776 insertions(+), 590 deletions(-) diff --git a/src/controller/channel/ChannelProcessingManager.java b/src/controller/channel/ChannelProcessingManager.java index b48fee23..9b0e9df1 100644 --- a/src/controller/channel/ChannelProcessingManager.java +++ b/src/controller/channel/ChannelProcessingManager.java @@ -207,37 +207,6 @@ public class ChannelProcessingManager implements ChannelEventListener MessageActivityModel messageModel = new MessageActivityModel(messageFilter); processingChain.setMessageActivityModel(messageModel); - /* Setup event logging */ - List loggers = mEventLogManager.getLoggers( - channel.getEventLogConfiguration(), channel.getName()); - - if (!loggers.isEmpty()) - { - processingChain.addModules(loggers); - } - - /* Setup recorders */ - List recorders = channel.getRecordConfiguration().getRecorders(); - - if (!recorders.isEmpty()) - { - /* Add baseband recorder */ - if ((recorders.contains(RecorderType.BASEBAND) && - channel.getChannelType() == ChannelType.STANDARD)) - { - processingChain.addModule(mRecorderManager.getBasebandRecorder( - channel.toString())); - } - - /* Add traffic channel baseband recorder */ - if (recorders.contains(RecorderType.TRAFFIC_BASEBAND) && - channel.getChannelType() == ChannelType.TRAFFIC) - { - processingChain.addModule(mRecorderManager - .getBasebandRecorder(channel.toString())); - } - } - /* Inject channel metadata that will be inserted into audio packets * for the recorder manager and streaming */ processingChain.broadcast( @@ -248,6 +217,37 @@ public class ChannelProcessingManager implements ChannelEventListener new Metadata(MetadataType.CHANNEL_NAME, channel.getName())); } + /* Setup event logging */ + List loggers = mEventLogManager.getLoggers( + channel.getEventLogConfiguration(), channel.getName()); + + if (!loggers.isEmpty()) + { + processingChain.addModules(loggers); + } + + /* Setup recorders */ + List recorders = channel.getRecordConfiguration().getRecorders(); + + if (!recorders.isEmpty()) + { + /* Add baseband recorder */ + if ((recorders.contains(RecorderType.BASEBAND) && + channel.getChannelType() == ChannelType.STANDARD)) + { + processingChain.addModule(mRecorderManager.getBasebandRecorder( + channel.toString())); + } + + /* Add traffic channel baseband recorder */ + if (recorders.contains(RecorderType.TRAFFIC_BASEBAND) && + channel.getChannelType() == ChannelType.TRAFFIC) + { + processingChain.addModule(mRecorderManager + .getBasebandRecorder(channel.toString())); + } + } + processingChain.setSource(source); if (event instanceof TrafficChannelEvent) diff --git a/src/module/ProcessingChain.java b/src/module/ProcessingChain.java index bcc564fc..298ba1e3 100644 --- a/src/module/ProcessingChain.java +++ b/src/module/ProcessingChain.java @@ -1,28 +1,36 @@ /******************************************************************************* * SDR Trunk * Copyright (C) 2015-2016 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 module; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.atomic.AtomicBoolean; - +import audio.AudioPacket; +import audio.IAudioPacketListener; +import audio.IAudioPacketProvider; +import audio.metadata.IMetadataListener; +import audio.metadata.IMetadataProvider; +import audio.metadata.Metadata; +import audio.squelch.ISquelchStateListener; +import audio.squelch.ISquelchStateProvider; +import audio.squelch.SquelchState; +import controller.NamingThreadFactory; +import controller.channel.Channel.ChannelType; +import controller.channel.ChannelEvent; +import controller.channel.IChannelEventListener; +import controller.channel.IChannelEventProvider; import message.IMessageListener; import message.IMessageProvider; import message.Message; @@ -38,10 +46,11 @@ import module.decode.state.DecoderStateEvent.Event; import module.decode.state.IDecoderStateEventListener; import module.decode.state.IDecoderStateEventProvider; import module.decode.state.State; - +import module.log.EventLogger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - +import record.wave.ComplexBufferWaveRecorder; +import record.wave.RealBufferWaveRecorder; import sample.Broadcaster; import sample.Listener; import sample.complex.ComplexBuffer; @@ -59,604 +68,781 @@ import source.tuner.TunerChannelSource; import source.tuner.frequency.FrequencyChangeEvent; import source.tuner.frequency.IFrequencyChangeListener; import source.tuner.frequency.IFrequencyChangeProvider; -import audio.AudioPacket; -import audio.IAudioPacketListener; -import audio.IAudioPacketProvider; -import audio.metadata.IMetadataListener; -import audio.metadata.IMetadataProvider; -import audio.metadata.Metadata; -import audio.squelch.ISquelchStateListener; -import audio.squelch.ISquelchStateProvider; -import audio.squelch.SquelchState; -import controller.NamingThreadFactory; -import controller.channel.Channel.ChannelType; -import controller.channel.ChannelEvent; -import controller.channel.IChannelEventListener; -import controller.channel.IChannelEventProvider; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; /** * Processing chain provides a framework for connecting a complex or real sample - * source to a set of one primary decoder and zero or more auxiliary decoders. + * source to a set of one primary decoder and zero or more auxiliary decoders. * All decoded messages and call events produced by the decoders and the decoder * call states are aggregated by the various broadcasters. You can register * listeners to receive aggregated messages, call events, and audio packets. - * + * * Normal setup sequence: - * + * * 1) Add one or more modules * 2) Register listeners to receive messages, call events, audio, etc. * 3) Add a valid source * 4) Invoke the start() method to start processing. * 5) Invoke the stop() method to stop processing. - * + * * Optional: if you want to reuse the processing chain with a new sample source, * invoke the following method sequence: stop(), setSource(), start() */ public class ProcessingChain implements IChannelEventListener { - private final static Logger mLog = LoggerFactory.getLogger( ProcessingChain.class ); + private final static Logger mLog = LoggerFactory.getLogger(ProcessingChain.class); - private Broadcaster mAudioPacketBroadcaster = new Broadcaster<>(); - private Broadcaster mMetadataBroadcaster = new Broadcaster<>(); - private Broadcaster mCallEventBroadcaster = new Broadcaster<>(); - private Broadcaster mChannelEventBroadcaster = new Broadcaster<>(); - private Broadcaster mComplexBufferBroadcaster = new Broadcaster<>(); - private Broadcaster mDecoderStateEventBroadcaster = new Broadcaster<>(); - private Broadcaster mFrequencyChangeEventBroadcaster = new Broadcaster<>(); - private Broadcaster mMessageBroadcaster = new Broadcaster<>(); - private Broadcaster mFilteredRealBufferBroadcaster = new Broadcaster<>(); - private Broadcaster mUnFilteredRealBufferBroadcaster = new Broadcaster<>(); - private Broadcaster mSquelchStateBroadcaster = new Broadcaster<>(); - - private ScheduledExecutorService mScheduledExecutorService; - private String mName; - private AtomicBoolean mRunning = new AtomicBoolean(); - - protected Source mSource; - private List mModules = new ArrayList<>(); - private CallEventModel mCallEventModel; - private ChannelState mChannelState; - private MessageActivityModel mMessageActivityModel; + private Broadcaster mAudioPacketBroadcaster = new Broadcaster<>(); + private Broadcaster mMetadataBroadcaster = new Broadcaster<>(); + private Broadcaster mCallEventBroadcaster = new Broadcaster<>(); + private Broadcaster mChannelEventBroadcaster = new Broadcaster<>(); + private Broadcaster mComplexBufferBroadcaster = new Broadcaster<>(); + private Broadcaster mDecoderStateEventBroadcaster = new Broadcaster<>(); + private Broadcaster mFrequencyChangeEventBroadcaster = new Broadcaster<>(); + private Broadcaster mMessageBroadcaster = new Broadcaster<>(); + private Broadcaster mFilteredRealBufferBroadcaster = new Broadcaster<>(); + private Broadcaster mUnFilteredRealBufferBroadcaster = new Broadcaster<>(); + private Broadcaster mSquelchStateBroadcaster = new Broadcaster<>(); - /** - * Creates a processing chain for managing a set of modules - * - * @param name for thread pool naming each thread - * @param channelType - */ - public ProcessingChain( String name, ChannelType channelType ) - { - mName = name; - - mChannelState = new ChannelState( channelType ); - addModule( mChannelState ); + private ScheduledExecutorService mScheduledExecutorService; + private String mName; + private AtomicBoolean mRunning = new AtomicBoolean(); - mCallEventModel = new CallEventModel(); - addCallEventListener( mCallEventModel ); - } - - public CallEventModel getCallEventModel() - { - return mCallEventModel; - } - - public ChannelState getChannelState() - { - return mChannelState; - } - - public MessageActivityModel getMessageActivityModel() - { - return mMessageActivityModel; - } - - public void setMessageActivityModel( MessageActivityModel model ) - { - mMessageActivityModel = model; - - addMessageListener( mMessageActivityModel ); - } - - public void dispose() - { - stop(); - - for( Module module: mModules ) - { - module.dispose(); - } - - mModules.clear(); - - mScheduledExecutorService.shutdownNow(); - mScheduledExecutorService = null; - - mAudioPacketBroadcaster.dispose(); - mCallEventBroadcaster.dispose(); - mChannelEventBroadcaster.dispose(); - mComplexBufferBroadcaster.dispose(); - mMessageBroadcaster.dispose(); - mFilteredRealBufferBroadcaster.dispose(); - mSquelchStateBroadcaster.dispose(); - - if( mScheduledExecutorService != null ) - { - mScheduledExecutorService.shutdownNow(); - } - } + protected Source mSource; + private List mModules = new ArrayList<>(); + private CallEventModel mCallEventModel; + private ChannelState mChannelState; + private MessageActivityModel mMessageActivityModel; - /** - * Broadcasts the metadata to any registered listeners - */ - public void broadcast( Metadata metadata ) - { - if( mMetadataBroadcaster != null ) - { - mMetadataBroadcaster.broadcast( metadata ); - } - } - - /** - * Indicates if this processing chain is currently receiving samples from - * a source and sending those samples to the decoders. - */ - public boolean isProcessing() - { - return mRunning.get(); - } + /** + * Creates a processing chain for managing a set of modules + * + * @param name for thread pool naming each thread + * @param channelType + */ + public ProcessingChain(String name, ChannelType channelType) + { + mName = name; - /** - * Indicates if this chain currently has a valid sample source. - */ - public boolean hasSource() - { - return mSource != null; - } + mChannelState = new ChannelState(channelType); + addModule(mChannelState); - /** - * Applies a sample source to this processing chain. Processing won't - * start until the start() method is invoked. - * - * @param source - real or complex sample source - * - * @throws IllegalStateException if the processing chain is currently - * processing with another source. Invoke stop() before applying a new - * source. - */ - public void setSource( Source source ) throws IllegalStateException - { - if( isProcessing() ) - { - throw new IllegalStateException( "Processing chain is currently " - + "processing. Invoke stop() on the processing chain before " - + "applying a new sample source" ); - } - - mSource = source; - } + mCallEventModel = new CallEventModel(); + addCallEventListener(mCallEventModel); + } - /** - * List of current modules for this processing chain - */ - public List getModules() - { - return mModules; - } + public CallEventModel getCallEventModel() + { + return mCallEventModel; + } - /** - * List of decoder states for this processing chain - */ - public List getDecoderStates() - { - List decoderStates = new ArrayList<>(); - - for( Module module: mModules ) - { - if( module instanceof DecoderState ) - { - decoderStates.add( (DecoderState)module ); - } - } - - return decoderStates; - } - - /** - * Adds the list of modules to this processing chain - */ - public void addModules( List modules ) - { - for( Module module: modules ) - { - addModule( module ); - } - } + public ChannelState getChannelState() + { + return mChannelState; + } - /** - * Adds a module to the processing chain. Each module is tested for the - * interfaces that it supports and is registered or receives a listener - * to consume or produce the supported interface data type. All elements - * and events that are produced by any module are automatically routed to - * all other components that support the corresponding listener interface. - * - * At least one module should consume complex samples and either produce - * decoded messages and/or audio, or produce decoded real sample buffers - * for all other modules to consume. - * - * @param module - processing module, demodulator, decoder, source, state - * machine, etc. - */ - public void addModule( Module module ) - { - mModules.add( module ); - - registerListeners( module ); - registerProviders( module ); - } + public MessageActivityModel getMessageActivityModel() + { + return mMessageActivityModel; + } - /** - * Registers the module as a listener to each of the broadcasters that - * provide the data interface(s) supported by the module. - */ - private void registerListeners( Module module ) - { - if( module instanceof IAudioPacketListener ) - { - mAudioPacketBroadcaster.addListener( - ((IAudioPacketListener)module).getAudioPacketListener() ); - } - - if( module instanceof ICallEventListener ) - { - mCallEventBroadcaster.addListener( - ((ICallEventListener)module).getCallEventListener() ); - } - - if( module instanceof IChannelEventListener ) - { - mChannelEventBroadcaster.addListener( - ((IChannelEventListener)module).getChannelEventListener() ); - } - - if( module instanceof IComplexBufferListener ) - { - mComplexBufferBroadcaster.addListener( - ((IComplexBufferListener)module).getComplexBufferListener() ); - } - - if( module instanceof IDecoderStateEventListener ) - { - mDecoderStateEventBroadcaster.addListener( - ((IDecoderStateEventListener)module).getDecoderStateListener() ); - } + public void setMessageActivityModel(MessageActivityModel model) + { + mMessageActivityModel = model; - if( module instanceof IFrequencyChangeListener ) - { - mFrequencyChangeEventBroadcaster.addListener( - ((IFrequencyChangeListener)module).getFrequencyChangeListener()); - } - - if( module instanceof IMessageListener ) - { - mMessageBroadcaster.addListener( - ((IMessageListener)module).getMessageListener() ); - } - - if( module instanceof IMetadataListener ) - { - mMetadataBroadcaster.addListener( - ((IMetadataListener)module).getMetadataListener() ); - } - - if( module instanceof IFilteredRealBufferListener ) - { - mFilteredRealBufferBroadcaster.addListener( - ((IFilteredRealBufferListener)module).getFilteredRealBufferListener() ); - } + addMessageListener(mMessageActivityModel); + } - if( module instanceof ISquelchStateListener ) - { - mSquelchStateBroadcaster.addListener( - ((ISquelchStateListener)module).getSquelchStateListener() ); - } + public void dispose() + { + stop(); - if( module instanceof IUnFilteredRealBufferListener ) - { - mUnFilteredRealBufferBroadcaster.addListener( - ((IUnFilteredRealBufferListener)module).getUnFilteredRealBufferListener() ); - } - } - - /** - * Registers the broadcaster(s) as listeners to the module for each - * provider interface that is supported by the module. - */ - private void registerProviders( Module module ) - { - if( module instanceof IAudioPacketProvider ) - { - ((IAudioPacketProvider)module).setAudioPacketListener( - mAudioPacketBroadcaster ); - } - - if( module instanceof ICallEventProvider ) - { - ((ICallEventProvider)module).addCallEventListener( mCallEventBroadcaster ); - } + for(Module module : mModules) + { + module.dispose(); + } - if( module instanceof IChannelEventProvider ) - { - ((IChannelEventProvider)module).setChannelEventListener( - mChannelEventBroadcaster ); - } - - if( module instanceof IDecoderStateEventProvider ) - { - ((IDecoderStateEventProvider)module).setDecoderStateListener( - mDecoderStateEventBroadcaster ); - } + mModules.clear(); - if( module instanceof IFrequencyChangeProvider ) - { - ((IFrequencyChangeProvider)module).setFrequencyChangeListener( - mFrequencyChangeEventBroadcaster ); - } - - if( module instanceof IMessageProvider ) - { - ((IMessageProvider)module).setMessageListener( mMessageBroadcaster ); - } - - if( module instanceof IMetadataProvider ) - { - ((IMetadataProvider)module).setMetadataListener( - mMetadataBroadcaster ); - } + mScheduledExecutorService.shutdownNow(); + mScheduledExecutorService = null; - if( module instanceof IFilteredRealBufferProvider ) - { - ((IFilteredRealBufferProvider)module) - .setFilteredRealBufferListener( mFilteredRealBufferBroadcaster ); - } - - if( module instanceof ISquelchStateProvider ) - { - ((ISquelchStateProvider)module).setSquelchStateListener( - mSquelchStateBroadcaster ); - } + mAudioPacketBroadcaster.dispose(); + mCallEventBroadcaster.dispose(); + mChannelEventBroadcaster.dispose(); + mComplexBufferBroadcaster.dispose(); + mMessageBroadcaster.dispose(); + mFilteredRealBufferBroadcaster.dispose(); + mSquelchStateBroadcaster.dispose(); - if( module instanceof IUnFilteredRealBufferProvider ) - { - ((IUnFilteredRealBufferProvider)module) - .setUnFilteredRealBufferListener( mUnFilteredRealBufferBroadcaster ); - } - } - - /** - * Starts processing if the chain has a valid source. Invocations on an - * already started chain have no effect. - */ - public void start() - { - if( mRunning.compareAndSet( false, true ) ) - { - if( mSource != null ) - { - /* Reset each of the modules */ - for( Module module: mModules ) - { - module.reset(); - } + if(mScheduledExecutorService != null) + { + mScheduledExecutorService.shutdownNow(); + } + } + + /** + * Broadcasts the metadata to any registered listeners + */ + public void broadcast(Metadata metadata) + { + if(mMetadataBroadcaster != null) + { + mMetadataBroadcaster.broadcast(metadata); + } + } + + /** + * Indicates if this processing chain is currently receiving samples from + * a source and sending those samples to the decoders. + */ + public boolean isProcessing() + { + return mRunning.get(); + } + + /** + * Indicates if this chain currently has a valid sample source. + */ + public boolean hasSource() + { + return mSource != null; + } + + /** + * Applies a sample source to this processing chain. Processing won't + * start until the start() method is invoked. + * + * @param source - real or complex sample source + * @throws IllegalStateException if the processing chain is currently + * processing with another source. Invoke stop() before applying a new + * source. + */ + public void setSource(Source source) throws IllegalStateException + { + if(isProcessing()) + { + throw new IllegalStateException("Processing chain is currently " + + "processing. Invoke stop() on the processing chain before " + + "applying a new sample source"); + } + + mSource = source; + } + + /** + * List of current modules for this processing chain + */ + public List getModules() + { + return mModules; + } + + /** + * List of decoder states for this processing chain + */ + public List getDecoderStates() + { + List decoderStates = new ArrayList<>(); + + for(Module module : mModules) + { + if(module instanceof DecoderState) + { + decoderStates.add((DecoderState) module); + } + } + + return decoderStates; + } + + /** + * Adds the list of modules to this processing chain + */ + public void addModules(List modules) + { + for(Module module : modules) + { + addModule(module); + } + } + + /** + * Adds a module to the processing chain. Each module is tested for the + * interfaces that it supports and is registered or receives a listener + * to consume or produce the supported interface data type. All elements + * and events that are produced by any module are automatically routed to + * all other components that support the corresponding listener interface. + * + * At least one module should consume complex samples and either produce + * decoded messages and/or audio, or produce decoded real sample buffers + * for all other modules to consume. + * + * @param module - processing module, demodulator, decoder, source, state + * machine, etc. + */ + public void addModule(Module module) + { + mModules.add(module); + + registerListeners(module); + registerProviders(module); + } + + /** + * Removes the module from the processing chain and deregisters the module as a listener and + * as a provider + */ + public void removeModule(Module module) + { + unregisterListeners(module); + unregisterProviders(module); + + mModules.remove(module); + } + + /** + * Registers the module as a listener to each of the broadcasters that + * provide the data interface(s) supported by the module. + */ + private void registerListeners(Module module) + { + if(module instanceof IAudioPacketListener) + { + mAudioPacketBroadcaster.addListener( + ((IAudioPacketListener) module).getAudioPacketListener()); + } + + if(module instanceof ICallEventListener) + { + mCallEventBroadcaster.addListener( + ((ICallEventListener) module).getCallEventListener()); + } + + if(module instanceof IChannelEventListener) + { + mChannelEventBroadcaster.addListener( + ((IChannelEventListener) module).getChannelEventListener()); + } + + if(module instanceof IComplexBufferListener) + { + mComplexBufferBroadcaster.addListener( + ((IComplexBufferListener) module).getComplexBufferListener()); + } + + if(module instanceof IDecoderStateEventListener) + { + mDecoderStateEventBroadcaster.addListener( + ((IDecoderStateEventListener) module).getDecoderStateListener()); + } + + if(module instanceof IFrequencyChangeListener) + { + mFrequencyChangeEventBroadcaster.addListener( + ((IFrequencyChangeListener) module).getFrequencyChangeListener()); + } + + if(module instanceof IMessageListener) + { + mMessageBroadcaster.addListener( + ((IMessageListener) module).getMessageListener()); + } + + if(module instanceof IMetadataListener) + { + mMetadataBroadcaster.addListener( + ((IMetadataListener) module).getMetadataListener()); + } + + if(module instanceof IFilteredRealBufferListener) + { + mFilteredRealBufferBroadcaster.addListener( + ((IFilteredRealBufferListener) module).getFilteredRealBufferListener()); + } + + if(module instanceof ISquelchStateListener) + { + mSquelchStateBroadcaster.addListener( + ((ISquelchStateListener) module).getSquelchStateListener()); + } + + if(module instanceof IUnFilteredRealBufferListener) + { + mUnFilteredRealBufferBroadcaster.addListener( + ((IUnFilteredRealBufferListener) module).getUnFilteredRealBufferListener()); + } + } + + /** + * Registers the module as a listener to each of the broadcasters that + * provide the data interface(s) supported by the module. + */ + private void unregisterListeners(Module module) + { + if(module instanceof IAudioPacketListener) + { + mAudioPacketBroadcaster.removeListener( + ((IAudioPacketListener) module).getAudioPacketListener()); + } + + if(module instanceof ICallEventListener) + { + mCallEventBroadcaster.removeListener( + ((ICallEventListener) module).getCallEventListener()); + } + + if(module instanceof IChannelEventListener) + { + mChannelEventBroadcaster.removeListener( + ((IChannelEventListener) module).getChannelEventListener()); + } + + if(module instanceof IComplexBufferListener) + { + mComplexBufferBroadcaster.removeListener( + ((IComplexBufferListener) module).getComplexBufferListener()); + } + + if(module instanceof IDecoderStateEventListener) + { + mDecoderStateEventBroadcaster.removeListener( + ((IDecoderStateEventListener) module).getDecoderStateListener()); + } + + if(module instanceof IFrequencyChangeListener) + { + mFrequencyChangeEventBroadcaster.removeListener( + ((IFrequencyChangeListener) module).getFrequencyChangeListener()); + } + + if(module instanceof IMessageListener) + { + mMessageBroadcaster.removeListener(((IMessageListener) module).getMessageListener()); + } + + if(module instanceof IMetadataListener) + { + mMetadataBroadcaster.removeListener(((IMetadataListener) module).getMetadataListener()); + } + + if(module instanceof IFilteredRealBufferListener) + { + mFilteredRealBufferBroadcaster.removeListener( + ((IFilteredRealBufferListener) module).getFilteredRealBufferListener()); + } + + if(module instanceof ISquelchStateListener) + { + mSquelchStateBroadcaster.removeListener( + ((ISquelchStateListener) module).getSquelchStateListener()); + } + + if(module instanceof IUnFilteredRealBufferListener) + { + mUnFilteredRealBufferBroadcaster.removeListener( + ((IUnFilteredRealBufferListener) module).getUnFilteredRealBufferListener()); + } + } + + /** + * Registers the broadcaster(s) as listeners to the module for each + * provider interface that is supported by the module. + */ + private void registerProviders(Module module) + { + if(module instanceof IAudioPacketProvider) + { + ((IAudioPacketProvider) module).setAudioPacketListener( + mAudioPacketBroadcaster); + } + + if(module instanceof ICallEventProvider) + { + ((ICallEventProvider) module).addCallEventListener(mCallEventBroadcaster); + } + + if(module instanceof IChannelEventProvider) + { + ((IChannelEventProvider) module).setChannelEventListener( + mChannelEventBroadcaster); + } + + if(module instanceof IDecoderStateEventProvider) + { + ((IDecoderStateEventProvider) module).setDecoderStateListener( + mDecoderStateEventBroadcaster); + } + + if(module instanceof IFrequencyChangeProvider) + { + ((IFrequencyChangeProvider) module).setFrequencyChangeListener( + mFrequencyChangeEventBroadcaster); + } + + if(module instanceof IMessageProvider) + { + ((IMessageProvider) module).setMessageListener(mMessageBroadcaster); + } + + if(module instanceof IMetadataProvider) + { + ((IMetadataProvider) module).setMetadataListener( + mMetadataBroadcaster); + } + + if(module instanceof IFilteredRealBufferProvider) + { + ((IFilteredRealBufferProvider) module) + .setFilteredRealBufferListener(mFilteredRealBufferBroadcaster); + } + + if(module instanceof ISquelchStateProvider) + { + ((ISquelchStateProvider) module).setSquelchStateListener( + mSquelchStateBroadcaster); + } + + if(module instanceof IUnFilteredRealBufferProvider) + { + ((IUnFilteredRealBufferProvider) module) + .setUnFilteredRealBufferListener(mUnFilteredRealBufferBroadcaster); + } + } + + /** + * Unregisters the broadcaster(s) as listeners to the module for each + * provider interface that is supported by the module. + */ + private void unregisterProviders(Module module) + { + if(module instanceof IAudioPacketProvider) + { + ((IAudioPacketProvider) module).setAudioPacketListener(null); + } + + if(module instanceof ICallEventProvider) + { + ((ICallEventProvider) module).removeCallEventListener(mCallEventBroadcaster); + } + + if(module instanceof IChannelEventProvider) + { + ((IChannelEventProvider) module).setChannelEventListener(null); + } + + if(module instanceof IDecoderStateEventProvider) + { + ((IDecoderStateEventProvider) module).setDecoderStateListener(null); + } + + if(module instanceof IFrequencyChangeProvider) + { + ((IFrequencyChangeProvider) module).setFrequencyChangeListener(null); + } + + if(module instanceof IMessageProvider) + { + ((IMessageProvider) module).setMessageListener(null); + } + + if(module instanceof IMetadataProvider) + { + ((IMetadataProvider) module).setMetadataListener(null); + } + + if(module instanceof IFilteredRealBufferProvider) + { + ((IFilteredRealBufferProvider) module).setFilteredRealBufferListener(null); + } + + if(module instanceof ISquelchStateProvider) + { + ((ISquelchStateProvider) module).setSquelchStateListener(null); + } + + if(module instanceof IUnFilteredRealBufferProvider) + { + ((IUnFilteredRealBufferProvider) module).setUnFilteredRealBufferListener(null); + } + } + + /** + * Starts processing if the chain has a valid source. Invocations on an + * already started chain have no effect. + */ + public void start() + { + if(mRunning.compareAndSet(false, true)) + { + if(mSource != null) + { + /* Reset each of the modules */ + for(Module module : mModules) + { + module.reset(); + } /* Register with the source to receive sample data. Setup a * timer task to process the buffer queues 50 times a second * (every 20 ms) */ - switch( mSource.getSampleType() ) - { - case COMPLEX: - ((ComplexSource)mSource).setListener( mComplexBufferBroadcaster ); - break; - case REAL: - ((RealSource)mSource).setListener( mFilteredRealBufferBroadcaster ); - break; - default: - throw new IllegalArgumentException( "Unrecognized source " - + "sample type - cannot start processing chain" ); - } + switch(mSource.getSampleType()) + { + case COMPLEX: + ((ComplexSource) mSource).setListener(mComplexBufferBroadcaster); + break; + case REAL: + ((RealSource) mSource).setListener(mFilteredRealBufferBroadcaster); + break; + default: + throw new IllegalArgumentException("Unrecognized source " + + "sample type - cannot start processing chain"); + } /* If this is a tuner source, broadcast the frequency to all * of the decoder state's and start the samples flowing */ - if( mSource instanceof TunerChannelSource ) - { - TunerChannelSource tcs = (TunerChannelSource)mSource; + if(mSource instanceof TunerChannelSource) + { + TunerChannelSource tcs = (TunerChannelSource) mSource; - try - { - long frequency = tcs.getFrequency(); + try + { + long frequency = tcs.getFrequency(); - mDecoderStateEventBroadcaster.broadcast( - new DecoderStateEvent( this, Event.SOURCE_FREQUENCY, - State.IDLE, frequency ) ); - } - catch( SourceException e ) - { - mLog.error( "Error getting frequency from tuner channel source", e ); - } - } - - if( mScheduledExecutorService == null ) - { - mScheduledExecutorService = Executors.newScheduledThreadPool( - 1, new NamingThreadFactory( "channel " + mName ) ); - } + mDecoderStateEventBroadcaster.broadcast( + new DecoderStateEvent(this, Event.SOURCE_FREQUENCY, + State.IDLE, frequency)); + } + catch(SourceException e) + { + mLog.error("Error getting frequency from tuner channel source", e); + } + } + + if(mScheduledExecutorService == null) + { + mScheduledExecutorService = Executors.newScheduledThreadPool( + 1, new NamingThreadFactory("channel " + mName)); + } /* Start each of the modules */ - for( Module module: mModules ) - { - try - { - module.start( mScheduledExecutorService ); - } - catch( Exception e ) - { - mLog.error( "Error starting module", e ); - } - } + for(Module module : mModules) + { + try + { + module.start(mScheduledExecutorService); + } + catch(Exception e) + { + mLog.error("Error starting module", e); + } + } - mSource.start( mScheduledExecutorService ); - } - else - { - mLog.error( "Source is null on start()" ); - } - } - } + mSource.start(mScheduledExecutorService); + } + else + { + mLog.error("Source is null on start()"); + } + } + } - /** - * Stops processing if the chain is currently processing. Invocations on - * an already stopped chain have no effect. - */ - public void stop() - { - if( mRunning.compareAndSet( true, false ) ) - { - if( mSource != null ) - { - mSource.stop(); - - switch( mSource.getSampleType() ) - { - case COMPLEX: - ((ComplexSource)mSource).removeListener( mComplexBufferBroadcaster ); - break; - case REAL: - ((RealSource)mSource).removeListener( mFilteredRealBufferBroadcaster ); - break; - default: - throw new IllegalArgumentException( "Unrecognized source " - + "sample type - cannot start processing chain" ); - } + /** + * Stops processing if the chain is currently processing. Invocations on + * an already stopped chain have no effect. + */ + public void stop() + { + if(mRunning.compareAndSet(true, false)) + { + if(mSource != null) + { + mSource.stop(); + + switch(mSource.getSampleType()) + { + case COMPLEX: + ((ComplexSource) mSource).removeListener(mComplexBufferBroadcaster); + break; + case REAL: + ((RealSource) mSource).removeListener(mFilteredRealBufferBroadcaster); + break; + default: + throw new IllegalArgumentException("Unrecognized source " + + "sample type - cannot start processing chain"); + } /* Release the source */ - mSource.dispose(); - mSource = null; - } + mSource.dispose(); + mSource = null; + } /* Stop each of the modules */ - for( Module module: mModules ) - { - module.stop(); - } - } - } - - /** - * Adds the listener to receive audio packets from all modules. - */ - public void addAudioPacketListener( Listener listener ) - { - mAudioPacketBroadcaster.addListener( listener ); - } - - public void removeAudioPacketListener( Listener listener ) - { - mAudioPacketBroadcaster.removeListener( listener ); - } - - /** - * Adds the listener to receive call events from all modules. - */ - public void addCallEventListener( Listener listener ) - { - mCallEventBroadcaster.addListener( listener ); - } - - public void removeCallEventListener( Listener listener ) - { - mCallEventBroadcaster.removeListener( listener ); - } - - /** - * Adds the listener to receive call events from all modules. - */ - public void addChannelEventListener( Listener listener ) - { - mChannelEventBroadcaster.addListener( listener ); - } - - public void removeChannelEventListener( Listener listener ) - { - mChannelEventBroadcaster.removeListener( listener ); - } + for(Module module : mModules) + { + module.stop(); + } - /** - * Adds the listener to receive decoder state events from decoder modules - */ - public void addDecoderStateEventListener( Listener listener ) - { - mDecoderStateEventBroadcaster.addListener( listener ); - } - - public void removeDecoderStateEventListener( Listener listener) - { - mDecoderStateEventBroadcaster.removeListener( listener ); - } - - public Listener getDecoderStateEventListener() - { - return mDecoderStateEventBroadcaster; - } - - /** - * Adds the listener to receive decoded messages from all decoders. - */ - public void addMessageListener( Listener listener ) - { - mMessageBroadcaster.addListener( listener ); - } - - /** - * Adds the list of listeners to receive decoded messages from all decoders. - */ - public void addMessageListeners( List> listeners ) - { - for( Listener listener: listeners ) - { - mMessageBroadcaster.addListener( listener ); - } - } - - public void removeMessageListener( Listener listener ) - { - mMessageBroadcaster.removeListener( listener ); - } - - /** - * Adds the listener to receive call events from all modules. - */ - public void addSquelchStateListener( Listener listener ) - { - mSquelchStateBroadcaster.addListener( listener ); - } - - public void removeSquelchStateListener( Listener listener ) - { - mSquelchStateBroadcaster.removeListener( listener ); - } - - public void addRealBufferListener( Listener listener ) - { - mFilteredRealBufferBroadcaster.addListener( listener ); - } - - public void removeRealBufferListener( Listener listener ) - { - mFilteredRealBufferBroadcaster.removeListener( listener ); - } - - @Override - public Listener getChannelEventListener() - { - return mChannelEventBroadcaster; - } + removeEventLoggingModules(); + + removeRecordingModules(); + } + } + + /** + * Removes any logging modules that are currently registered with this processing chain + */ + private void removeEventLoggingModules() + { + List eventLoggingModules = new ArrayList<>(); + + for(Module module: mModules) + { + if(module instanceof EventLogger) + { + eventLoggingModules.add(module); + } + } + + for(Module eventLoggingModule: eventLoggingModules) + { + removeModule(eventLoggingModule); + } + } + + /** + * Removes any recording modules that are currently registered with this processing chain + */ + private void removeRecordingModules() + { + List recordingModules = new ArrayList<>(); + + for(Module module: mModules) + { + if(module instanceof RealBufferWaveRecorder || module instanceof ComplexBufferWaveRecorder) + { + recordingModules.add(module); + } + } + + for(Module recordingModule: recordingModules) + { + removeModule(recordingModule); + } + } + + /** + * Adds the listener to receive audio packets from all modules. + */ + public void addAudioPacketListener(Listener listener) + { + mAudioPacketBroadcaster.addListener(listener); + } + + public void removeAudioPacketListener(Listener listener) + { + mAudioPacketBroadcaster.removeListener(listener); + } + + /** + * Adds the listener to receive call events from all modules. + */ + public void addCallEventListener(Listener listener) + { + mCallEventBroadcaster.addListener(listener); + } + + public void removeCallEventListener(Listener listener) + { + mCallEventBroadcaster.removeListener(listener); + } + + /** + * Adds the listener to receive call events from all modules. + */ + public void addChannelEventListener(Listener listener) + { + mChannelEventBroadcaster.addListener(listener); + } + + public void removeChannelEventListener(Listener listener) + { + mChannelEventBroadcaster.removeListener(listener); + } + + /** + * Adds the listener to receive decoder state events from decoder modules + */ + public void addDecoderStateEventListener(Listener listener) + { + mDecoderStateEventBroadcaster.addListener(listener); + } + + public void removeDecoderStateEventListener(Listener listener) + { + mDecoderStateEventBroadcaster.removeListener(listener); + } + + public Listener getDecoderStateEventListener() + { + return mDecoderStateEventBroadcaster; + } + + /** + * Adds the listener to receive decoded messages from all decoders. + */ + public void addMessageListener(Listener listener) + { + mMessageBroadcaster.addListener(listener); + } + + /** + * Adds the list of listeners to receive decoded messages from all decoders. + */ + public void addMessageListeners(List> listeners) + { + for(Listener listener : listeners) + { + mMessageBroadcaster.addListener(listener); + } + } + + public void removeMessageListener(Listener listener) + { + mMessageBroadcaster.removeListener(listener); + } + + /** + * Adds the listener to receive call events from all modules. + */ + public void addSquelchStateListener(Listener listener) + { + mSquelchStateBroadcaster.addListener(listener); + } + + public void removeSquelchStateListener(Listener listener) + { + mSquelchStateBroadcaster.removeListener(listener); + } + + public void addRealBufferListener(Listener listener) + { + mFilteredRealBufferBroadcaster.addListener(listener); + } + + public void removeRealBufferListener(Listener listener) + { + mFilteredRealBufferBroadcaster.removeListener(listener); + } + + @Override + public Listener getChannelEventListener() + { + return mChannelEventBroadcaster; + } } From 7865e813dedb70a2c33d2e6fd36292075774d703 Mon Sep 17 00:00:00 2001 From: Denny Date: Sun, 8 Jan 2017 05:21:42 -0500 Subject: [PATCH 2/2] #66 resolves issue with baseband recorders and loggers not being removed from channel processing chains when processing stops. This is now handled in the channel processing manager where baseband recorders and message/event loggers are added at processing start and removed at processing stop. --- .../channel/ChannelProcessingManager.java | 26 +++++++++---------- src/module/ProcessingChain.java | 8 ++---- src/record/RecorderManager.java | 3 ++- 3 files changed, 17 insertions(+), 20 deletions(-) diff --git a/src/controller/channel/ChannelProcessingManager.java b/src/controller/channel/ChannelProcessingManager.java index 9b0e9df1..65aa700e 100644 --- a/src/controller/channel/ChannelProcessingManager.java +++ b/src/controller/channel/ChannelProcessingManager.java @@ -218,8 +218,7 @@ public class ChannelProcessingManager implements ChannelEventListener } /* Setup event logging */ - List loggers = mEventLogManager.getLoggers( - channel.getEventLogConfiguration(), channel.getName()); + List loggers = mEventLogManager.getLoggers(channel.getEventLogConfiguration(), channel.getName()); if (!loggers.isEmpty()) { @@ -232,19 +231,15 @@ public class ChannelProcessingManager implements ChannelEventListener if (!recorders.isEmpty()) { /* Add baseband recorder */ - if ((recorders.contains(RecorderType.BASEBAND) && - channel.getChannelType() == ChannelType.STANDARD)) + if ((recorders.contains(RecorderType.BASEBAND) && channel.getChannelType() == ChannelType.STANDARD)) { - processingChain.addModule(mRecorderManager.getBasebandRecorder( - channel.toString())); + processingChain.addModule(mRecorderManager.getBasebandRecorder(channel.toString())); } /* Add traffic channel baseband recorder */ - if (recorders.contains(RecorderType.TRAFFIC_BASEBAND) && - channel.getChannelType() == ChannelType.TRAFFIC) + if (recorders.contains(RecorderType.TRAFFIC_BASEBAND) && channel.getChannelType() == ChannelType.TRAFFIC) { - processingChain.addModule(mRecorderManager - .getBasebandRecorder(channel.toString())); + processingChain.addModule(mRecorderManager.getBasebandRecorder(channel.toString())); } } @@ -275,9 +270,14 @@ public class ChannelProcessingManager implements ChannelEventListener if (mProcessingChains.containsKey(channel.getChannelID())) { - ProcessingChain chain = mProcessingChains.get(channel.getChannelID()); + ProcessingChain processingChain = mProcessingChains.get(channel.getChannelID()); + + processingChain.stop(); + + processingChain.removeEventLoggingModules(); + + processingChain.removeRecordingModules(); - chain.stop(); mChannelModel.broadcast(new ChannelEvent(channel, Event.NOTIFICATION_PROCESSING_STOP)); @@ -285,7 +285,7 @@ public class ChannelProcessingManager implements ChannelEventListener { mProcessingChains.remove(channel.getChannelID()); - chain.dispose(); + processingChain.dispose(); } } } diff --git a/src/module/ProcessingChain.java b/src/module/ProcessingChain.java index 298ba1e3..6d6f91e7 100644 --- a/src/module/ProcessingChain.java +++ b/src/module/ProcessingChain.java @@ -687,17 +687,13 @@ public class ProcessingChain implements IChannelEventListener { module.stop(); } - - removeEventLoggingModules(); - - removeRecordingModules(); } } /** * Removes any logging modules that are currently registered with this processing chain */ - private void removeEventLoggingModules() + public void removeEventLoggingModules() { List eventLoggingModules = new ArrayList<>(); @@ -718,7 +714,7 @@ public class ProcessingChain implements IChannelEventListener /** * Removes any recording modules that are currently registered with this processing chain */ - private void removeRecordingModules() + public void removeRecordingModules() { List recordingModules = new ArrayList<>(); diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index e71689b7..a35f5c17 100644 --- a/src/record/RecorderManager.java +++ b/src/record/RecorderManager.java @@ -45,7 +45,8 @@ public class RecorderManager implements Listener private boolean mCanStartNewRecorders = true; /** - * Manages starting and stopping all system recorders and distributes channel audio to each of the recorders. + * Manages all audio recording for all processing channels. Reconstructs audio streams and distributes channel + * audio to each of the recorders, starting and stopping the recorders as needed. */ public RecorderManager(ThreadPoolManager threadPoolManager) {