Rebase channel/alias model redesign branch on current master

(v0.2.0.beta5)

src/audio/output/AudioOutput.java

@@ -1,10 +1,9 @@
 package audio.output;
 
 import java.util.concurrent.LinkedTransferQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.sound.sampled.FloatControl;
+import javax.sound.sampled.LineListener;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -12,23 +11,26 @@ import org.slf4j.LoggerFactory;
 import sample.Broadcaster;
 import sample.Listener;
 import audio.AudioEvent;
-import audio.AudioEvent.Type;
 import audio.AudioPacket;
 import audio.metadata.AudioMetadata;
+import controller.ThreadPoolManager;
 
-public abstract class AudioOutput implements Listener<AudioPacket>
+public abstract class AudioOutput implements Listener<AudioPacket>, LineListener
 {
 	private final static Logger mLog = LoggerFactory.getLogger( AudioOutput.class );
 	
-	private AtomicBoolean mSquelched = new AtomicBoolean( false );
 	private Listener<AudioMetadata> mAudioMetadataListener;
-
 	protected Broadcaster<AudioEvent> mAudioEventBroadcaster = new Broadcaster<>();
 	protected LinkedTransferQueue<AudioPacket> mBuffer = new LinkedTransferQueue<>();
-	protected ScheduledExecutorService mExecutorService;
+	protected ThreadPoolManager mThreadPoolManager;
 	protected boolean mCanProcessAudio;
 	protected long mLastActivity;
 
+	public AudioOutput( ThreadPoolManager threadPoolManager )
+	{
+		mThreadPoolManager = threadPoolManager;
+	}
+	
 	public void reset()
 	{
 		broadcast( new AudioEvent( AudioEvent.Type.AUDIO_STOPPED, 
@@ -41,14 +43,11 @@ public abstract class AudioOutput implements Listener<AudioPacket>
 
 		mBuffer.clear();
 
-		if( mExecutorService != null )
-		{
-			mExecutorService.shutdown();
-		}
-
 		mAudioEventBroadcaster.dispose();
 		mAudioEventBroadcaster = null;
 		mAudioMetadataListener = null;
+		
+		mThreadPoolManager = null;
 	}
 	
 	public abstract String getChannelName();
@@ -72,19 +71,6 @@ public abstract class AudioOutput implements Listener<AudioPacket>
 		mAudioEventBroadcaster.broadcast( audioEvent );
 	}
 	
-	public void setSquelched( boolean squelched )
-	{
-		mSquelched.set( squelched );
-		
-		broadcast( new AudioEvent( squelched ? Type.AUDIO_SQUELCHED : 
-			Type.AUDIO_UNSQUELCHED, getChannelName() ) );
-	}
-	
-	public boolean isSquelched()
-	{
-		return mSquelched.get();
-	}
-	
 	/**
 	 * Registers a single listener to receive audio start and audio stop events
 	 */
@@ -130,4 +116,4 @@ public abstract class AudioOutput implements Listener<AudioPacket>
 			mBuffer.add( packet );
 		}
 	}
-}
+}

src/audio/output/MonoAudioOutput.java

@@ -5,13 +5,14 @@ import java.nio.ByteOrder;
 import java.nio.ShortBuffer;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.sound.sampled.BooleanControl;
 import javax.sound.sampled.Control;
 import javax.sound.sampled.FloatControl;
+import javax.sound.sampled.LineEvent;
 import javax.sound.sampled.LineUnavailableException;
 import javax.sound.sampled.Mixer;
 import javax.sound.sampled.SourceDataLine;
@@ -24,8 +25,8 @@ import audio.AudioEvent;
 import audio.AudioFormats;
 import audio.AudioPacket;
 import audio.AudioPacket.Type;
