diff --git a/src/audio/AudioModule.java b/src/audio/AudioModule.java index 1f87f38d..01ad02d4 100644 --- a/src/audio/AudioModule.java +++ b/src/audio/AudioModule.java @@ -1,6 +1,7 @@ package audio; import java.util.Arrays; +import java.util.concurrent.ScheduledExecutorService; import module.Module; @@ -72,7 +73,7 @@ public class AudioModule extends Module implements IAudioPacketProvider, } @Override - public void start() + public void start( ScheduledExecutorService executor ) { /* No start operations provided */ } diff --git a/src/controller/ThreadPoolManager.java b/src/controller/ThreadPoolManager.java index c076422a..ca993295 100644 --- a/src/controller/ThreadPoolManager.java +++ b/src/controller/ThreadPoolManager.java @@ -41,6 +41,11 @@ public class ThreadPoolManager public ThreadPoolManager() { } + + public ScheduledExecutorService getScheduledExecutorService() + { + return mExecutor; + } public ScheduledFuture scheduleFixedRate( ThreadType type, Runnable command, diff --git a/src/controller/channel/ChannelProcessingManager.java b/src/controller/channel/ChannelProcessingManager.java index d1a5f423..dac06b4d 100644 --- a/src/controller/channel/ChannelProcessingManager.java +++ b/src/controller/channel/ChannelProcessingManager.java @@ -185,7 +185,7 @@ public class ChannelProcessingManager implements ChannelEventListener if( processingChain == null ) { - processingChain = new ProcessingChain( mThreadPoolManager, + processingChain = new ProcessingChain( channel.getName(), channel.getChannelType() ); /* Register global listeners */ diff --git a/src/instrument/InstrumentableProcessingChain.java b/src/instrument/InstrumentableProcessingChain.java index 94181b18..882eb891 100644 --- a/src/instrument/InstrumentableProcessingChain.java +++ b/src/instrument/InstrumentableProcessingChain.java @@ -15,7 +15,7 @@ public class InstrumentableProcessingChain extends ProcessingChain implements In { public InstrumentableProcessingChain() { - super( ChannelType.STANDARD ); + super( "instrumented", ChannelType.STANDARD ); } public void setSource( Source source ) throws IllegalStateException diff --git a/src/module/Module.java b/src/module/Module.java index 87295327..a46b3e9c 100644 --- a/src/module/Module.java +++ b/src/module/Module.java @@ -18,6 +18,8 @@ package module; +import java.util.concurrent.ScheduledExecutorService; + public abstract class Module { /** @@ -57,9 +59,9 @@ public abstract class Module public abstract void reset(); /** - * Start processing. + * Start processing. Provides an executor to use for scheduling tasks */ - public abstract void start(); + public abstract void start( ScheduledExecutorService executor ); /** * Stop processing diff --git a/src/module/ProcessingChain.java b/src/module/ProcessingChain.java index 046d8c51..e263a920 100644 --- a/src/module/ProcessingChain.java +++ b/src/module/ProcessingChain.java @@ -19,7 +19,8 @@ package module; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import message.IMessageListener; @@ -67,6 +68,7 @@ import audio.metadata.Metadata; import audio.squelch.ISquelchStateListener; import audio.squelch.ISquelchStateProvider; import audio.squelch.SquelchState; +import controller.NamingThreadFactory; import controller.ThreadPoolManager; import controller.channel.Channel.ChannelType; import controller.channel.ChannelEvent; @@ -108,8 +110,8 @@ public class ProcessingChain implements IChannelEventListener private Broadcaster mUnFilteredRealBufferBroadcaster = new Broadcaster<>(); private Broadcaster mSquelchStateBroadcaster = new Broadcaster<>(); - private ThreadPoolManager mThreadPoolManager; - private ScheduledFuture mBufferProcessingTask; + private ScheduledExecutorService mScheduledExecutorService; + private String mName; private AtomicBoolean mRunning = new AtomicBoolean(); protected Source mSource; @@ -117,19 +119,18 @@ public class ProcessingChain implements IChannelEventListener private CallEventModel mCallEventModel; private ChannelState mChannelState; private MessageActivityModel mMessageActivityModel; - - public ProcessingChain( ChannelType channelType ) + + /** + * 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 ) { - this( new ThreadPoolManager(), channelType ); - } - - - public ProcessingChain( ThreadPoolManager threadManager, - ChannelType channelType ) - { - mThreadPoolManager = threadManager; - - mChannelState = new ChannelState( mThreadPoolManager, channelType ); + mName = name; + + mChannelState = new ChannelState( channelType ); addModule( mChannelState ); mCallEventModel = new CallEventModel(); @@ -169,10 +170,6 @@ public class ProcessingChain implements IChannelEventListener mModules.clear(); - - mThreadPoolManager = null; - mBufferProcessingTask = null; - mAudioPacketBroadcaster.dispose(); mCallEventBroadcaster.dispose(); mChannelEventBroadcaster.dispose(); @@ -445,19 +442,6 @@ public class ProcessingChain implements IChannelEventListener module.reset(); } - /* Start each of the modules */ - for( Module module: mModules ) - { - try - { - module.start(); - } - catch( Exception e ) - { - mLog.error( "Error starting module", e ); - } - } - /* Register with the source to receive sample data. Setup a * timer task to process the buffer queues 50 times a second * (every 20 ms) */ @@ -492,8 +476,22 @@ public class ProcessingChain implements IChannelEventListener { mLog.error( "Error getting frequency from tuner channel source", e ); } - - tcs.start(); + } + + 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( Listener listener: mUnFilteredRealBufferBroadcaster.getListeners() ) @@ -536,12 +534,6 @@ public class ProcessingChain implements IChannelEventListener mSource = null; } - if( mBufferProcessingTask != null ) - { - mThreadPoolManager.cancel( mBufferProcessingTask ); - mBufferProcessingTask = null; - } - /* Stop each of the modules */ for( Module module: mModules ) { diff --git a/src/module/decode/am/AMDecoder.java b/src/module/decode/am/AMDecoder.java index 8b407b67..c5b371e3 100644 --- a/src/module/decode/am/AMDecoder.java +++ b/src/module/decode/am/AMDecoder.java @@ -18,6 +18,8 @@ package module.decode.am; +import java.util.concurrent.ScheduledExecutorService; + import module.decode.DecoderType; import module.decode.PrimaryDecoder; import module.decode.config.DecodeConfiguration; @@ -41,7 +43,7 @@ public class AMDecoder extends PrimaryDecoder } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/fleetsync2/Fleetsync2Decoder.java b/src/module/decode/fleetsync2/Fleetsync2Decoder.java index 43243de6..5b18628b 100644 --- a/src/module/decode/fleetsync2/Fleetsync2Decoder.java +++ b/src/module/decode/fleetsync2/Fleetsync2Decoder.java @@ -25,6 +25,7 @@ import instrument.tap.stream.FloatBufferTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import module.decode.Decoder; import module.decode.DecoderType; @@ -180,7 +181,7 @@ public class Fleetsync2Decoder extends Decoder implements IFilteredRealBufferLis } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/fleetsync2/Fleetsync2DecoderState.java b/src/module/decode/fleetsync2/Fleetsync2DecoderState.java index 8fb2cd00..bb5d614d 100644 --- a/src/module/decode/fleetsync2/Fleetsync2DecoderState.java +++ b/src/module/decode/fleetsync2/Fleetsync2DecoderState.java @@ -19,6 +19,7 @@ package module.decode.fleetsync2; import java.util.Iterator; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -58,7 +59,7 @@ public class Fleetsync2DecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/lj1200/LJ1200Decoder.java b/src/module/decode/lj1200/LJ1200Decoder.java index 5b33ed07..3e12b544 100644 --- a/src/module/decode/lj1200/LJ1200Decoder.java +++ b/src/module/decode/lj1200/LJ1200Decoder.java @@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import module.decode.Decoder; import module.decode.DecoderType; @@ -198,7 +199,7 @@ public class LJ1200Decoder extends Decoder implements IFilteredRealBufferListene } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/lj1200/LJ1200DecoderState.java b/src/module/decode/lj1200/LJ1200DecoderState.java index 71b72f0f..1069068f 100644 --- a/src/module/decode/lj1200/LJ1200DecoderState.java +++ b/src/module/decode/lj1200/LJ1200DecoderState.java @@ -22,6 +22,7 @@ import java.util.Collections; import java.util.List; import java.util.Set; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,7 +59,7 @@ public class LJ1200DecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/ltrnet/LTRNetDecoder.java b/src/module/decode/ltrnet/LTRNetDecoder.java index 544d8cf6..632a967d 100644 --- a/src/module/decode/ltrnet/LTRNetDecoder.java +++ b/src/module/decode/ltrnet/LTRNetDecoder.java @@ -23,6 +23,7 @@ import instrument.tap.TapGroup; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import message.MessageDirection; import module.decode.Decoder; @@ -120,7 +121,7 @@ public class LTRNetDecoder extends Decoder } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/ltrnet/LTRNetDecoderState.java b/src/module/decode/ltrnet/LTRNetDecoderState.java index 245f7ace..6f17d77a 100644 --- a/src/module/decode/ltrnet/LTRNetDecoderState.java +++ b/src/module/decode/ltrnet/LTRNetDecoderState.java @@ -21,6 +21,7 @@ import java.text.DecimalFormat; import java.util.HashMap; import java.util.Iterator; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -777,7 +778,7 @@ public class LTRNetDecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/ltrstandard/LTRStandardDecoder.java b/src/module/decode/ltrstandard/LTRStandardDecoder.java index 0324e029..b04f0e47 100644 --- a/src/module/decode/ltrstandard/LTRStandardDecoder.java +++ b/src/module/decode/ltrstandard/LTRStandardDecoder.java @@ -17,6 +17,8 @@ ******************************************************************************/ package module.decode.ltrstandard; +import java.util.concurrent.ScheduledExecutorService; + import message.MessageDirection; import module.decode.Decoder; import module.decode.DecoderType; @@ -83,7 +85,7 @@ public class LTRStandardDecoder extends Decoder implements IUnFilteredRealBuffer } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/ltrstandard/LTRStandardDecoderState.java b/src/module/decode/ltrstandard/LTRStandardDecoderState.java index 94cd32b9..9443660a 100644 --- a/src/module/decode/ltrstandard/LTRStandardDecoderState.java +++ b/src/module/decode/ltrstandard/LTRStandardDecoderState.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -61,7 +62,7 @@ public class LTRStandardDecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/mdc1200/MDCDecoder.java b/src/module/decode/mdc1200/MDCDecoder.java index d4351d56..89f04afc 100644 --- a/src/module/decode/mdc1200/MDCDecoder.java +++ b/src/module/decode/mdc1200/MDCDecoder.java @@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import module.decode.Decoder; import module.decode.DecoderType; @@ -210,7 +211,7 @@ public class MDCDecoder extends Decoder implements IFilteredRealBufferListener, } @Override - public void start() + public void start( ScheduledExecutorService executor ) { // TODO Auto-generated method stub diff --git a/src/module/decode/mdc1200/MDCDecoderState.java b/src/module/decode/mdc1200/MDCDecoderState.java index f896a429..0a9d3a7c 100644 --- a/src/module/decode/mdc1200/MDCDecoderState.java +++ b/src/module/decode/mdc1200/MDCDecoderState.java @@ -19,6 +19,7 @@ package module.decode.mdc1200; import java.util.Iterator; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -54,7 +55,7 @@ public class MDCDecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/mpt1327/MPT1327Decoder.java b/src/module/decode/mpt1327/MPT1327Decoder.java index 9363b19c..e7ea5b8a 100644 --- a/src/module/decode/mpt1327/MPT1327Decoder.java +++ b/src/module/decode/mpt1327/MPT1327Decoder.java @@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import module.decode.Decoder; import module.decode.DecoderType; @@ -306,7 +307,7 @@ public class MPT1327Decoder extends Decoder } @Override - public void start() + public void start( ScheduledExecutorService executor ) { // TODO Auto-generated method stub diff --git a/src/module/decode/mpt1327/MPT1327DecoderState.java b/src/module/decode/mpt1327/MPT1327DecoderState.java index a5b94f27..20858552 100644 --- a/src/module/decode/mpt1327/MPT1327DecoderState.java +++ b/src/module/decode/mpt1327/MPT1327DecoderState.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -85,7 +86,7 @@ public class MPT1327DecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/nbfm/NBFMDecoder.java b/src/module/decode/nbfm/NBFMDecoder.java index 1e715288..1b055a91 100644 --- a/src/module/decode/nbfm/NBFMDecoder.java +++ b/src/module/decode/nbfm/NBFMDecoder.java @@ -17,6 +17,8 @@ ******************************************************************************/ package module.decode.nbfm; +import java.util.concurrent.ScheduledExecutorService; + import module.decode.DecoderType; import module.decode.PrimaryDecoder; import module.decode.config.DecodeConfiguration; @@ -40,7 +42,7 @@ public class NBFMDecoder extends PrimaryDecoder } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/p25/P25DecoderState.java b/src/module/decode/p25/P25DecoderState.java index 55340dba..5fce8640 100644 --- a/src/module/decode/p25/P25DecoderState.java +++ b/src/module/decode/p25/P25DecoderState.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -204,7 +205,7 @@ public class P25DecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/p25/P25_C4FMDecoder.java b/src/module/decode/p25/P25_C4FMDecoder.java index 3bae443d..09d6e45d 100644 --- a/src/module/decode/p25/P25_C4FMDecoder.java +++ b/src/module/decode/p25/P25_C4FMDecoder.java @@ -25,6 +25,7 @@ import instrument.tap.stream.FloatTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -229,7 +230,7 @@ public class P25_C4FMDecoder extends P25Decoder } @Override - public void start() + public void start( ScheduledExecutorService executor ) { // TODO Auto-generated method stub diff --git a/src/module/decode/p25/P25_LSMDecoder.java b/src/module/decode/p25/P25_LSMDecoder.java index 7614ac24..b6e64bf3 100644 --- a/src/module/decode/p25/P25_LSMDecoder.java +++ b/src/module/decode/p25/P25_LSMDecoder.java @@ -26,6 +26,7 @@ import instrument.tap.stream.QPSKTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -222,7 +223,7 @@ public class P25_LSMDecoder extends P25Decoder implements IComplexBufferListener } @Override - public void start() + public void start( ScheduledExecutorService executor ) { // TODO Auto-generated method stub diff --git a/src/module/decode/p25/audio/P25AudioModule.java b/src/module/decode/p25/audio/P25AudioModule.java index 6e5bfdf9..68496892 100644 --- a/src/module/decode/p25/audio/P25AudioModule.java +++ b/src/module/decode/p25/audio/P25AudioModule.java @@ -1,5 +1,7 @@ package module.decode.p25.audio; +import java.util.concurrent.ScheduledExecutorService; + import jmbe.iface.AudioConversionLibrary; import jmbe.iface.AudioConverter; import message.IMessageListener; @@ -78,7 +80,7 @@ public class P25AudioModule extends Module implements Listener, } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/passport/PassportDecoder.java b/src/module/decode/passport/PassportDecoder.java index ddae98ed..ceb2291d 100644 --- a/src/module/decode/passport/PassportDecoder.java +++ b/src/module/decode/passport/PassportDecoder.java @@ -23,6 +23,7 @@ import instrument.tap.TapGroup; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import module.decode.Decoder; import module.decode.DecoderType; @@ -109,7 +110,7 @@ public class PassportDecoder extends Decoder } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/passport/PassportDecoderState.java b/src/module/decode/passport/PassportDecoderState.java index 003d2318..99004efd 100644 --- a/src/module/decode/passport/PassportDecoderState.java +++ b/src/module/decode/passport/PassportDecoderState.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import message.Message; import module.decode.DecoderType; @@ -75,7 +76,7 @@ public class PassportDecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/state/AlwaysUnsquelchedDecoderState.java b/src/module/decode/state/AlwaysUnsquelchedDecoderState.java index ad74e2f8..5fd65adb 100644 --- a/src/module/decode/state/AlwaysUnsquelchedDecoderState.java +++ b/src/module/decode/state/AlwaysUnsquelchedDecoderState.java @@ -17,6 +17,8 @@ ******************************************************************************/ package module.decode.state; +import java.util.concurrent.ScheduledExecutorService; + import message.Message; import module.decode.DecoderType; import module.decode.state.DecoderStateEvent.Event; @@ -90,7 +92,7 @@ public class AlwaysUnsquelchedDecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { broadcast( new Metadata( MetadataType.TO, mChannelName, false ) ); broadcast( new DecoderStateEvent( this, Event.ALWAYS_UNSQUELCH, State.IDLE ) ); diff --git a/src/module/decode/state/ChannelState.java b/src/module/decode/state/ChannelState.java index 450ed92d..836531a7 100644 --- a/src/module/decode/state/ChannelState.java +++ b/src/module/decode/state/ChannelState.java @@ -18,6 +18,7 @@ package module.decode.state; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -74,7 +75,6 @@ public class ChannelState extends Module implements ICallEventProvider, private AtomicBoolean mMonitoring = new AtomicBoolean(); private ScheduledFuture mStateMonitorFuture; - private ThreadPoolManager mThreadPoolManager; private ChannelType mChannelType; private TrafficChannelStatusListener mTrafficChannelStatusListener; @@ -106,9 +106,8 @@ public class ChannelState extends Module implements ICallEventProvider, * ended, while continuing to display the call details for the user * TEARDOWN Indicates a traffic channel that will be torn down for reuse. */ - public ChannelState( ThreadPoolManager manager, ChannelType channelType ) + public ChannelState( ChannelType channelType ) { - mThreadPoolManager = manager; mChannelType = channelType; } @@ -126,7 +125,7 @@ public class ChannelState extends Module implements ICallEventProvider, } @Override - public void start() + public void start( ScheduledExecutorService executor ) { //Start a monitoring task to check for FADE and END event timeouts if( mMonitoring.compareAndSet( false, true ) ) @@ -135,8 +134,8 @@ public class ChannelState extends Module implements ICallEventProvider, { try { - mStateMonitorFuture = mThreadPoolManager.scheduleFixedRate( - ThreadType.DECODER, mStateMonitor, 20, TimeUnit.MILLISECONDS ); + mStateMonitorFuture = executor.scheduleAtFixedRate( + mStateMonitor, 0, 20, TimeUnit.MILLISECONDS ); } catch( RejectedExecutionException ree ) { @@ -157,7 +156,7 @@ public class ChannelState extends Module implements ICallEventProvider, { if( mStateMonitorFuture != null ) { - boolean success = mThreadPoolManager.cancel( mStateMonitorFuture ); + boolean success = mStateMonitorFuture.cancel( true ); if( !success ) { @@ -175,7 +174,6 @@ public class ChannelState extends Module implements ICallEventProvider, mChangedAttributeListener = null; mDecoderStateListener = null; mSquelchStateListener = null; - mThreadPoolManager = null; mStateMonitor = null; } diff --git a/src/module/decode/state/TrafficChannelManager.java b/src/module/decode/state/TrafficChannelManager.java index ec64ab84..5455d7d5 100644 --- a/src/module/decode/state/TrafficChannelManager.java +++ b/src/module/decode/state/TrafficChannelManager.java @@ -4,6 +4,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; import module.Module; import module.ProcessingChain; @@ -86,13 +87,21 @@ public class TrafficChannelManager extends Module mSystem = system; mSite = site; mAliasListName = aliasListName; -// mTrafficChannelTimeout = trafficChannelTimeout; mTrafficChannelPoolMaximumSize = trafficChannelPoolSize; } @Override public void dispose() { + for( Channel trafficChannel: mTrafficChannelPool ) + { + mChannelModel.broadcast( new ChannelEvent( trafficChannel, Event.REQUEST_DISABLE ) ); + } + + mTrafficChannelPool.clear(); + + mTrafficChannelsInUse.clear(); + mCallEventListener = null; mDecodeConfiguration = null; mPreviousDoNotMonitorCallEvent = null; @@ -350,7 +359,7 @@ public class TrafficChannelManager extends Module } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/decode/tait/Tait1200Decoder.java b/src/module/decode/tait/Tait1200Decoder.java index 9f19f99a..97b9b788 100644 --- a/src/module/decode/tait/Tait1200Decoder.java +++ b/src/module/decode/tait/Tait1200Decoder.java @@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import module.decode.Decoder; import module.decode.DecoderType; @@ -198,7 +199,7 @@ public class Tait1200Decoder extends Decoder implements IFilteredRealBufferListe } @Override - public void start() + public void start( ScheduledExecutorService executor ) { // TODO Auto-generated method stub diff --git a/src/module/decode/tait/Tait1200DecoderState.java b/src/module/decode/tait/Tait1200DecoderState.java index 589e9d5f..569c6a36 100644 --- a/src/module/decode/tait/Tait1200DecoderState.java +++ b/src/module/decode/tait/Tait1200DecoderState.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.TreeSet; +import java.util.concurrent.ScheduledExecutorService; import org.jdesktop.swingx.mapviewer.GeoPosition; @@ -60,7 +61,7 @@ public class Tait1200DecoderState extends DecoderState } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/demodulate/am/AMDemodulatorModule.java b/src/module/demodulate/am/AMDemodulatorModule.java index 8e311102..374491aa 100644 --- a/src/module/demodulate/am/AMDemodulatorModule.java +++ b/src/module/demodulate/am/AMDemodulatorModule.java @@ -17,6 +17,8 @@ ******************************************************************************/ package module.demodulate.am; +import java.util.concurrent.ScheduledExecutorService; + import module.Module; import sample.Listener; import sample.complex.ComplexBuffer; @@ -111,7 +113,7 @@ public class AMDemodulatorModule extends Module } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/demodulate/audio/DemodulatedAudioFilterModule.java b/src/module/demodulate/audio/DemodulatedAudioFilterModule.java index c6535268..7920d9a1 100644 --- a/src/module/demodulate/audio/DemodulatedAudioFilterModule.java +++ b/src/module/demodulate/audio/DemodulatedAudioFilterModule.java @@ -17,6 +17,8 @@ ******************************************************************************/ package module.demodulate.audio; +import java.util.concurrent.ScheduledExecutorService; + import module.Module; import org.slf4j.Logger; @@ -84,7 +86,7 @@ public class DemodulatedAudioFilterModule extends Module } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/demodulate/fm/FMDemodulatorModule.java b/src/module/demodulate/fm/FMDemodulatorModule.java index da39a197..f96f9528 100644 --- a/src/module/demodulate/fm/FMDemodulatorModule.java +++ b/src/module/demodulate/fm/FMDemodulatorModule.java @@ -17,6 +17,8 @@ ******************************************************************************/ package module.demodulate.fm; +import java.util.concurrent.ScheduledExecutorService; + import module.Module; import org.slf4j.Logger; @@ -97,7 +99,7 @@ public class FMDemodulatorModule extends Module } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/module/log/CallEventLogger.java b/src/module/log/CallEventLogger.java index 3d01e45b..a3e5ab14 100644 --- a/src/module/log/CallEventLogger.java +++ b/src/module/log/CallEventLogger.java @@ -58,6 +58,5 @@ public class CallEventLogger extends EventLogger implements ICallEventListener, @Override public void reset() { - super.start(); } } diff --git a/src/module/log/EventLogger.java b/src/module/log/EventLogger.java index b1d5e46f..120ee29a 100644 --- a/src/module/log/EventLogger.java +++ b/src/module/log/EventLogger.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.io.OutputStreamWriter; import java.io.Writer; import java.nio.file.Path; +import java.util.concurrent.ScheduledExecutorService; import module.Module; @@ -67,7 +68,8 @@ public abstract class EventLogger extends Module public abstract String getHeader(); - public void start() + @Override + public void start( ScheduledExecutorService executor ) { if( mLogFile == null ) { diff --git a/src/module/log/MessageEventLogger.java b/src/module/log/MessageEventLogger.java index 65c1c918..de021df7 100644 --- a/src/module/log/MessageEventLogger.java +++ b/src/module/log/MessageEventLogger.java @@ -58,7 +58,6 @@ public class MessageEventLogger extends EventLogger implements @Override public void reset() { - super.start(); } @Override diff --git a/src/record/RecorderManager.java b/src/record/RecorderManager.java index b4bf4aad..4cc30ab2 100644 --- a/src/record/RecorderManager.java +++ b/src/record/RecorderManager.java @@ -86,10 +86,9 @@ public class RecorderManager implements Listener try { RealBufferWaveRecorder recorder = - new RealBufferWaveRecorder( mThreadPoolManager, - AUDIO_SAMPLE_RATE, filePrefix ); + new RealBufferWaveRecorder( AUDIO_SAMPLE_RATE, filePrefix ); - recorder.start(); + recorder.start( mThreadPoolManager.getScheduledExecutorService() ); recorder.receive( audioPacket.getAudioBuffer() ); mRecorders.put( identifier, recorder ); @@ -177,7 +176,6 @@ public class RecorderManager implements Listener sb.append( channelName ); sb.append( "_baseband" ); - return new ComplexBufferWaveRecorder( mThreadPoolManager, - AUDIO_SAMPLE_RATE, sb.toString() ); + return new ComplexBufferWaveRecorder( AUDIO_SAMPLE_RATE, sb.toString() ); } } diff --git a/src/record/wave/ComplexBufferWaveRecorder.java b/src/record/wave/ComplexBufferWaveRecorder.java index 53eed8cb..f738bc7f 100644 --- a/src/record/wave/ComplexBufferWaveRecorder.java +++ b/src/record/wave/ComplexBufferWaveRecorder.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -38,8 +39,6 @@ import sample.Listener; import sample.complex.ComplexBuffer; import sample.complex.IComplexBufferListener; import util.TimeStamp; -import controller.ThreadPoolManager; -import controller.ThreadPoolManager.ThreadType; /** * WAVE audio recorder module for recording complex (I&Q) samples to a wave file @@ -55,18 +54,14 @@ public class ComplexBufferWaveRecorder extends Module private Path mFile; private AudioFormat mAudioFormat; - private ThreadPoolManager mThreadPoolManager; private BufferProcessor mBufferProcessor; private ScheduledFuture mProcessorHandle; private LinkedBlockingQueue mBuffers = new LinkedBlockingQueue<>( 500 ); private AtomicBoolean mRunning = new AtomicBoolean(); - public ComplexBufferWaveRecorder( ThreadPoolManager threadPoolManager, - int sampleRate, String filePrefix ) + public ComplexBufferWaveRecorder( int sampleRate, String filePrefix ) { - mThreadPoolManager = threadPoolManager; - mAudioFormat = new AudioFormat( sampleRate, //SampleRate 16, //Sample Size 2, //Channels @@ -81,7 +76,7 @@ public class ComplexBufferWaveRecorder extends Module return mFile; } - public void start() + public void start( ScheduledExecutorService executor ) { if( mRunning.compareAndSet( false, true ) ) { @@ -103,9 +98,8 @@ public class ComplexBufferWaveRecorder extends Module mWriter = new WaveWriter( mAudioFormat, mFile ); /* Schedule the processor to run every 500 milliseconds */ - mProcessorHandle = mThreadPoolManager.scheduleFixedRate( - ThreadType.BASEBAND_RECORDING, mBufferProcessor, 500, - TimeUnit.MILLISECONDS ); + mProcessorHandle = executor.scheduleAtFixedRate( + mBufferProcessor, 0, 500, TimeUnit.MILLISECONDS ); } catch( IOException io ) { @@ -149,8 +143,6 @@ public class ComplexBufferWaveRecorder extends Module public void dispose() { stop(); - - mThreadPoolManager = null; } @Override @@ -190,7 +182,10 @@ public class ComplexBufferWaveRecorder extends Module mFile = null; - mThreadPoolManager.cancel( mProcessorHandle ); + if( mProcessorHandle != null ) + { + mProcessorHandle.cancel( true ); + } mProcessorHandle = null; } diff --git a/src/record/wave/RealBufferWaveRecorder.java b/src/record/wave/RealBufferWaveRecorder.java index 0ad0d4f7..ac17988a 100644 --- a/src/record/wave/RealBufferWaveRecorder.java +++ b/src/record/wave/RealBufferWaveRecorder.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -37,8 +38,6 @@ import sample.Listener; import sample.real.IFilteredRealBufferListener; import sample.real.RealBuffer; import util.TimeStamp; -import controller.ThreadPoolManager; -import controller.ThreadPoolManager.ThreadType; /** * WAVE audio recorder module for recording real sample buffers to a wave file @@ -54,7 +53,6 @@ public class RealBufferWaveRecorder extends Module private Path mFile; private AudioFormat mAudioFormat; - private ThreadPoolManager mThreadPoolManager; private BufferProcessor mBufferProcessor; private ScheduledFuture mProcessorHandle; private LinkedBlockingQueue mBuffers = @@ -63,11 +61,8 @@ public class RealBufferWaveRecorder extends Module private AtomicBoolean mRunning = new AtomicBoolean(); - public RealBufferWaveRecorder( ThreadPoolManager threadPoolManager, - int sampleRate, String filePrefix ) + public RealBufferWaveRecorder( int sampleRate, String filePrefix ) { - mThreadPoolManager = threadPoolManager; - mAudioFormat = new AudioFormat( sampleRate, //SampleRate 16, //Sample Size 1, //Channels @@ -90,7 +85,7 @@ public class RealBufferWaveRecorder extends Module return mFile; } - public void start() + public void start( ScheduledExecutorService executor ) { if( mRunning.compareAndSet( false, true ) ) { @@ -112,9 +107,8 @@ public class RealBufferWaveRecorder extends Module mWriter = new WaveWriter( mAudioFormat, mFile ); /* Schedule the processor to run every 500 milliseconds */ - mProcessorHandle = mThreadPoolManager.scheduleFixedRate( - ThreadType.AUDIO_RECORDING, mBufferProcessor, 500, - TimeUnit.MILLISECONDS ); + mProcessorHandle = executor.scheduleAtFixedRate( + mBufferProcessor, 0, 500, TimeUnit.MILLISECONDS ); } catch( IOException io ) { @@ -160,8 +154,6 @@ public class RealBufferWaveRecorder extends Module public void dispose() { stop(); - - mThreadPoolManager = null; } @Override @@ -198,12 +190,14 @@ public class RealBufferWaveRecorder extends Module } } - mThreadPoolManager.cancel( mProcessorHandle ); + if( mProcessorHandle != null ) + { + mProcessorHandle.cancel( true ); + } + mProcessorHandle = null; mWriter = null; mFile = null; - mProcessorHandle = null; - mThreadPoolManager = null; } else { diff --git a/src/source/mixer/RealMixerSource.java b/src/source/mixer/RealMixerSource.java index 240e55bb..7ea8dd7b 100644 --- a/src/source/mixer/RealMixerSource.java +++ b/src/source/mixer/RealMixerSource.java @@ -19,6 +19,7 @@ package source.mixer; import java.util.Iterator; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import javax.sound.sampled.AudioFormat; @@ -29,12 +30,12 @@ import javax.sound.sampled.TargetDataLine; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import dsp.gain.AutomaticGainControl; import sample.Listener; import sample.adapter.ISampleAdapter; import sample.real.RealBuffer; import source.RealSource; import source.SourceException; +import dsp.gain.AutomaticGainControl; public class RealMixerSource extends RealSource { @@ -86,7 +87,7 @@ public class RealMixerSource extends RealSource } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/source/tuner/TunerChannelSource.java b/src/source/tuner/TunerChannelSource.java index f75e95e7..fada5575 100644 --- a/src/source/tuner/TunerChannelSource.java +++ b/src/source/tuner/TunerChannelSource.java @@ -19,7 +19,6 @@ package source.tuner; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.Executors; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; @@ -39,8 +38,6 @@ import source.SourceException; import source.tuner.frequency.FrequencyChangeEvent; import source.tuner.frequency.FrequencyChangeEvent.Event; import source.tuner.frequency.IFrequencyChangeProcessor; -import controller.NamingThreadFactory; -import controller.ThreadPoolManager.ThreadType; import dsp.filter.FilterFactory; import dsp.filter.Window.WindowType; import dsp.filter.cic.ComplexPrimeCICDecimate; @@ -63,7 +60,6 @@ public class TunerChannelSource extends ComplexSource private ComplexPrimeCICDecimate mDecimationFilter; private Listener mListener; private IFrequencyChangeProcessor mFrequencyChangeProcessor; - private ScheduledExecutorService mScheduledExecutorService; private ScheduledFuture mTaskHandle; private long mTunerFrequency = 0; @@ -114,14 +110,7 @@ public class TunerChannelSource extends ComplexSource } - @Override - public void reset() - { - } - - - @Override - public void start() + public void start( ScheduledExecutorService executor ) { if( mExpended ) { @@ -132,12 +121,9 @@ public class TunerChannelSource extends ComplexSource if( mRunning.compareAndSet( false, true )) { - mScheduledExecutorService = Executors.newScheduledThreadPool( 1, - new NamingThreadFactory( "tuner channel source" ) ); - /* Schedule the decimation task to run every 10 ms (100 iterations/second) */ - mTaskHandle = mScheduledExecutorService.scheduleAtFixedRate( - mDecimationProcessor, 0, 10, TimeUnit.MILLISECONDS ); + mTaskHandle = executor.scheduleAtFixedRate( mDecimationProcessor, + 0, 10, TimeUnit.MILLISECONDS ); /* Finally, register to receive samples from the tuner */ mTuner.addListener( (Listener)this ); @@ -149,6 +135,10 @@ public class TunerChannelSource extends ComplexSource } } + @Override + public void reset() + { + } @Override public void stop() @@ -158,15 +148,10 @@ public class TunerChannelSource extends ComplexSource mTuner.releaseChannel( this ); mDecimationProcessor.shutdown(); - if( mScheduledExecutorService != null && mTaskHandle != null ) + if( mTaskHandle != null ) { - if( mTaskHandle != null ) - { - mTaskHandle.cancel( true ); - mTaskHandle = null; - } - - mScheduledExecutorService.shutdownNow(); + mTaskHandle.cancel( true ); + mTaskHandle = null; } mBuffer.clear(); @@ -349,7 +334,7 @@ public class TunerChannelSource extends ComplexSource { long start = System.currentTimeMillis(); - mBuffer.drainTo( mSampleBuffers ); + mBuffer.drainTo( mSampleBuffers, 8 ); for( Buffer buffer: mSampleBuffers ) { diff --git a/src/source/tuner/frequency/DCTrackingFrequencyControl.java b/src/source/tuner/frequency/DCTrackingFrequencyControl.java index 17704dfd..3c1cfe68 100644 --- a/src/source/tuner/frequency/DCTrackingFrequencyControl.java +++ b/src/source/tuner/frequency/DCTrackingFrequencyControl.java @@ -18,6 +18,8 @@ package source.tuner.frequency; * along with this program. If not, see ******************************************************************************/ +import java.util.concurrent.ScheduledExecutorService; + import module.Module; import org.slf4j.Logger; @@ -118,7 +120,7 @@ public class DCTrackingFrequencyControl extends Module } @Override - public void start() + public void start( ScheduledExecutorService executor ) { } diff --git a/src/source/wave/ComplexWaveSource.java b/src/source/wave/ComplexWaveSource.java index 9e4a65a8..efa8b14b 100644 --- a/src/source/wave/ComplexWaveSource.java +++ b/src/source/wave/ComplexWaveSource.java @@ -20,6 +20,7 @@ package source.wave; import java.io.File; import java.io.IOException; import java.util.Arrays; +import java.util.concurrent.ScheduledExecutorService; import javax.sound.sampled.AudioFormat; import javax.sound.sampled.AudioInputStream; @@ -58,11 +59,11 @@ public class ComplexWaveSource extends ComplexSource implements IControllableFil public void reset() { stop(); - start(); + start( null ); } @Override - public void start() + public void start( ScheduledExecutorService executor ) { try { diff --git a/src/source/wave/RealWaveSource.java b/src/source/wave/RealWaveSource.java index 9af020c1..06e4fc25 100644 --- a/src/source/wave/RealWaveSource.java +++ b/src/source/wave/RealWaveSource.java @@ -20,6 +20,7 @@ package source.wave; import java.io.File; import java.io.IOException; import java.util.Arrays; +import java.util.concurrent.ScheduledExecutorService; import javax.sound.sampled.AudioFormat; import javax.sound.sampled.AudioInputStream; @@ -60,12 +61,12 @@ public class RealWaveSource extends RealSource public void reset() { stop(); - start(); + start( null ); } @Override - public void start() + public void start( ScheduledExecutorService executor ) { try {