From ddfefe4b85afd61d27d8d33812b27924ee614da8 Mon Sep 17 00:00:00 2001 From: denny Date: Tue, 2 Feb 2016 19:44:18 -0500 Subject: [PATCH] Resolves issue with ChannelState FADE monitor - changes from starting/stopping the monitor for each call to simply leaving the monitor task running continuously. Was causing issue with scheduled thread executor not shutting down the task in a timely manner -- new approach should avoid start/stop related issues altogether. Updates P25DecoderState with reordering of timing of when we send updates for the TO/FROM identifiers so that the audio manager receives up-to-date metadata and hopefully resolves issue with 'Do Not Monitor' talkgroups being monitored by the audio manager. Updates the AudioManager with a try/catch block around the processing task so that we can catch any audio processing exceptions that occur during the timer task and might get suppressed by the scheduled thread executor. --- src/audio/AudioManager.java | 37 +-- src/module/decode/p25/P25DecoderState.java | 248 ++++++++++----------- src/module/decode/state/ChannelState.java | 106 ++++----- 3 files changed, 176 insertions(+), 215 deletions(-) diff --git a/src/audio/AudioManager.java b/src/audio/AudioManager.java index eca4675b..809105eb 100644 --- a/src/audio/AudioManager.java +++ b/src/audio/AudioManager.java @@ -17,7 +17,6 @@ package audio; * You should have received a copy of the GNU General Public License * along with this program. If not, see ******************************************************************************/ - import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -41,7 +40,6 @@ import source.mixer.MixerChannel; import source.mixer.MixerChannelConfiguration; import source.mixer.MixerManager; import audio.AudioEvent.Type; -import audio.AudioPanel.MixerSelectionItem; import audio.output.AudioOutput; import audio.output.MonoAudioOutput; import audio.output.StereoAudioOutput; @@ -392,29 +390,36 @@ public class AudioManager implements Listener, IAudioController @Override public void run() { - disconnectInactiveChannelAssignments(); - - if( mAudioPacketQueue != null ) + try { - List packets = new ArrayList(); - - mAudioPacketQueue.drainTo( packets ); + disconnectInactiveChannelAssignments(); - for( AudioPacket packet: packets ) + if( mAudioPacketQueue != null ) { - /* Don't process any packet's marked as do not monitor */ - if( !packet.getAudioMetadata().isDoNotMonitor() && - packet.getType() == AudioPacket.Type.AUDIO ) - { - AudioOutputConnection connection = getConnection( packet ); + List packets = new ArrayList(); - if( connection != null ) + mAudioPacketQueue.drainTo( packets ); + + for( AudioPacket packet: packets ) + { + /* Don't process any packet's marked as do not monitor */ + if( !packet.getAudioMetadata().isDoNotMonitor() && + packet.getType() == AudioPacket.Type.AUDIO ) { - connection.receive( packet ); + AudioOutputConnection connection = getConnection( packet ); + + if( connection != null ) + { + connection.receive( packet ); + } } } } } + catch( Exception e ) + { + mLog.error( "Encountered error while processing audio packets", e ); + } } } diff --git a/src/module/decode/p25/P25DecoderState.java b/src/module/decode/p25/P25DecoderState.java index 6c096666..b07d0197 100644 --- a/src/module/decode/p25/P25DecoderState.java +++ b/src/module/decode/p25/P25DecoderState.java @@ -405,12 +405,12 @@ public class P25DecoderState extends DecoderState { if( hdu.isValid() ) { - State state = hdu.isEncrypted() ? State.ENCRYPTED : State.CALL; - - broadcast( new DecoderStateEvent( this, Event.START, state ) ); + broadcast( new DecoderStateEvent( this, Event.START, State.CALL ) ); String to = hdu.getToID(); - + + updateTo( to ); + if( mCurrentCallEvent == null ) { mCurrentCallEvent = new P25CallEvent.Builder( CallEventType.CALL ) @@ -424,7 +424,6 @@ public class P25DecoderState extends DecoderState broadcast( mCurrentCallEvent ); } - updateTo( to ); } } @@ -614,12 +613,14 @@ public class P25DecoderState extends DecoderState module.decode.p25.message.tdu.lc.GroupVoiceChannelUser gvcuser = (module.decode.p25.message.tdu.lc.GroupVoiceChannelUser)tdulc; - broadcast( new DecoderStateEvent( this, Event.CONTINUATION, - ( gvcuser.isEncrypted() ? State.ENCRYPTED : State.CALL ) ) ); + broadcast( new DecoderStateEvent( this, Event.CONTINUATION, State.CALL ) ); String from = gvcuser.getSourceAddress(); String to = gvcuser.getGroupAddress(); + updateFrom( from ); + updateTo( to ); + if( mCurrentCallEvent == null ) { mCurrentCallEvent = new P25CallEvent.Builder( CallEventType.GROUP_CALL ) @@ -635,9 +636,6 @@ public class P25DecoderState extends DecoderState broadcast( mCurrentCallEvent ); } - - updateFrom( from ); - updateTo( to ); } break; case MESSAGE_UPDATE: @@ -844,6 +842,8 @@ public class P25DecoderState extends DecoderState String to = tivcu.getAddress(); + updateTo( to ); + if( mCurrentCallEvent == null ) { mCurrentCallEvent = new P25CallEvent.Builder( CallEventType.TELEPHONE_INTERCONNECT ) @@ -857,8 +857,6 @@ public class P25DecoderState extends DecoderState broadcast( mCurrentCallEvent ); } - - updateTo( to ); } break; case UNIT_AUTHENTICATION_COMMAND: @@ -921,7 +919,10 @@ public class P25DecoderState extends DecoderState (module.decode.p25.message.tdu.lc.UnitToUnitVoiceChannelUser)tdulc; String from = uuvcu.getSourceAddress(); + updateFrom( from ); + String to = uuvcu.getTargetAddress(); + updateTo( to ); if( mCurrentCallEvent != null ) { @@ -935,9 +936,6 @@ public class P25DecoderState extends DecoderState .to( to ) .build(); } - - updateFrom( from ); - updateTo( to ); } break; default: @@ -949,35 +947,14 @@ public class P25DecoderState extends DecoderState * Log optional vendor format messages that we don't yet support so that we * can understand those messages and eventually add support. */ - private void logAlternateVendorMessage( P25Message p25 ) + private void logAlternateVendorMessage( P25Message message ) { -// if( p25 instanceof TDULinkControlMessage ) +// if( message.isValid() ) // { -// TDULinkControlMessage tdulc = (TDULinkControlMessage)p25; -// -// mLog.debug( "TDULC - VENDOR [" + tdulc.getVendor() + -// "] SPECIFIC FORMAT FOR " + tdulc.getVendorOpcode().getLabel() + -// " " + tdulc.getBinaryMessage() ); -// } -// else if( p25 instanceof LDU1Message ) -// { -// LDU1Message ldu1 = (LDU1Message)p25; -// -// mLog.debug( "LDU1 - VENDOR [" + ldu1.getVendor() + -// "] SPECIFIC FORMAT FOR " + ldu1.getVendorOpcode().getLabel() + -// " " + ldu1.getBinaryMessage() ); -// } -// else if( p25 instanceof PDUMessage ) -// { -// PDUMessage pdu = (PDUMessage)p25; -// -// mLog.debug( "PDU - VENDOR [" + pdu.getVendor() + -// "] SPECIFIC FORMAT FOR " + pdu.getVendorOpcode().getLabel() + -// " " + pdu.getBinaryMessage() ); -// } -// else -// { -// mLog.debug( "P25 - ALTERNATE VENDOR MESSAGE [" + p25.getClass() + "]" ); +// mLog.info( "PLEASE NOTIFY DEVELOPER - UNRECOGNIZED P25 VENDOR FORMAT -" +// + " DUID:" + message.getDUID() +// + " MSG:" + message.getBinaryMessage() +// + " CLASS:" + message.getClass() ); // } } @@ -988,15 +965,11 @@ public class P25DecoderState extends DecoderState */ private void processLDU( LDUMessage ldu ) { - State state = ( ldu.isValid() && ldu.isEncrypted() ) ? - State.ENCRYPTED : State.CALL; - - broadcast( new DecoderStateEvent( this, Event.CONTINUATION, state ) ); + broadcast( new DecoderStateEvent( this, Event.CONTINUATION, State.CALL ) ); if( mCurrentCallEvent == null ) { - mCurrentCallEvent = new P25CallEvent.Builder( - ( state == State.CALL ) ? CallEventType.CALL : CallEventType.ENCRYPTED_CALL ) + mCurrentCallEvent = new P25CallEvent.Builder( CallEventType.CALL ) .aliasList( getAliasList() ) .channel( mCurrentChannel ) .frequency( mCurrentChannelFrequency ) @@ -1114,42 +1087,43 @@ public class P25DecoderState extends DecoderState } break; case GROUP_VOICE_CHANNEL_UPDATE: - if( mChannelType == ChannelType.STANDARD && - ldu instanceof module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdate ) + if( ldu instanceof module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdate ) { module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdate gvcu = - (module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdate)ldu; + (module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdate)ldu; String userA = gvcu.getGroupAddressA(); - - if( !mCallDetectTalkgroups.contains( userA ) ) - { - broadcast( new P25CallEvent.Builder( CallEventType.CALL_DETECT ) - .aliasList( getAliasList() ) - .channel( gvcu.getChannelA() ) - .details( ( gvcu.isEncrypted() ? "ENCRYPTED" : "" ) ) - .frequency( gvcu.getDownlinkFrequencyA() ) - .to( userA ) - .build() ); - - mCallDetectTalkgroups.add( userA ); - } - String userB = gvcu.getGroupAddressB(); - if( userB != null && - !userB.contentEquals( "0000" ) && - !mCallDetectTalkgroups.contains( userB ) ) + if( mChannelType == ChannelType.STANDARD ) { - broadcast( new P25CallEvent.Builder( CallEventType.CALL_DETECT ) - .aliasList( getAliasList() ) - .channel( gvcu.getChannelB() ) - .details( ( gvcu.isEncrypted() ? "ENCRYPTED" : "" ) ) - .frequency( gvcu.getDownlinkFrequencyB() ) - .to( userB ) - .build() ); + if( !mCallDetectTalkgroups.contains( userA ) ) + { + broadcast( new P25CallEvent.Builder( CallEventType.CALL_DETECT ) + .aliasList( getAliasList() ) + .channel( gvcu.getChannelA() ) + .details( ( gvcu.isEncrypted() ? "ENCRYPTED" : "" ) ) + .frequency( gvcu.getDownlinkFrequencyA() ) + .to( userA ) + .build() ); + + mCallDetectTalkgroups.add( userA ); + } - mCallDetectTalkgroups.add( userB ); + if( userB != null && + !userB.contentEquals( "0000" ) && + !mCallDetectTalkgroups.contains( userB ) ) + { + broadcast( new P25CallEvent.Builder( CallEventType.CALL_DETECT ) + .aliasList( getAliasList() ) + .channel( gvcu.getChannelB() ) + .details( ( gvcu.isEncrypted() ? "ENCRYPTED" : "" ) ) + .frequency( gvcu.getDownlinkFrequencyB() ) + .to( userB ) + .build() ); + + mCallDetectTalkgroups.add( userB ); + } } } else @@ -1158,25 +1132,27 @@ public class P25DecoderState extends DecoderState } break; case GROUP_VOICE_CHANNEL_UPDATE_EXPLICIT: - if( mChannelType == ChannelType.STANDARD && - ldu instanceof module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdateExplicit ) + if( ldu instanceof module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdateExplicit ) { module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdateExplicit gvcue = (module.decode.p25.message.ldu.lc.GroupVoiceChannelUpdateExplicit)ldu; String group = gvcue.getGroupAddress(); - if( !mCallDetectTalkgroups.contains( group ) ) + if( mChannelType == ChannelType.STANDARD ) { - broadcast( new P25CallEvent.Builder( CallEventType.CALL_DETECT ) - .aliasList( getAliasList() ) - .channel( gvcue.getTransmitChannel() ) - .details( ( gvcue.isEncrypted() ? "ENCRYPTED" : "" ) ) - .frequency( gvcue.getDownlinkFrequency() ) - .to( group ) - .build() ); - - mCallDetectTalkgroups.add( group ); + if( !mCallDetectTalkgroups.contains( group ) ) + { + broadcast( new P25CallEvent.Builder( CallEventType.CALL_DETECT ) + .aliasList( getAliasList() ) + .channel( gvcue.getTransmitChannel() ) + .details( ( gvcue.isEncrypted() ? "ENCRYPTED" : "" ) ) + .frequency( gvcue.getDownlinkFrequency() ) + .to( group ) + .build() ); + + mCallDetectTalkgroups.add( group ); + } } } else @@ -1186,29 +1162,32 @@ public class P25DecoderState extends DecoderState break; case GROUP_VOICE_CHANNEL_USER: /* Indicates the current user of the current channel */ - if( mChannelType == ChannelType.STANDARD && - ldu instanceof module.decode.p25.message.ldu.lc.GroupVoiceChannelUser ) + if( ldu instanceof module.decode.p25.message.ldu.lc.GroupVoiceChannelUser ) { module.decode.p25.message.ldu.lc.GroupVoiceChannelUser gvcuser = (module.decode.p25.message.ldu.lc.GroupVoiceChannelUser)ldu; - if( mCurrentCallEvent.getCallEventType() != CallEventType.GROUP_CALL ) - { - mCurrentCallEvent.setCallEventType( CallEventType.GROUP_CALL ); - broadcast( mCurrentCallEvent ); - } - - if( mCurrentCallEvent.getDetails() == null ) - { - mCurrentCallEvent.setDetails( - ( gvcuser.isEncrypted() ? "ENCRYPTED " : "" ) + - ( gvcuser.isEmergency() ? "EMERGENCY " : "" ) ); - - broadcast( mCurrentCallEvent ); - } - updateFrom( gvcuser.getSourceAddress() ); + updateTo( gvcuser.getGroupAddress() ); + + if( mChannelType == ChannelType.STANDARD ) + { + if( mCurrentCallEvent.getCallEventType() != CallEventType.GROUP_CALL ) + { + mCurrentCallEvent.setCallEventType( CallEventType.GROUP_CALL ); + broadcast( mCurrentCallEvent ); + } + + if( mCurrentCallEvent.getDetails() == null ) + { + mCurrentCallEvent.setDetails( + ( gvcuser.isEncrypted() ? "ENCRYPTED " : "" ) + + ( gvcuser.isEmergency() ? "EMERGENCY " : "" ) ); + + broadcast( mCurrentCallEvent ); + } + } } else { @@ -1404,6 +1383,8 @@ public class P25DecoderState extends DecoderState TelephoneInterconnectVoiceChannelUser tivcu = (TelephoneInterconnectVoiceChannelUser)ldu; + updateTo( tivcu.getAddress() ); + if( mCurrentCallEvent.getCallEventType() != CallEventType.TELEPHONE_INTERCONNECT ) { mCurrentCallEvent.setCallEventType( CallEventType.TELEPHONE_INTERCONNECT ); @@ -1418,8 +1399,6 @@ public class P25DecoderState extends DecoderState broadcast( mCurrentCallEvent ); } - - updateTo( tivcu.getAddress() ); } else { @@ -1485,6 +1464,10 @@ public class P25DecoderState extends DecoderState UnitToUnitVoiceChannelUser uuvcu = (UnitToUnitVoiceChannelUser)ldu; + updateFrom( uuvcu.getSourceAddress() ); + + updateTo( uuvcu.getTargetAddress() ); + if( mCurrentCallEvent.getCallEventType() != CallEventType.UNIT_TO_UNIT_CALL ) { mCurrentCallEvent.setCallEventType( CallEventType.UNIT_TO_UNIT_CALL ); @@ -1499,9 +1482,6 @@ public class P25DecoderState extends DecoderState broadcast( mCurrentCallEvent ); } - - updateFrom( uuvcu.getSourceAddress() ); - updateTo( uuvcu.getTargetAddress() ); } else { @@ -3101,16 +3081,9 @@ public class P25DecoderState extends DecoderState if( mToTalkgroup == null || !mToTalkgroup.contentEquals( to ) ) { mToTalkgroup = to; + broadcast( ChangedAttribute.TO_TALKGROUP ); - if( mCurrentCallEvent != null && - ( mCurrentCallEvent.getToID() == null || - !mCurrentCallEvent.getToID().contentEquals( to ) ) ) - { - mCurrentCallEvent.setToID( to ); - broadcast( mCurrentCallEvent ); - } - if( hasAliasList() ) { mToAlias = getAliasList().getTalkgroupAlias( mToTalkgroup ); @@ -3119,12 +3092,20 @@ public class P25DecoderState extends DecoderState { mToAlias = null; } - - broadcast( ChangedAttribute.TO_TALKGROUP_ALIAS ); - } - /* Send audio metadata update */ - broadcast( new Metadata( MetadataType.TO, mToTalkgroup, mToAlias, true ) ); + broadcast( ChangedAttribute.TO_TALKGROUP_ALIAS ); + + /* Send audio metadata update */ + broadcast( new Metadata( MetadataType.TO, mToTalkgroup, mToAlias, true ) ); + + if( mCurrentCallEvent != null && + ( mCurrentCallEvent.getToID() == null || + !mCurrentCallEvent.getToID().contentEquals( to ) ) ) + { + mCurrentCallEvent.setToID( to ); + broadcast( mCurrentCallEvent ); + } + } } } @@ -3140,14 +3121,6 @@ public class P25DecoderState extends DecoderState mFromTalkgroup = from; broadcast( ChangedAttribute.FROM_TALKGROUP ); - if( mCurrentCallEvent != null && - mCurrentCallEvent.getFromID() == null || - !mCurrentCallEvent.getFromID().contentEquals( from ) ) - { - mCurrentCallEvent.setFromID( from ); - broadcast( mCurrentCallEvent ); - } - if( hasAliasList() ) { mFromAlias = getAliasList().getTalkgroupAlias( mFromTalkgroup ); @@ -3158,10 +3131,19 @@ public class P25DecoderState extends DecoderState } broadcast( ChangedAttribute.FROM_TALKGROUP_ALIAS ); + + /* Send audio metadata update */ + broadcast( new Metadata( MetadataType.FROM, mFromTalkgroup, mFromAlias, true ) ); + + if( mCurrentCallEvent != null && + mCurrentCallEvent.getFromID() == null || + !mCurrentCallEvent.getFromID().contentEquals( from ) ) + { + mCurrentCallEvent.setFromID( from ); + broadcast( mCurrentCallEvent ); + } } - /* Send audio metadata update */ - broadcast( new Metadata( MetadataType.FROM, mFromTalkgroup, mFromAlias, true ) ); } } diff --git a/src/module/decode/state/ChannelState.java b/src/module/decode/state/ChannelState.java index 71f999c9..927d1a44 100644 --- a/src/module/decode/state/ChannelState.java +++ b/src/module/decode/state/ChannelState.java @@ -82,7 +82,8 @@ public class ChannelState extends Module implements ICallEventProvider, /** * Channel state tracks the overall state of all processing modules and decoders - * configured for the channel and provides squelch control events. + * configured for the channel and provides squelch control and decoder state + * reset events. * * Uses a state enumeration that defines allowable channel state transitions in * order to track a call or data decode event from start to finish. Uses a @@ -127,19 +128,54 @@ public class ChannelState extends Module implements ICallEventProvider, @Override public void start() { + //Start a monitoring task to check for FADE and END event timeouts + if( mMonitoring.compareAndSet( false, true ) ) + { + if( mStateMonitorFuture == null && mStateMonitor != null ) + { + try + { + mStateMonitorFuture = mThreadPoolManager.scheduleFixedRate( + ThreadType.DECODER, mStateMonitor, 20, TimeUnit.MILLISECONDS ); + } + catch( RejectedExecutionException ree ) + { + mLog.error( "state monitor scheduled task rejected", ree ); + mMonitoring.set( false ); + } + } + else + { + throw new RuntimeException( "Channel state monitor's scheduled " + + "future pointer was not null" ); + } + } } @Override public void stop() { mSquelchLocked = false; - stopMonitor(); + + //Shutdown the state montoring task + if( mMonitoring.compareAndSet( true, false ) ) + { + if( mStateMonitorFuture != null ) + { + boolean success = mThreadPoolManager.cancel( mStateMonitorFuture ); + + if( !success ) + { + mLog.error("Couldn't stop monitoring scheduled future" ); + } + } + + mStateMonitorFuture = null; + } } public void dispose() { - stopMonitor(); - mCallEventListener = null; mChangedAttributeListener = null; mDecoderStateListener = null; @@ -193,57 +229,6 @@ public class ChannelState extends Module implements ICallEventProvider, return mState; } - /** - * Starts the state monitor running when we're not in IDLE state. Calls to - * start an already started monitor are ignored. - */ - private void startMonitor() - { - if( mMonitoring.compareAndSet( false, true ) ) - { - if( mStateMonitorFuture == null && mStateMonitor != null ) - { - try - { - mStateMonitorFuture = mThreadPoolManager.scheduleFixedRate( - ThreadType.DECODER, mStateMonitor, 20, TimeUnit.MILLISECONDS ); - } - catch( RejectedExecutionException ree ) - { - mLog.error( "state monitor scheduled task rejected", ree ); - mMonitoring.set( false ); - } - } - else - { - throw new RuntimeException( "Channel state monitor's scheduled " - + "future pointer was not null" ); - } - } - } - - /** - * Stops the state monitor when we're in an IDLE state. Calls to stop an - * already stopped monitor are ignored. - */ - private void stopMonitor() - { - if( mMonitoring.compareAndSet( true, false ) ) - { - if( mStateMonitorFuture != null ) - { - boolean success = mThreadPoolManager.cancel( mStateMonitorFuture ); - - if( !success ) - { - mLog.error("Couldn't stop monitoring scheduled future" ); - } - } - - mStateMonitorFuture = null; - } - } - /** * Updates the fade timeout threshold to the current time plus delay */ @@ -332,7 +317,6 @@ public class ChannelState extends Module implements ICallEventProvider, { broadcast( SquelchState.SQUELCH ); updateFadeTimeout(); - startMonitor(); mState = state; broadcast( ChangedAttribute.CHANNEL_STATE ); } @@ -341,14 +325,12 @@ public class ChannelState extends Module implements ICallEventProvider, case ENCRYPTED: broadcast( SquelchState.SQUELCH ); updateFadeTimeout(); - startMonitor(); mState = state; broadcast( ChangedAttribute.CHANNEL_STATE ); break; case CALL: broadcast( SquelchState.UNSQUELCH ); updateFadeTimeout(); - startMonitor(); mState = state; broadcast( ChangedAttribute.CHANNEL_STATE ); break; @@ -368,10 +350,6 @@ public class ChannelState extends Module implements ICallEventProvider, break; } } - else if( mState == State.ENCRYPTED && state == State.CALL ) - { - updateFadeTimeout(); - } else { mLog.debug( "Can't change from [" + mState + "] to [" + state + "]" ); @@ -386,7 +364,6 @@ public class ChannelState extends Module implements ICallEventProvider, { updateResetTimeout(); mState = State.FADE; - startMonitor(); broadcast( SquelchState.SQUELCH ); broadcast( ChangedAttribute.CHANNEL_STATE ); @@ -402,7 +379,6 @@ public class ChannelState extends Module implements ICallEventProvider, broadcast( new MetadataReset() ); } - stopMonitor(); mState = State.IDLE; broadcast( ChangedAttribute.CHANNEL_STATE ); @@ -412,8 +388,6 @@ public class ChannelState extends Module implements ICallEventProvider, { broadcast( SquelchState.SQUELCH ); - stopMonitor(); - mState = State.TEARDOWN; broadcast( ChangedAttribute.CHANNEL_STATE );