-import audio.output.StereoAudioOutput.QueueProcessor;
-import controller.NamingThreadFactory;
+import controller.ThreadPoolManager;
+import controller.ThreadPoolManager.ThreadType;
 
 /**
  * Mono Audio output with automatic flow control based on the availability of 
@@ -50,14 +51,14 @@ public class MonoAudioOutput extends AudioOutput
 	private SourceDataLine mOutput;
 	private FloatControl mGainControl;
 	private BooleanControl mMuteControl;
-	
 	private AudioEvent mAudioStartEvent;
 	private AudioEvent mAudioStopEvent;
 	private AudioEvent mAudioContinuationEvent;
+	private ScheduledFuture<?> mProcessorTask;
 	
-	public MonoAudioOutput( Mixer mixer )
+	public MonoAudioOutput( ThreadPoolManager threadPoolManager, Mixer mixer )
 	{
-		super();
+		super( threadPoolManager );
 
 		mAudioStartEvent = new AudioEvent( AudioEvent.Type.AUDIO_STARTED, 
 				getChannelName() );
@@ -74,15 +75,8 @@ public class MonoAudioOutput extends AudioOutput
 			if( mOutput != null )
 			{
 				mOutput.open( AudioFormats.PCM_SIGNED_48KHZ_16BITS_MONO, BUFFER_SIZE );
-				
+				mOutput.addLineListener( this );
 				mCanProcessAudio = true;
-				
-				mExecutorService = Executors.newSingleThreadScheduledExecutor( 
-						new NamingThreadFactory( "audio (mono) output" ) );
-
-				/* Run the queue processor task every 40 milliseconds */
-				mExecutorService.scheduleAtFixedRate( new QueueProcessor(), 
-						0, 40, TimeUnit.MILLISECONDS );
 			}
 		} 
         catch ( LineUnavailableException e )
@@ -115,12 +109,10 @@ public class MonoAudioOutput extends AudioOutput
 					mixer.getMixerInfo().getName() + "]" );
 			}
 			
-			mExecutorService = Executors.newSingleThreadScheduledExecutor( 
-					new NamingThreadFactory( "audio (mono) output" ) );
-
 			/* Run the queue processor task every 40 milliseconds */
-			mExecutorService.scheduleAtFixedRate( new QueueProcessor(), 
-					0, 40, TimeUnit.MILLISECONDS );
+			mProcessorTask = mThreadPoolManager.scheduleFixedRate( 
+				ThreadType.AUDIO_PROCESSING, new QueueProcessor(), 
+				40, TimeUnit.MILLISECONDS );
 		}
 	}
 
