diff --git a/.idea/dictionaries/denny.xml b/.idea/dictionaries/denny.xml index 24e59925..7f72f826 100644 --- a/.idea/dictionaries/denny.xml +++ b/.idea/dictionaries/denny.xml @@ -7,6 +7,7 @@ fleetsync icecast ident + overflowable polyphase remez samplerate diff --git a/CHANGELOG b/CHANGELOG index b040d09a..5fee2ff5 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,10 +1,10 @@ 0.3.0-beta1 20170129 + - REQUIRES: Java 8 (or newer) for compatibility and eventual changeover to JavaFX - FEATURE: audio streaming - adds capability to stream to multiple remote broadcast servers (Broadcastify, Icecast v2.3 and v2.4+, and Shoutcast V1 and V2). Each stream can be independently configured with a delay. Audio calls are directed to one or more streams by attaching one or more stream identifiers to each alias. - - Requires Java 8 (or newer) for compatibility and eventual changeover to JavaFX - Extensive GUI updates for tuner, channel and alias management. Includes new 'Now Playing' channel view and 'Detail' view. - Updated alias list management of alias id values and wildcards. Weighting is now @@ -17,7 +17,7 @@ - Implemented bounded queues for both tuner and individual channel IQ sample buffers. Bounded queues have ability to enter/exit overflow state when queue size reaches overflow threshold and reset when queue size falls below the reset - threshold. + threshold. Channel state will show OVERFLOW in red/yellow. - Updated airspy controller for airspy-mini specific sample rate handling. - New developer remez filter designer tool and internal usage diff --git a/src/alias/PatchGroupAlias.java b/src/alias/PatchGroupAlias.java index b4788a86..b2c0d2fa 100644 --- a/src/alias/PatchGroupAlias.java +++ b/src/alias/PatchGroupAlias.java @@ -168,11 +168,14 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - for(AliasAction action: alias.getAction()) + if(!(alias instanceof PatchGroupAlias)) { - if(!aliasActions.contains(action)) + for(AliasAction action: alias.getAction()) { - aliasActions.add(action); + if(!aliasActions.contains(action)) + { + aliasActions.add(action); + } } } } @@ -185,7 +188,7 @@ public class PatchGroupAlias extends Alias { for(Alias alias: mPatchedAliases) { - if(alias.hasActions()) + if(!(alias instanceof PatchGroupAlias) && alias.hasActions()) { return true; } @@ -201,7 +204,9 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.hasCallPriority() && alias.getCallPriority() < highestPriority) + if(!(alias instanceof PatchGroupAlias) && + alias.hasCallPriority() && + alias.getCallPriority() < highestPriority) { highestPriority = alias.getCallPriority(); } @@ -220,7 +225,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.hasCallPriority()) + if(!(alias instanceof PatchGroupAlias) && alias.hasCallPriority()) { return true; } @@ -239,7 +244,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.isRecordable()) + if(!(alias instanceof PatchGroupAlias) && alias.isRecordable()) { return true; } @@ -258,7 +263,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.isStreamable()) + if(!(alias instanceof PatchGroupAlias) && alias.isStreamable()) { return true; } @@ -279,11 +284,14 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - for(BroadcastChannel broadcastChannel: alias.getBroadcastChannels()) + if(!(alias instanceof PatchGroupAlias)) { - if(!broadcastChannels.contains(broadcastChannel)) + for(BroadcastChannel broadcastChannel: alias.getBroadcastChannels()) { - broadcastChannels.add(broadcastChannel); + if(!broadcastChannels.contains(broadcastChannel)) + { + broadcastChannels.add(broadcastChannel); + } } } } @@ -301,7 +309,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.hasBroadcastChannel(channel)) + if(!(alias instanceof PatchGroupAlias) && alias.hasBroadcastChannel(channel)) { return true; } diff --git a/src/channel/metadata/ChannelMetadataPanel.java b/src/channel/metadata/ChannelMetadataPanel.java index e9ecef00..5fa80231 100644 --- a/src/channel/metadata/ChannelMetadataPanel.java +++ b/src/channel/metadata/ChannelMetadataPanel.java @@ -210,6 +210,10 @@ public class ChannelMetadataPanel extends JPanel implements ListSelectionListene background = SystemProperties.getInstance().get(PROPERTY_PREFIX_BACKGROUND + "idle", Color.WHITE); foreground = SystemProperties.getInstance().get(PROPERTY_PREFIX_FOREGROUND + "idle", Color.BLUE); break; + case OVERFLOW: + background = SystemProperties.getInstance().get(PROPERTY_PREFIX_BACKGROUND + "overflow", Color.RED); + foreground = SystemProperties.getInstance().get(PROPERTY_PREFIX_FOREGROUND + "overflow", Color.YELLOW); + break; case RESET: background = SystemProperties.getInstance().get(PROPERTY_PREFIX_BACKGROUND + "reset", Color.PINK); foreground = SystemProperties.getInstance().get(PROPERTY_PREFIX_FOREGROUND + "reset", Color.YELLOW); diff --git a/src/channel/metadata/ChannelMetadataViewer.java b/src/channel/metadata/ChannelMetadataViewer.java index 53996650..e7efc2c2 100644 --- a/src/channel/metadata/ChannelMetadataViewer.java +++ b/src/channel/metadata/ChannelMetadataViewer.java @@ -51,7 +51,7 @@ public class ChannelMetadataViewer extends JPanel mChannelDetailPanel = new ChannelDetailPanel(channelProcessingManager); mCallEventPanel = new CallEventPanel(iconManager); mMessageActivityPanel = new MessageActivityPanel(channelProcessingManager); - mChannelSpectrumPanel = new ChannelSpectrumPanel(settingsManager, channelProcessingManager); +// mChannelSpectrumPanel = new ChannelSpectrumPanel(settingsManager, channelProcessingManager); mChannelMetadataPanel = new ChannelMetadataPanel(channelProcessingManager, iconManager); init(); diff --git a/src/channel/state/ChannelState.java b/src/channel/state/ChannelState.java index 66a22984..768c1062 100644 --- a/src/channel/state/ChannelState.java +++ b/src/channel/state/ChannelState.java @@ -35,6 +35,7 @@ import module.decode.event.ICallEventProvider; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import sample.Listener; +import sample.real.IOverflowListener; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; @@ -42,7 +43,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; public class ChannelState extends Module implements ICallEventProvider, IDecoderStateEventListener, - IDecoderStateEventProvider, ISquelchStateProvider, IAttributeChangeRequestListener + IDecoderStateEventProvider, ISquelchStateProvider, IAttributeChangeRequestListener, IOverflowListener { private final static Logger mLog = LoggerFactory.getLogger(ChannelState.class); @@ -63,6 +64,7 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder private boolean mSquelchLocked = false; private boolean mSelected = false; + private boolean mSourceOverflow = false; private long mStandardChannelFadeTimeout = FADE_TIMEOUT_DELAY; private long mTrafficChannelFadeTimeout = DecodeConfiguration.DEFAULT_CALL_TIMEOUT_SECONDS * 1000; @@ -123,6 +125,7 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder mState = State.IDLE; mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, mState)); + mSourceOverflow = false; } @Override @@ -350,15 +353,26 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder break; } - mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, mState)); - } - else - { -// mLog.debug( "Can't change from [" + mState + "] to [" + state + "]" ); + mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, + mSourceOverflow ? State.OVERFLOW : mState)); } } } + /** + * This method is invoked if the source buffer provider goes into overflow state. Since this is an external state, + * we use the mSourceOverflow variable to override the internal state reported to external listeners. + * @param overflow true to indicate an overflow state + */ + @Override + public void sourceOverflow(boolean overflow) + { + mSourceOverflow = overflow; + + mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, + mSourceOverflow ? State.OVERFLOW : mState)); + } + /** * Sets the state and processes related actions */ diff --git a/src/channel/state/DecoderState.java b/src/channel/state/DecoderState.java index a27dd7be..8f7ecbf8 100644 --- a/src/channel/state/DecoderState.java +++ b/src/channel/state/DecoderState.java @@ -51,6 +51,14 @@ public abstract class DecoderState extends Module implements ActivitySummaryProv return mAliasList != null; } + /** + * Provides subclass reference to the call event broadcaster + */ + protected Broadcaster getCallEventBroadcaster() + { + return mCallEventBroadcaster; + } + /** * Reset the decoder state to prepare for processing a different sample * source diff --git a/src/channel/state/State.java b/src/channel/state/State.java index 83336d2c..f2c75767 100644 --- a/src/channel/state/State.java +++ b/src/channel/state/State.java @@ -1,17 +1,17 @@ /******************************************************************************* * SDR Trunk * Copyright (C) 2014-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 ******************************************************************************/ @@ -23,98 +23,94 @@ import java.util.EnumSet; * Details the set of states for a channel and the allowable transition states */ public enum State -{ - IDLE( "IDLE" ) - { - @Override - public boolean canChangeTo( State state ) +{ + IDLE("IDLE") { - return state != TEARDOWN && - state != RESET; - } - }, - CALL( "CALL" ) - { - @Override - public boolean canChangeTo( State state ) + @Override + public boolean canChangeTo(State state) + { + return state != TEARDOWN && state != RESET && state != OVERFLOW; + } + }, + CALL("CALL") { - return state == CONTROL || - state == DATA || - state == ENCRYPTED || - state == FADE || - state == TEARDOWN; - } - }, - DATA( "DATA" ) - { - @Override - public boolean canChangeTo( State state ) + @Override + public boolean canChangeTo(State state) + { + return state == CONTROL || state == DATA || state == ENCRYPTED || state == FADE || state == TEARDOWN; + } + }, + DATA("DATA") { - return state == CALL || - state == CONTROL || - state == ENCRYPTED || - state == FADE || - state == TEARDOWN; - } - }, - ENCRYPTED( "ENCRYPTED" ) + @Override + public boolean canChangeTo(State state) + { + return state == CALL || state == CONTROL || state == ENCRYPTED || state == FADE || state == TEARDOWN; + } + }, + ENCRYPTED("ENCRYPTED") + { + @Override + public boolean canChangeTo(State state) + { + return state == FADE || state == TEARDOWN; + } + }, + CONTROL("CONTROL") + { + @Override + public boolean canChangeTo(State state) + { + return state == IDLE || state == FADE; + } + }, + FADE("FADE") + { + @Override + public boolean canChangeTo(State state) + { + return state != FADE && state != RESET && state != OVERFLOW; + } + }, + OVERFLOW("OVERFLOW") + { + @Override + public boolean canChangeTo(State state) + { + return false; + } + }, + RESET("RESET") + { + @Override + public boolean canChangeTo(State state) + { + return state == IDLE; + } + }, + TEARDOWN("TEARDOWN") + { + @Override + public boolean canChangeTo(State state) + { + return state == RESET; + } + }; + + private String mDisplayValue; + + public static final EnumSet CALL_STATES = EnumSet.of(CALL, CONTROL, DATA, ENCRYPTED); + public static final EnumSet IDLE_STATES = EnumSet.of(IDLE, FADE); + + private State(String displayValue) { - @Override - public boolean canChangeTo( State state ) - { - return state == FADE || state == TEARDOWN; - } - }, - CONTROL( "CONTROL" ) + mDisplayValue = displayValue; + } + + public abstract boolean canChangeTo(State state); + + public String getDisplayValue() { - @Override - public boolean canChangeTo( State state ) - { - return state == IDLE || - state == FADE; - } - }, - FADE( "FADE" ) - { - @Override - public boolean canChangeTo( State state ) - { - return state != FADE && - state != RESET; - } - }, - RESET( "RESET" ) - { - @Override - public boolean canChangeTo(State state) - { - return state == IDLE; - } - }, - - TEARDOWN( "TEARDOWN" ) - { - @Override - public boolean canChangeTo(State state) - { - return state == RESET; - } - }; - - private String mDisplayValue; - - public static final EnumSet CALL_STATES = EnumSet.of( CALL, CONTROL, DATA, ENCRYPTED ); - public static final EnumSet IDLE_STATES = EnumSet.of( IDLE, FADE ); - - private State( String displayValue ) - { - mDisplayValue = displayValue; - } - - public abstract boolean canChangeTo( State state ); - - public String getDisplayValue() - { - return mDisplayValue; - } + return mDisplayValue; + } } \ No newline at end of file diff --git a/src/controller/NamingThreadFactory.java b/src/controller/NamingThreadFactory.java index ae3302b1..ed7bd0dc 100644 --- a/src/controller/NamingThreadFactory.java +++ b/src/controller/NamingThreadFactory.java @@ -1,5 +1,9 @@ package controller; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import util.ThreadPool; + import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; @@ -8,24 +12,23 @@ import java.util.concurrent.atomic.AtomicInteger; */ public class NamingThreadFactory implements ThreadFactory { - static final AtomicInteger poolNumber = new AtomicInteger(1); + private final static Logger mLog = LoggerFactory.getLogger(ThreadPool.class); + + private static final AtomicInteger mPoolNumber = new AtomicInteger(1); - final ThreadGroup mThreadGroup; + private final ThreadGroup mThreadGroup; - final AtomicInteger mThreadNumber = new AtomicInteger(1); + private final AtomicInteger mThreadNumber = new AtomicInteger(1); - final String mNamePrefix; + private final String mNamePrefix; public NamingThreadFactory( String prefix ) { SecurityManager s = System.getSecurityManager(); - mThreadGroup = (s != null)? s.getThreadGroup() : - Thread.currentThread().getThreadGroup(); + mThreadGroup = (s != null)? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); - mNamePrefix = prefix + " pool-" + - poolNumber.getAndIncrement() + - "-thread-"; + mNamePrefix = prefix + " pool-" + mPoolNumber.getAndIncrement() + "-thread-"; } public Thread newThread( Runnable runnable ) @@ -42,7 +45,16 @@ public class NamingThreadFactory implements ThreadFactory { thread.setPriority( Thread.NORM_PRIORITY ); } - + + thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() + { + @Override + public void uncaughtException(Thread t, Throwable e) + { + mLog.error("Error while executing runnable in scheduled thread pool [" + t.getName() + "]", e); + } + }); + return thread; } } \ No newline at end of file diff --git a/src/module/ProcessingChain.java b/src/module/ProcessingChain.java index d8a13e9a..fdf60e86 100644 --- a/src/module/ProcessingChain.java +++ b/src/module/ProcessingChain.java @@ -538,6 +538,9 @@ public class ProcessingChain implements IChannelEventListener module.reset(); } + //Setup the channel state to monitor source overflow conditions + mSource.setOverflowListener(mChannelState); + /* Register with the source to receive sample data. Setup a * timer task to process the buffer queues 50 times a second * (every 20 ms) */ @@ -606,6 +609,7 @@ public class ProcessingChain implements IChannelEventListener if(mSource != null) { mSource.stop(); + mSource.setOverflowListener(null); switch(mSource.getSampleType()) { diff --git a/src/module/decode/p25/P25DecoderState.java b/src/module/decode/p25/P25DecoderState.java index 156c4fdd..2b835de3 100644 --- a/src/module/decode/p25/P25DecoderState.java +++ b/src/module/decode/p25/P25DecoderState.java @@ -19,9 +19,7 @@ package module.decode.p25; import alias.Alias; import alias.AliasList; -import alias.PatchGroupAlias; import alias.id.AliasIDType; -import alias.id.talkgroup.TalkgroupID; import channel.metadata.AliasedStringAttributeMonitor; import channel.metadata.Attribute; import channel.metadata.AttributeChangeRequest; @@ -163,7 +161,6 @@ public class P25DecoderState extends DecoderState private Map mBands = new HashMap<>(); private Map mRegistrations = new HashMap<>(); private Map mNeighborMap = new HashMap<>(); - private Map> mPatchGroupMap = new HashMap<>(); private String mLastCommandEventID; private String mLastPageEventID; @@ -188,6 +185,7 @@ public class P25DecoderState extends DecoderState private P25CallEvent mCurrentCallEvent; private List mCallDetectTalkgroups = new ArrayList<>(); private Map mChannelCallMap = new HashMap<>(); + private PatchGroupManager mPatchGroupManager; public P25DecoderState(AliasList aliasList, ChannelType channelType, @@ -199,6 +197,7 @@ public class P25DecoderState extends DecoderState mModulation = modulation; mIgnoreDataCalls = ignoreDataCalls; + mPatchGroupManager = new PatchGroupManager(aliasList, getCallEventBroadcaster()); mSiteAttributeMonitor = new AliasedStringAttributeMonitor(Attribute.NETWORK_ID_2, getAttributeChangeRequestListener(), getAliasList(), AliasIDType.SITE); mFromTalkgroupMonitor = new AliasedStringAttributeMonitor(Attribute.PRIMARY_ADDRESS_FROM, @@ -2719,6 +2718,9 @@ public class P25DecoderState extends DecoderState switch(((MotorolaTSBKMessage)tsbk).getMotorolaOpcode()) { case PATCH_GROUP_CHANNEL_GRANT: + //Cleanup patch groups - auto-expire any patch groups before we allocate a channel + mPatchGroupManager.cleanupPatchGroups(); + PatchGroupVoiceChannelGrant pgvcg = (PatchGroupVoiceChannelGrant)tsbk; channel = pgvcg.getChannel(); @@ -2754,6 +2756,9 @@ public class P25DecoderState extends DecoderState mChannelCallMap.get(channel))); break; case PATCH_GROUP_CHANNEL_GRANT_UPDATE: + //Cleanup patch groups - auto-expire any patch groups before we allocate a channel + mPatchGroupManager.cleanupPatchGroups(); + PatchGroupVoiceChannelGrantUpdate gvcgu = (PatchGroupVoiceChannelGrantUpdate)tsbk; channel = gvcgu.getChannel1(); @@ -2784,35 +2789,11 @@ public class P25DecoderState extends DecoderState break; case PATCH_GROUP_ADD: PatchGroupAdd pga = (PatchGroupAdd)tsbk; - - if(addPatchGroup(pga.getPatchGroupAddress(), pga.getPatchedTalkgroups())) - { - StringBuilder sb = new StringBuilder(); - - sb.append("PATCH GROUP:").append(pga.getPatchGroupAddress()); - sb.append(" ").append(pga.getPatchedTalkgroups()); - - broadcast(new P25CallEvent.Builder(CallEventType.PATCH_GROUP_ADD) - .aliasList(getAliasList()) - .details(sb.toString()) - .build()); - } + mPatchGroupManager.updatePatchGroup(pga.getPatchGroupAddress(), pga.getPatchedTalkgroups()); break; case PATCH_GROUP_DELETE: PatchGroupDelete pgd = (PatchGroupDelete)tsbk; - - if(removePatchGroup(pgd.getPatchGroupAddress(), pgd.getPatchedTalkgroups())) - { - StringBuilder sbpgd = new StringBuilder(); - - sbpgd.append("PATCH GROUP:").append(pgd.getPatchGroupAddress()); - sbpgd.append(" ").append(pgd.getPatchedTalkgroups()); - - broadcast(new P25CallEvent.Builder(CallEventType.PATCH_GROUP_DELETE) - .aliasList(getAliasList()) - .details(sbpgd.toString()) - .build()); - } + mPatchGroupManager.removePatchGroup(pgd.getPatchGroupAddress()); break; case CCH_PLANNED_SHUTDOWN: if(!mControlChannelShutdownLogged) @@ -2827,174 +2808,14 @@ public class P25DecoderState extends DecoderState } } - /** - * Creates a new patch group or updates an existing patch group with the patched talkgroup entries. - * - * @param patchGroupID for the patch group - * @param talkgroupsToAdd to include in the patch group - * @return true if the patch group or the patch group map was updated - */ - private boolean addPatchGroup(String patchGroupID, List talkgroupsToAdd) - { - boolean updated = false; - - if(patchGroupID != null && !talkgroupsToAdd.isEmpty()) - { - List patchedTalkgroups = mPatchGroupMap.get(patchGroupID); - - if(patchedTalkgroups == null) - { - patchedTalkgroups = new ArrayList<>(); - } - - for(String talkgroupToAdd : talkgroupsToAdd) - { - //Exclude the patch group ID if it is included in the patched talkgroup list so that we don't have - //an infinite loop situation - if(talkgroupToAdd.equals(patchGroupID)) - { - continue; - } - - if(!patchedTalkgroups.contains(talkgroupToAdd)) - { - patchedTalkgroups.add(talkgroupToAdd); - updated = true; - } - } - - if(updated) - { - mPatchGroupMap.put(patchGroupID, patchedTalkgroups); - updatePatchAlias(patchGroupID, patchedTalkgroups); - } - } - - return updated; - } - - /** - * Removes the patched talkgroups from the patch group and removes the patch group from the patch group map if - * there are no more patched talkgroups in the patch group. - * - * @param patchGroupID for the patch group - * @param talkgroupsToRemove to remove from the patch group - * @return true if the patch group or the patch group map was updated - */ - private boolean removePatchGroup(String patchGroupID, List talkgroupsToRemove) - { - boolean updated = false; - - if(patchGroupID != null && !talkgroupsToRemove.isEmpty()) - { - List patchedTalkgroups = mPatchGroupMap.get(patchGroupID); - - if(patchedTalkgroups != null) - { - for(String talkgroupToRemove : talkgroupsToRemove) - { - if(patchedTalkgroups.contains(talkgroupToRemove)) - { - patchedTalkgroups.remove(talkgroupToRemove); - updated = true; - } - } - - if(patchedTalkgroups.isEmpty()) - { - mPatchGroupMap.remove(patchGroupID); - removePatchGroupAlias(patchGroupID); - updated = true; - } - else - { - updatePatchAlias(patchGroupID, patchedTalkgroups); - } - } - } - - return updated; - } - - /** - * Adds/updates a patch group alias to the alias list containing aliases for each of the patched talkgroups. If the - * patch group alias already exists, any patched talkgroups will be removed from the existing patch group alias and - * replaced with the aliases corresponding to the patched talkgroup aliases. - * - * @param patchGroupID for the patch group - * @param patchedTalkgroups containing the talkgroup IDs for each of the patched talkgroups - */ - private void updatePatchAlias(String patchGroupID, List patchedTalkgroups) - { - PatchGroupAlias patchGroupAlias = null; - - //Check for an existing alias for the patch talkgroup - do not include wildcard aliases - Alias existingAlias = getAliasList().getTalkgroupAlias(patchGroupID, false); - - if(existingAlias instanceof PatchGroupAlias) - { - patchGroupAlias = (PatchGroupAlias)existingAlias; - } - else - { - patchGroupAlias = new PatchGroupAlias(); - - patchGroupAlias.addAliasID(new TalkgroupID(patchGroupID)); - - if(existingAlias != null) - { - getAliasList().removeAlias(existingAlias); - - patchGroupAlias.setPatchGroupAlias(existingAlias); - } - - getAliasList().addAlias(patchGroupAlias); - } - - if(patchGroupAlias != null) - { - patchGroupAlias.setPatchedTalkgroupIDs(patchedTalkgroups); - - patchGroupAlias.clearPatchedAliases(); - - for(String patchedTalkgroup : patchedTalkgroups) - { - Alias patchedAlias = getAliasList().getTalkgroupAlias(patchedTalkgroup); - - if(patchedAlias != null) - { - patchGroupAlias.addPatchedAlias(patchedAlias); - } - } - } - } - - /** - * Removes the temporary patch group alias from the alias list - */ - private void removePatchGroupAlias(String patchGroupID) - { - Alias alias = getAliasList().getTalkgroupAlias(patchGroupID); - - if(alias instanceof PatchGroupAlias) - { - PatchGroupAlias patchGroupAlias = (PatchGroupAlias)alias; - - getAliasList().removeAlias(patchGroupAlias); - - //Replace our temporary patch group alias with the original alias for the patch group ID - if(patchGroupAlias.hasPatchGroupAlias()) - { - getAliasList().addAlias(patchGroupAlias.getPatchGroupAlias()); - } - } - } - /** * Process a traffic channel allocation message */ private void processTSBKChannelGrant(TSBKMessage message) { + //Cleanup patch groups - auto-expire any patch groups before we allocate a channel + mPatchGroupManager.cleanupPatchGroups(); + String channel = null; String from = null; String to = null; diff --git a/src/module/decode/p25/PatchGroup.java b/src/module/decode/p25/PatchGroup.java new file mode 100644 index 00000000..525602ad --- /dev/null +++ b/src/module/decode/p25/PatchGroup.java @@ -0,0 +1,102 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2017 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.decode.p25; + +import java.util.ArrayList; +import java.util.List; + +public class PatchGroup +{ + private long mExpireThreshold = 5000; //5 seconds + private long mLastUpdated; + private String mPatchGroupID; + private List mPatchedGroups = new ArrayList<>(); + + /** + * Constructs a patch group with the specified patch group ID + * @param patchGroupID that identifies this patch group + */ + public PatchGroup(String patchGroupID) + { + mPatchGroupID = patchGroupID; + mLastUpdated = System.currentTimeMillis(); + } + + /** + * Patch group ID for this patch group + */ + public String getPatchGroupID() + { + return mPatchGroupID; + } + + /** + * Adds the patched group to this patch group. Duplicate patched groups are ignored. + * + * Note: the patch group ID will not be added as a patched group, to avoid circular dependencies. + * + * @return true if the patched group was added to this patch group + */ + public boolean addPatchedGroup(String patchedGroup) + { + if(patchedGroup != null && !patchedGroup.equals(mPatchGroupID) && !mPatchedGroups.contains(patchedGroup)) + { + mPatchedGroups.add(patchedGroup); + return true; + } + + return false; + } + + /** + * Patched talkgroups that are part of this patch group. + */ + public List getPatchedGroups() + { + return mPatchedGroups; + } + + /** + * Updates the timestamp for this patch group. + */ + public void updateTimestamp() + { + mLastUpdated = System.currentTimeMillis(); + } + + /** + * Indicates if this patch group is expired, meaning that it hasn't been updated in the past (expire + * threshold) seconds. + */ + public boolean isExpired() + { + return System.currentTimeMillis() - mLastUpdated > mExpireThreshold; + } + + /** + * Sets the patch group expiration threshold (default = 5000 ms). A patch group is considered expired if + * no new updates have been received within the past (expire threshold) milliseconds + * + * @param expireThreshold in milliseconds + */ + public void setExpireThreshold(long expireThreshold) + { + mExpireThreshold = expireThreshold; + } +} diff --git a/src/module/decode/p25/PatchGroupManager.java b/src/module/decode/p25/PatchGroupManager.java new file mode 100644 index 00000000..8642923e --- /dev/null +++ b/src/module/decode/p25/PatchGroupManager.java @@ -0,0 +1,245 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2017 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.decode.p25; + +import alias.Alias; +import alias.AliasList; +import alias.PatchGroupAlias; +import alias.id.talkgroup.TalkgroupID; +import module.decode.event.CallEvent; +import sample.Broadcaster; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +public class PatchGroupManager +{ + private AliasList mAliasList; + private List mPatchGroups = new ArrayList<>(); + private Broadcaster mCallEventBroadcaster; + + /** + * Manages active P25 patch groups and auto-expires patch groups if not updated every 5 seconds. + * + * @param aliasList to keep updated with patch group changes + * @param broadcaster for rebroadcasting patch group related call events + */ + public PatchGroupManager(AliasList aliasList, Broadcaster broadcaster) + { + mAliasList = aliasList; + mCallEventBroadcaster = broadcaster; + } + + /** + * Removes the patch group. + * + * @param patchGroupID of the patch group to remove + */ + public void removePatchGroup(String patchGroupID) + { + removePatchGroup(patchGroupID, false); + } + + /** + * Removes the patch group. + * + * @param patchGroupID of the patch group to remove + * @param expired true if the patch group is being removed because it expired, false if we receive a + * message commanding commanding a patch group delete. + */ + private void removePatchGroup(String patchGroupID, boolean expired) + { + PatchGroup patchGroup = getPatchGroup(patchGroupID); + + if(patchGroup != null) + { + //Cleanup the alias list + if(mAliasList != null) + { + Alias alias = mAliasList.getTalkgroupAlias(patchGroupID); + + if(alias instanceof PatchGroupAlias) + { + PatchGroupAlias patchGroupAlias = (PatchGroupAlias)alias; + + mAliasList.removeAlias(patchGroupAlias); + + //Replace our temporary patch group alias with the original alias for the patch group ID + if(patchGroupAlias.hasPatchGroupAlias()) + { + mAliasList.addAlias(patchGroupAlias.getPatchGroupAlias()); + } + } + } + + mPatchGroups.remove(patchGroup); + + if(mCallEventBroadcaster != null) + { + StringBuilder sb = new StringBuilder(); + + sb.append((expired ? "EXPIRED " : "DELETED ")); + sb.append("PATCH GROUP:").append(patchGroup.getPatchGroupID()); + sb.append(" ").append(patchGroup.getPatchedGroups()); + + + mCallEventBroadcaster.broadcast(new P25CallEvent.Builder(CallEvent.CallEventType.PATCH_GROUP_DELETE) + .to(patchGroup.getPatchGroupID()) + .details(sb.toString()) + .build()); + } + } + } + + /** + * Adds/updates a patch group alias to the alias list containing aliases for each of the patched talkgroups. If the + * patch group alias already exists, any patched talkgroups will be removed from the existing patch group alias and + * replaced with the aliases corresponding to the patched talkgroup aliases. + * + * @param patchGroupID for the patch group + * @param patchedTalkgroups containing the talkgroup IDs for each of the patched talkgroups + */ + public void updatePatchGroup(String patchGroupID, List patchedTalkgroups) + { + boolean updated = false; + + PatchGroup patchGroup = getPatchGroup(patchGroupID); + + if(patchGroup == null) + { + //Remove any expired patch groups before we add a new one + cleanupPatchGroups(); + patchGroup = new PatchGroup(patchGroupID); + mPatchGroups.add(patchGroup); + updated = true; + } + else + { + patchGroup.updateTimestamp(); + } + + for(String patchedGroup : patchedTalkgroups) + { + if(patchGroup.addPatchedGroup(patchedGroup)) + { + updated = true; + } + } + + //If we make any changes to the patch group, update the patch group alias + if(updated) + { + PatchGroupAlias patchGroupAlias = null; + + //Check for an existing alias for the patch talkgroup - do not include wildcard aliases + if(mAliasList != null) + { + Alias existingAlias = mAliasList.getTalkgroupAlias(patchGroupID, false); + + if(existingAlias instanceof PatchGroupAlias) + { + patchGroupAlias = (PatchGroupAlias)existingAlias; + } + else + { + patchGroupAlias = new PatchGroupAlias(); + + patchGroupAlias.addAliasID(new TalkgroupID(patchGroupID)); + + if(existingAlias != null) + { + mAliasList.removeAlias(existingAlias); + + patchGroupAlias.setPatchGroupAlias(existingAlias); + } + + mAliasList.addAlias(patchGroupAlias); + } + + if(patchGroupAlias != null) + { + patchGroupAlias.setPatchedTalkgroupIDs(patchGroup.getPatchedGroups()); + + patchGroupAlias.clearPatchedAliases(); + + for(String patchedTalkgroup : patchGroup.getPatchedGroups()) + { + Alias patchedAlias = mAliasList.getTalkgroupAlias(patchedTalkgroup); + + if(patchedAlias != null) + { + patchGroupAlias.addPatchedAlias(patchedAlias); + } + } + } + } + + if(mCallEventBroadcaster != null) + { + StringBuilder sb = new StringBuilder(); + + sb.append("PATCH GROUP:").append(patchGroup.getPatchGroupID()); + sb.append(" ").append(patchGroup.getPatchedGroups()); + + mCallEventBroadcaster.broadcast(new P25CallEvent.Builder(CallEvent.CallEventType.PATCH_GROUP_ADD) + .aliasList(mAliasList) + .details(sb.toString()) + .build()); + } + } + } + + /** + * Returns the patch group matching the patch group ID or null + */ + private PatchGroup getPatchGroup(String patchGroupID) + { + for(PatchGroup patchGroup : mPatchGroups) + { + if(patchGroup.getPatchGroupID().equals(patchGroupID)) + { + return patchGroup; + } + } + + return null; + } + + /** + * Checks the last updated timestamp on all patch groups and removes any patch groups that have not been + * updated in the past 5 seconds + */ + public void cleanupPatchGroups() + { + Iterator it = mPatchGroups.iterator(); + + while(it.hasNext()) + { + PatchGroup patchGroup = it.next(); + + if(patchGroup.isExpired()) + { + it.remove(); + + removePatchGroup(patchGroup.getPatchGroupID(), true); + } + } + } +} diff --git a/src/sample/OverflowableTransferQueue.java b/src/sample/OverflowableTransferQueue.java index 24f4b949..f9c9c190 100644 --- a/src/sample/OverflowableTransferQueue.java +++ b/src/sample/OverflowableTransferQueue.java @@ -18,20 +18,28 @@ ******************************************************************************/ package sample; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sample.real.IOverflowListener; +import source.tuner.TunerChannelSource; + import java.util.Collection; import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; public class OverflowableTransferQueue { + private final static Logger mLog = LoggerFactory.getLogger(OverflowableTransferQueue.class); + public enum State {NORMAL, OVERFLOW}; - private Listener mStateListener; + private IOverflowListener mOverflowListener; private LinkedTransferQueue mQueue = new LinkedTransferQueue(); private AtomicInteger mCounter = new AtomicInteger(); + private AtomicBoolean mOverflow = new AtomicBoolean(); private int mMaximumSize; private int mResetThreshold; - private boolean mOverflow = false; /** * Concurrent transfer queue that couples a higher-throughput linked transfer queue with an atomic integer for @@ -53,7 +61,7 @@ public class OverflowableTransferQueue */ public void offer(E e) { - if(!mOverflow) + if(!mOverflow.get()) { mQueue.offer(e); @@ -73,46 +81,36 @@ public class OverflowableTransferQueue { int drainCount = mQueue.drainTo(collection, maxElements); - if(drainCount > 0) - { - int size = mCounter.addAndGet(-drainCount); + int size = mCounter.addAndGet(-drainCount); - if(mOverflow && size <= mResetThreshold) - { - setOverflow(false); - } + if(mOverflow.get() && size <= mResetThreshold) + { + setOverflow(false); } return drainCount; } /** - * Sets a listener to receive state change events + * Sets a listener to receive overflow state change events */ - public void setStateListener(Listener listener) + public void setOverflowListener(IOverflowListener listener) { - mStateListener = listener; + mOverflowListener = listener; } /** - * Current state of the queue, normal or overflow - */ - public State getState() - { - return mOverflow ? State.OVERFLOW : State.NORMAL; - } - - /** - * Toggles the overflow state + * Toggles the overflow state and broadcast state change to listener */ private void setOverflow(boolean overflow) { - mOverflow = overflow; - - if(mStateListener != null) + if(mOverflow.compareAndSet(!overflow, overflow)) { - mStateListener.receive(getState()); + if(mOverflowListener != null) + { + mOverflowListener.sourceOverflow(overflow); + } } } @@ -125,7 +123,11 @@ public class OverflowableTransferQueue { mQueue.clear(); mCounter.set(0); - mOverflow = false; + mOverflow.set(false); + if(mOverflowListener != null) + { + mOverflowListener.sourceOverflow(false); + } } } } diff --git a/src/sample/real/IOverflowListener.java b/src/sample/real/IOverflowListener.java new file mode 100644 index 00000000..61202ea3 --- /dev/null +++ b/src/sample/real/IOverflowListener.java @@ -0,0 +1,28 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2017 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 sample.real; + +public interface IOverflowListener +{ + /** + * Indicates that the source is in overflow state (true) or out of overflow state (false) + * @param overflow true to indicate an overflow state + */ + void sourceOverflow(boolean overflow); +} diff --git a/src/source/Source.java b/src/source/Source.java index 217def00..2de310b8 100644 --- a/src/source/Source.java +++ b/src/source/Source.java @@ -19,6 +19,7 @@ package source; import module.Module; import sample.SampleType; +import sample.real.IOverflowListener; /** * Abstract class to define the minimum functionality of a sample data provider. @@ -26,6 +27,7 @@ import sample.SampleType; public abstract class Source extends Module { protected SampleType mSampleType; + protected IOverflowListener mOverflowListener; public Source( SampleType sampleType ) { @@ -47,4 +49,25 @@ public abstract class Source extends Module { mSampleType = sampleType; } + + /** + * Registers the listener to receive overflow state changes. Use null argument to clear the listener + */ + public void setOverflowListener(IOverflowListener listener) + { + mOverflowListener = listener; + } + + /** + * Broadcasts an overflow state + * + * @param overflow true if overlow, false if normal + */ + protected void broadcastOverflowState(boolean overflow) + { + if(mOverflowListener != null) + { + mOverflowListener.sourceOverflow(overflow); + } + } } diff --git a/src/source/tuner/TunerChannelSource.java b/src/source/tuner/TunerChannelSource.java index c50fd267..fa0cef46 100644 --- a/src/source/tuner/TunerChannelSource.java +++ b/src/source/tuner/TunerChannelSource.java @@ -29,6 +29,7 @@ import sample.Listener; import sample.OverflowableTransferQueue; import sample.complex.Complex; import sample.complex.ComplexBuffer; +import sample.real.IOverflowListener; import source.ComplexSource; import source.SourceException; import source.tuner.frequency.FrequencyChangeEvent; @@ -51,7 +52,7 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang private static final int BUFFER_MAX_CAPACITY = 300; //Threshold for resetting buffer overflow condition - private static final int BUFFER_OVERFLOW_RESET_THRESHOLD = 200; + private static final int BUFFER_OVERFLOW_RESET_THRESHOLD = 100; private static int CHANNEL_RATE = 48000; private static int CHANNEL_PASS_FREQUENCY = 12000; @@ -101,21 +102,6 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang mTunerFrequency = mTuner.getTunerController().getFrequency(); mBuffer = new OverflowableTransferQueue<>(BUFFER_MAX_CAPACITY, BUFFER_OVERFLOW_RESET_THRESHOLD); - mBuffer.setStateListener(new Listener() - { - @Override - public void receive(OverflowableTransferQueue.State state) - { - if(state == OverflowableTransferQueue.State.NORMAL) - { - mLog.debug("Channel [" + mTunerFrequency + "] buffer overflow - temporary pause until processing catches up"); - } - else - { - mLog.debug("Channel [" + mTunerFrequency + "] buffer overflow cleared - resuming normal processing"); - } - } - }); /* Setup the frequency translator to the current source frequency */ long frequencyOffset = mTunerFrequency - mTunerChannel.getFrequency(); @@ -126,6 +112,14 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang frequencyChanged(new FrequencyChangeEvent(Event.NOTIFICATION_SAMPLE_RATE_CHANGE, mTuner.getTunerController().getSampleRate())); } + /** + * Overrides the default source overflow listener management to delegate responsibility to the overflow buffer + */ + @Override + public void setOverflowListener(IOverflowListener listener) + { + mBuffer.setOverflowListener(listener); + } public void start(ScheduledExecutorService executor) { @@ -175,7 +169,7 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang } else { - mLog.warn("Attempt to stop() and already stopped tuner channel source was ignored"); + mLog.warn("Attempt to stop() an already stopped tuner channel source was ignored"); } } @@ -379,6 +373,10 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang mLog.error("Error encountered during decimation process", e); } } + catch(Throwable throwable) + { + mLog.error("Code error encountered during decimation process - channel thread will probably die", throwable); + } /* Check to see if we've been shutdown */ if(!mProcessing) diff --git a/src/source/tuner/usb/USBTransferProcessor.java b/src/source/tuner/usb/USBTransferProcessor.java index 053250f3..aeb56bb9 100644 --- a/src/source/tuner/usb/USBTransferProcessor.java +++ b/src/source/tuner/usb/USBTransferProcessor.java @@ -30,6 +30,7 @@ import sample.Listener; import sample.OverflowableTransferQueue; import sample.adapter.ISampleAdapter; import sample.complex.ComplexBuffer; +import sample.real.IOverflowListener; import source.tuner.TunerManager; import util.ThreadPool; @@ -56,7 +57,7 @@ public class USBTransferProcessor implements TransferCallback private static final int FILLED_BUFFER_MAX_CAPACITY = 300; //Threshold for resetting buffer overflow condition - private static final int FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD = 200; + private static final int FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD = 100; private String mDeviceName; @@ -101,12 +102,12 @@ public class USBTransferProcessor implements TransferCallback mBufferSize = bufferSize; mFilledBuffers = new OverflowableTransferQueue<>(FILLED_BUFFER_MAX_CAPACITY, FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD); - mFilledBuffers.setStateListener(new Listener() + mFilledBuffers.setOverflowListener(new IOverflowListener() { @Override - public void receive(OverflowableTransferQueue.State state) + public void sourceOverflow(boolean overflow) { - if(state == OverflowableTransferQueue.State.NORMAL) + if(overflow) { mLog.debug(mDeviceName + " - buffer overflow - temporary pause until processing catches up"); }