@@ -132,102 +124,108 @@ public class MonoAudioOutput extends AudioOutput
 
 	public void dispose()
 	{
+		if( mProcessorTask != null && mThreadPoolManager != null )
+		{
+			mThreadPoolManager.cancel( mProcessorTask );
+		}
+		
+		mProcessorTask = null;
+		
 		super.dispose();
 		
 		if( mOutput != null )
 		{
 			mOutput.close();
 		}
+		
+		mOutput = null;
+		mGainControl = null;
+		mMuteControl = null;
 	}
 	
 	public class QueueProcessor implements Runnable
 	{
 		private AtomicBoolean mProcessing = new AtomicBoolean();
 		
-		public QueueProcessor()
-		{
-		}
-
 		@Override
 		public void run()
 		{
-			/* The mProcessing flag ensures that only one instance of the
-			 * processor can run at any given time */
-			if( mProcessing.compareAndSet( false, true ) )
+			try
 			{
-				List<AudioPacket> packets = new ArrayList<AudioPacket>();
-				
-				mBuffer.drainTo( packets );
-				
-				if( packets.size() > 0 )
+				/* The mProcessing flag ensures that only one instance of the
+				 * processor can run at any given time */
+				if( mProcessing.compareAndSet( false, true ) )
 				{
-					broadcast( mAudioContinuationEvent );
+					List<AudioPacket> packets = new ArrayList<AudioPacket>();
 					
-					for( AudioPacket packet: packets )
+					mBuffer.drainTo( packets );
+					
+					if( packets.size() > 0 )
 					{
-						if( mCanProcessAudio )
+						for( AudioPacket packet: packets )
 						{
-							if( isSquelched() )
+							broadcast( mAudioContinuationEvent );
+							
+							if( mCanProcessAudio )
 							{
-								checkStop();
-							}
-							else if( packet.getType() == Type.AUDIO )
-							{
-								float[] samples = packet.getAudioBuffer().getSamples();
-								
-								/* Little-endian byte buffer */
-								ByteBuffer buffer = 
-									ByteBuffer.allocate( samples.length * 2 )
-											.order( ByteOrder.LITTLE_ENDIAN );
-								
-								ShortBuffer shortBuffer = buffer.asShortBuffer();
-
-								for( float sample: samples )
+								if( packet.getType() == Type.AUDIO )
 								{
-									shortBuffer.put( (short)( sample * Short.MAX_VALUE ) );
-								}
-								
-								mOutput.write( buffer.array(), 0, 
-										buffer.array().length );
-								
-								checkStart();
-								
-								broadcast( packet.getAudioMetadata() );
+									float[] samples = packet.getAudioBuffer().getSamples();
+									
+									/* Little-endian byte buffer */
+									ByteBuffer buffer = 
+										ByteBuffer.allocate( samples.length * 2 )
+												.order( ByteOrder.LITTLE_ENDIAN );
+									
+									ShortBuffer shortBuffer = buffer.asShortBuffer();
 
-								mLastActivity = System.currentTimeMillis();
+									for( float sample: samples )
+									{
+										shortBuffer.put( (short)( sample * Short.MAX_VALUE ) );
+									}
+									
+									//This is a blocking write
+									mOutput.write( buffer.array(), 0, buffer.array().length );
+
+									checkStart();
+
+									broadcast( packet.getAudioMetadata() );
+
+									mLastActivity = System.currentTimeMillis();
+								}
 							}
 						}
 					}
+					else
+					{
+						checkStop();
+					}
+					
+					mProcessing.set( false );
 				}
-				else
-				{
-					checkStop();
-				}
-				
-				mProcessing.set( false );
+			}
+			catch( Exception e )
+			{
+				mLog.error( "Error while processing audio buffers", e );
 			}
 		}
 		
 		private void checkStart()
 		{
-			if( mCanProcessAudio && !isSquelched() && 
+			if( mCanProcessAudio && 
 				!mOutput.isRunning() && mOutput.available() <= START_THRESHOLD )
 			{
 				mOutput.start();
-				
-				mAudioEventBroadcaster.broadcast( mAudioStartEvent );
 			}
 		}
 		
 		private void checkStop()
 		{
 			if( mCanProcessAudio && mOutput.isRunning() &&
-				( mOutput.available() >= STOP_THRESHOLD || isSquelched() ) )
+				mOutput.available() >= STOP_THRESHOLD )
 			{
 				mOutput.drain();
 				mOutput.stop();
-
-				mAudioEventBroadcaster.broadcast( mAudioStopEvent );
 			}
 		}
 	}
@@ -266,4 +264,19 @@ public class MonoAudioOutput extends AudioOutput
 	{
 		return mGainControl != null;
 	}
-}
+
+	@Override
+	public void update( LineEvent event )
+	{
+		LineEvent.Type type = event.getType();
+
+		if( type == LineEvent.Type.START )
+		{
+			mAudioEventBroadcaster.broadcast( mAudioStartEvent );
+		}
+		else if( type == LineEvent.Type.STOP )
+		{
+			mAudioEventBroadcaster.broadcast( mAudioStopEvent );
+		}
+	}
+}

src/audio/output/StereoAudioOutput.java

@@ -5,13 +5,14 @@ import java.nio.ByteOrder;
 import java.nio.ShortBuffer;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.sound.sampled.BooleanControl;
 import javax.sound.sampled.Control;
 import javax.sound.sampled.FloatControl;
+import javax.sound.sampled.LineEvent;
 import javax.sound.sampled.LineUnavailableException;
 import javax.sound.sampled.Mixer;
 import javax.sound.sampled.SourceDataLine;
@@ -24,10 +25,11 @@ import audio.AudioEvent;
 import audio.AudioFormats;
 import audio.AudioPacket;
 import audio.AudioPacket.Type;
-import controller.NamingThreadFactory;
+import controller.ThreadPoolManager;
+import controller.ThreadPoolManager.ThreadType;
 
 /**
- * Mono Audio output in stereo format with automatic flow control based on the 
+ * Stereo Audio output in stereo format with automatic flow control based on the 
  * availability of audio data packets.  
  */
 public class StereoAudioOutput extends AudioOutput
@@ -39,9 +41,9 @@ public class StereoAudioOutput extends AudioOutput
 	private final static int BUFFER_SIZE = 96000;
 	
 	/* The queue processor will run every 20 milliseconds checking for inbound
-	 * audio and automatically starting or stopping audio playback.  We set the 
+	 * audio and automatically start or stop audio playback.  We set the 
 	 * data line available space threshold to start when we're 20 ms away from 
-	 * being full, and the stop threshold when 20 ms away from being empty */
+	 * being full and the stop threshold when 20 ms away from being empty */
 	private final static int SAMPLE_SIZE_40_MS = BUFFER_SIZE / 25;
 	private final static int START_THRESHOLD = SAMPLE_SIZE_40_MS;
 	private final static int STOP_THRESHOLD = BUFFER_SIZE - SAMPLE_SIZE_40_MS;
@@ -49,16 +51,16 @@ public class StereoAudioOutput extends AudioOutput
 	private SourceDataLine mOutput;
 	private FloatControl mGainControl;
 	private BooleanControl mMuteControl;
-	
 	private AudioEvent mAudioStartEvent;
 	private AudioEvent mAudioStopEvent;
 	private AudioEvent mAudioContinuationEvent;
-	
 	private MixerChannel mMixerChannel;
+	private ScheduledFuture<?> mProcessorTask;
 	
-	public StereoAudioOutput( Mixer mixer, MixerChannel channel )
+	public StereoAudioOutput( ThreadPoolManager threadPoolManager, 
+			Mixer mixer, MixerChannel channel )
 	{
-		super();
+		super( threadPoolManager );
 		
 		assert( channel == MixerChannel.LEFT || channel == MixerChannel.RIGHT );
 		
@@ -79,6 +81,7 @@ public class StereoAudioOutput extends AudioOutput
 			if( mOutput != null )
 			{
 				mOutput.open( AudioFormats.PCM_SIGNED_48KHZ_16BITS_STEREO, BUFFER_SIZE );
+				mOutput.addLineListener( this );
 				mCanProcessAudio = true;
 			}
 		} 
@@ -113,12 +116,10 @@ public class StereoAudioOutput extends AudioOutput
 					mixer.getMixerInfo().getName() + " | " + channel.name() + "]" );
 			}
 			
-			mExecutorService = Executors.newSingleThreadScheduledExecutor( 
-					new NamingThreadFactory( "audio (stereo) output" ) );
-
 			/* Run the queue processor task every 40 milliseconds */
-			mExecutorService.scheduleAtFixedRate( new QueueProcessor(), 
-					0, 40, TimeUnit.MILLISECONDS );
+			mProcessorTask = mThreadPoolManager.scheduleFixedRate( 
+				ThreadType.AUDIO_PROCESSING, new QueueProcessor(), 
+				40, TimeUnit.MILLISECONDS );
 		}
 	}
 
@@ -130,20 +131,30 @@ public class StereoAudioOutput extends AudioOutput
 
 	public void dispose()
 	{
+		if( mProcessorTask != null && mThreadPoolManager != null )
+		{
+			mThreadPoolManager.cancel( mProcessorTask );
+		}
+		
+		mProcessorTask = null;
+
 		super.dispose();
 		
 		if( mOutput != null )
 		{
 			mOutput.close();
 		}
+		
+		mOutput = null;
+		mGainControl = null;
+		mMuteControl = null;
 	}
 	
 	private ByteBuffer convert( float[] samples )
 	{
 		/* Little-endian byte buffer */
-		ByteBuffer buffer = 
-			ByteBuffer.allocate( samples.length * 4 )
-					.order( ByteOrder.LITTLE_ENDIAN );
+		ByteBuffer buffer = ByteBuffer.allocate( samples.length * 4 )
+				.order( ByteOrder.LITTLE_ENDIAN );
 		
 		ShortBuffer shortBuffer = buffer.asShortBuffer();
 
@@ -178,72 +189,72 @@ public class StereoAudioOutput extends AudioOutput
 		@Override
 		public void run()
 		{
-			/* The mProcessing flag ensures that only one instance of the
-			 * processor can run at any given time */
-			if( mProcessing.compareAndSet( false, true ) )
+			try
 			{
-				List<AudioPacket> packets = new ArrayList<AudioPacket>();
-				
-				mBuffer.drainTo( packets );
-				
-				if( packets.size() > 0 )
+				/* The mProcessing flag ensures that only one instance of the
+				 * processor can run at any given time */
+				if( mProcessing.compareAndSet( false, true ) )
 				{
-					broadcast( mAudioContinuationEvent );
+					List<AudioPacket> packets = new ArrayList<AudioPacket>();
 					
-					for( AudioPacket packet: packets )
+					mBuffer.drainTo( packets );
+					
+					if( packets.size() > 0 )
 					{
-						if( mCanProcessAudio )
+						
+						for( AudioPacket packet: packets )
 						{
-							if( isSquelched() )
+							broadcast( mAudioContinuationEvent );
+							
+							if( mCanProcessAudio )
 							{
-								checkStop();
-							}
-							else if( packet.getType() == Type.AUDIO )
-							{
-								ByteBuffer buffer = convert( 
-										packet.getAudioBuffer().getSamples() );
-								
-								mOutput.write( buffer.array(), 0, 
-										buffer.array().length );
-								
-								checkStart();
-								
-								broadcast( packet.getAudioMetadata() );
-								
-								mLastActivity = System.currentTimeMillis();
+								if( packet.getType() == Type.AUDIO )
+								{
+									ByteBuffer buffer = convert( 
+											packet.getAudioBuffer().getSamples() );
+									
+									mOutput.write( buffer.array(), 0, 
+											buffer.array().length );
+
+									checkStart();
+									
+									broadcast( packet.getAudioMetadata() );
+									
+									mLastActivity = System.currentTimeMillis();
+								}
 							}
 						}
 					}
+					else
+					{
+						checkStop();
+					}
+					
+					mProcessing.set( false );
 				}
-				else
-				{
-					checkStop();
-				}
-				
-				mProcessing.set( false );
+			}
+			catch( Exception e )
+			{
+				mLog.error( "Error while processing audio buffers", e );
 			}
 		}
 		
 		private void checkStart()
 		{
-			if( mCanProcessAudio && !isSquelched() && 
+			if( mCanProcessAudio && 
 				!mOutput.isRunning() && mOutput.available() <= START_THRESHOLD )
 			{
 				mOutput.start();
-				
-				mAudioEventBroadcaster.broadcast( mAudioStartEvent );
 			}
 		}
 		
 		private void checkStop()
 		{
 			if( mCanProcessAudio && mOutput.isRunning() &&
-				( mOutput.available() >= STOP_THRESHOLD || isSquelched() ) )
+				mOutput.available() >= STOP_THRESHOLD )
 			{
 				mOutput.drain();
 				mOutput.stop();
-
-				mAudioEventBroadcaster.broadcast( mAudioStopEvent );
 			}
 		}
 	}
@@ -282,4 +293,20 @@ public class StereoAudioOutput extends AudioOutput
 	{
 		return mGainControl != null;
 	}
-}
+	
+
+	@Override
+	public void update( LineEvent event )
+	{
+		LineEvent.Type type = event.getType();
+
+		if( type == LineEvent.Type.START )
+		{
+			mAudioEventBroadcaster.broadcast( mAudioStartEvent );
+		}
+		else if( type == LineEvent.Type.STOP )
+		{
+			mAudioEventBroadcaster.broadcast( mAudioStopEvent );
+		}
+	}
+}