mirror of
https://github.com/DSheirer/sdrtrunk.git
synced 2026-11-09 23:27:24 -05:00
905 lines
34 KiB
Java
905 lines
34 KiB
Java
/*
|
|
* *****************************************************************************
|
|
* Copyright (C) 2014-2024 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 <http://www.gnu.org/licenses/>
|
|
* ****************************************************************************
|
|
*/
|
|
package io.github.dsheirer.controller.channel;
|
|
|
|
import com.google.common.eventbus.Subscribe;
|
|
import io.github.dsheirer.alias.AliasModel;
|
|
import io.github.dsheirer.audio.AudioSegment;
|
|
import io.github.dsheirer.channel.metadata.ChannelAndMetadata;
|
|
import io.github.dsheirer.channel.metadata.ChannelMetadata;
|
|
import io.github.dsheirer.channel.metadata.ChannelMetadataModel;
|
|
import io.github.dsheirer.channel.state.AbstractChannelState;
|
|
import io.github.dsheirer.controller.channel.event.ChannelStartProcessingRequest;
|
|
import io.github.dsheirer.controller.channel.event.ChannelStopProcessingRequest;
|
|
import io.github.dsheirer.controller.channel.event.PreloadDataContent;
|
|
import io.github.dsheirer.controller.channel.map.ChannelMapModel;
|
|
import io.github.dsheirer.identifier.Form;
|
|
import io.github.dsheirer.identifier.Identifier;
|
|
import io.github.dsheirer.identifier.IdentifierClass;
|
|
import io.github.dsheirer.identifier.IdentifierUpdateNotification;
|
|
import io.github.dsheirer.identifier.decoder.DecoderLogicalChannelNameIdentifier;
|
|
import io.github.dsheirer.module.Module;
|
|
import io.github.dsheirer.module.ProcessingChain;
|
|
import io.github.dsheirer.module.decode.DecoderFactory;
|
|
import io.github.dsheirer.module.decode.event.IDecodeEvent;
|
|
import io.github.dsheirer.module.log.EventLogManager;
|
|
import io.github.dsheirer.preference.UserPreferences;
|
|
import io.github.dsheirer.record.RecorderFactory;
|
|
import io.github.dsheirer.sample.Broadcaster;
|
|
import io.github.dsheirer.sample.Listener;
|
|
import io.github.dsheirer.source.Source;
|
|
import io.github.dsheirer.source.SourceEvent;
|
|
import io.github.dsheirer.source.SourceException;
|
|
import io.github.dsheirer.source.config.SourceConfigTuner;
|
|
import io.github.dsheirer.source.config.SourceConfigTunerMultipleFrequency;
|
|
import io.github.dsheirer.source.tuner.channel.TunerChannelSource;
|
|
import io.github.dsheirer.source.tuner.manager.TunerManager;
|
|
import io.github.dsheirer.util.ThreadPool;
|
|
import java.awt.GraphicsEnvironment;
|
|
import java.util.ArrayList;
|
|
import java.util.HashMap;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
import java.util.concurrent.CopyOnWriteArrayList;
|
|
import java.util.concurrent.ScheduledFuture;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.concurrent.locks.Lock;
|
|
import java.util.concurrent.locks.ReentrantLock;
|
|
import javafx.application.Platform;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
/**
|
|
* Channel processing manager handles all starting and stopping of channel decoding. A processing chain is created
|
|
* for each channel that is enabled. The processing chain contains all of the components needed to decode a specific
|
|
* channel and protocol along with all logging and baseband or bitstream recording. Audio recording is handled outside
|
|
* of this class by the RecorderManager.
|
|
*/
|
|
public class ChannelProcessingManager implements Listener<ChannelEvent>
|
|
{
|
|
private static final String DIVIDER = "-------------------------------------------------------------------------\n";
|
|
private final static Logger mLog = LoggerFactory.getLogger(ChannelProcessingManager.class);
|
|
private static final String TUNER_UNAVAILABLE_DESCRIPTION = "TUNER UNAVAILABLE";
|
|
private Map<Channel,ProcessingChain> mProcessingChainsMap = new ConcurrentHashMap<>();
|
|
private Lock mLock = new ReentrantLock();
|
|
|
|
private ChannelSourceEventErrorListener mSourceErrorListener = new ChannelSourceEventErrorListener();
|
|
private List<Listener<AudioSegment>> mAudioSegmentListeners = new CopyOnWriteArrayList<>();
|
|
private List<Listener<IDecodeEvent>> mDecodeEventListeners = new CopyOnWriteArrayList<>();
|
|
private Broadcaster<ChannelEvent> mChannelEventBroadcaster = new Broadcaster();
|
|
|
|
private ChannelMapModel mChannelMapModel;
|
|
private ChannelMetadataModel mChannelMetadataModel;
|
|
private EventLogManager mEventLogManager;
|
|
private TunerManager mTunerManager;
|
|
private AliasModel mAliasModel;
|
|
private UserPreferences mUserPreferences;
|
|
private List<Long> mLoggedFrequencies = new ArrayList<>();
|
|
private List<ScheduledFuture<?>> mDelayedChannelStartTasks = new ArrayList<>();
|
|
|
|
/**
|
|
* Constructs the channel processing manager
|
|
*
|
|
* @param channelMapModel containing channel maps defined by the user
|
|
* @param eventLogManager for adding event loggers to channels
|
|
* @param tunerManager for obtaining a tuner channel source for the channel
|
|
* @param aliasModel for aliasing of identifiers produced by the channel
|
|
* @param userPreferences for user defined behavior and settings
|
|
*/
|
|
public ChannelProcessingManager(ChannelMapModel channelMapModel, EventLogManager eventLogManager,
|
|
TunerManager tunerManager, AliasModel aliasModel, UserPreferences userPreferences)
|
|
{
|
|
mChannelMapModel = channelMapModel;
|
|
mEventLogManager = eventLogManager;
|
|
mTunerManager = tunerManager;
|
|
mAliasModel = aliasModel;
|
|
mUserPreferences = userPreferences;
|
|
mChannelMetadataModel = new ChannelMetadataModel();
|
|
}
|
|
|
|
/**
|
|
* Channel metadata model containing metadata for each channel or channel time-slice that is currently processing.
|
|
*/
|
|
public ChannelMetadataModel getChannelMetadataModel()
|
|
{
|
|
return mChannelMetadataModel;
|
|
}
|
|
|
|
/**
|
|
* Indicates if a processing chain is constructed for the channel and that
|
|
* the processing chain is currently processing.
|
|
*/
|
|
private boolean isProcessing(Channel channel)
|
|
{
|
|
boolean isProcessing = false;
|
|
|
|
mLock.lock();
|
|
|
|
try
|
|
{
|
|
isProcessing = mProcessingChainsMap.containsKey(channel) && mProcessingChainsMap.get(channel).isProcessing();
|
|
}
|
|
finally
|
|
{
|
|
mLock.unlock();
|
|
}
|
|
|
|
return isProcessing;
|
|
}
|
|
|
|
/**
|
|
* Indicates if any channels are currently processing.
|
|
* @return true if channels are processing.
|
|
*/
|
|
public boolean isProcessing()
|
|
{
|
|
return !mProcessingChainsMap.isEmpty();
|
|
}
|
|
|
|
/**
|
|
* Returns the current processing chain associated with the channel, or
|
|
* null if a processing chain is not currently setup for the channel
|
|
*/
|
|
public ProcessingChain getProcessingChain(Channel channel)
|
|
{
|
|
return mProcessingChainsMap.get(channel);
|
|
}
|
|
|
|
/**
|
|
* Returns the channel associated with the processing chain
|
|
*
|
|
* @param processingChain
|
|
* @return channel associated with the processing chain or null
|
|
*/
|
|
public Channel getChannel(ProcessingChain processingChain)
|
|
{
|
|
Channel channel = null;
|
|
|
|
if(processingChain != null)
|
|
{
|
|
mLock.lock();
|
|
|
|
try
|
|
{
|
|
for(Map.Entry<Channel,ProcessingChain> entry : mProcessingChainsMap.entrySet())
|
|
{
|
|
if(entry.getValue() == processingChain)
|
|
{
|
|
channel = entry.getKey();
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
mLock.unlock();
|
|
}
|
|
}
|
|
|
|
return channel;
|
|
}
|
|
|
|
/**
|
|
* Retrieves the channel associated with the processing chain that is consuming from the tuner channel source
|
|
* @param tunerChannelSource to find the channel
|
|
* @return channel
|
|
*/
|
|
private Channel getChannel(TunerChannelSource tunerChannelSource)
|
|
{
|
|
Channel channel = null;
|
|
|
|
mLock.lock();
|
|
|
|
try
|
|
{
|
|
for(Map.Entry<Channel,ProcessingChain> entry : mProcessingChainsMap.entrySet())
|
|
{
|
|
if(entry.getValue().hasSource(tunerChannelSource))
|
|
{
|
|
channel = entry.getKey();
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
mLock.unlock();
|
|
}
|
|
|
|
return channel;
|
|
}
|
|
|
|
/**
|
|
* Primary method for receiving requests to start and stop a channel
|
|
*
|
|
* @param event that requests either enable/start or disable/stop a channel.
|
|
*/
|
|
@Override
|
|
public void receive(ChannelEvent event)
|
|
{
|
|
Channel channel = event.getChannel();
|
|
|
|
switch(event.getEvent())
|
|
{
|
|
case REQUEST_ENABLE:
|
|
if(!isProcessing(channel))
|
|
{
|
|
try
|
|
{
|
|
startProcessing(new ChannelStartProcessingRequest(event.getChannel()));
|
|
}
|
|
catch(ChannelException ce)
|
|
{
|
|
if(channel.getSourceConfiguration() instanceof SourceConfigTuner)
|
|
{
|
|
long frequency = ((SourceConfigTuner)channel.getSourceConfiguration()).getFrequency();
|
|
|
|
if(!mLoggedFrequencies.contains(frequency))
|
|
{
|
|
mLoggedFrequencies.add(frequency);
|
|
mLog.error("Error starting requested channel [" + channel.getName() + ":" + frequency +
|
|
"] - " + ce.getMessage());
|
|
}
|
|
}
|
|
else if(channel.getSourceConfiguration() instanceof SourceConfigTunerMultipleFrequency)
|
|
{
|
|
List<Long> frequencies = ((SourceConfigTunerMultipleFrequency)channel
|
|
.getSourceConfiguration()).getFrequencies();
|
|
|
|
if(frequencies.size() > 0 && !mLoggedFrequencies.contains(frequencies.get(0)))
|
|
{
|
|
mLoggedFrequencies.add(frequencies.get(0));
|
|
mLog.error("Error starting requested channel [" + channel.getName() + ":" + frequencies +
|
|
"] - " + ce.getMessage());
|
|
}
|
|
}
|
|
else
|
|
{
|
|
mLog.error("Error starting requested channel [" + channel.getName() + "] - " + ce.getMessage());
|
|
}
|
|
}
|
|
}
|
|
break;
|
|
case REQUEST_DISABLE:
|
|
case NOTIFICATION_DELETE:
|
|
if(channel != null && channel.isProcessing())
|
|
{
|
|
try
|
|
{
|
|
stopProcessing(channel);
|
|
}
|
|
catch(ChannelException ce)
|
|
{
|
|
mLog.error("Error stopping channel [" + channel.getName() + "] - " + ce.getMessage());
|
|
}
|
|
}
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Starts the specified channel.
|
|
* @param channel to start
|
|
* @throws ChannelException if the channel can't be started
|
|
*/
|
|
public void start(Channel channel) throws ChannelException
|
|
{
|
|
startProcessing(new ChannelStartProcessingRequest(channel));
|
|
}
|
|
|
|
/**
|
|
* Request to start processing a channel received over the Guava event bus.
|
|
*
|
|
* Note: since this is received over the event bus, we handle any channel exceptions inside this method.
|
|
*/
|
|
@Subscribe
|
|
public void startChannelRequest(ChannelStartProcessingRequest request)
|
|
{
|
|
if(!isProcessing(request.getChannel()))
|
|
{
|
|
try
|
|
{
|
|
startProcessing(request);
|
|
}
|
|
catch(ChannelException ce)
|
|
{
|
|
if(request.isPersistentAttempt())
|
|
{
|
|
ScheduledFuture<?> future = ThreadPool.SCHEDULED
|
|
.schedule(new DelayedChannelStartTask(request), 500, TimeUnit.MILLISECONDS);
|
|
mDelayedChannelStartTasks.add(future);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Request to stop a channel that is currently processing
|
|
* @param request with the tuner channel source feeding the channel to be stopped.
|
|
*/
|
|
@Subscribe
|
|
public void stopChannelRequest(ChannelStopProcessingRequest request)
|
|
{
|
|
Channel channel = getChannel(request.getTunerChannelSource());
|
|
|
|
if(channel != null)
|
|
{
|
|
try
|
|
{
|
|
stop(channel);
|
|
}
|
|
catch(ChannelException ce)
|
|
{
|
|
mLog.error("Error stopping channel [" + channel + "]", ce);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Stops the specified channel
|
|
* @param channel to stop
|
|
*/
|
|
public void stop(Channel channel) throws ChannelException
|
|
{
|
|
stopProcessing(channel);
|
|
}
|
|
|
|
/**
|
|
* Starts a channel processing
|
|
* @param request containing channel and other details
|
|
* @throws ChannelException if a source is not available for the channel
|
|
*/
|
|
private synchronized void startProcessing(ChannelStartProcessingRequest request) throws ChannelException
|
|
{
|
|
Channel channel = request.getChannel();
|
|
|
|
if(isProcessing(channel))
|
|
{
|
|
return;
|
|
}
|
|
|
|
//Ensure that we can get a source before we construct a new processing chain
|
|
Source source = null;
|
|
|
|
try
|
|
{
|
|
String threadName = "sdrtrunk channel [" + channel.getChannelID() + "/" +
|
|
channel.getDecodeConfiguration().getDecoderType().getShortDisplayString() + "]";
|
|
source = mTunerManager.getSource(channel.getSourceConfiguration(),
|
|
channel.getDecodeConfiguration().getChannelSpecification(), threadName);
|
|
}
|
|
catch(SourceException se)
|
|
{
|
|
mLog.debug("Error obtaining source for channel [" + channel.getName() + "]", se);
|
|
}
|
|
|
|
if(source == null)
|
|
{
|
|
//This has to be done on the FX event thread when the playlist editor is constructed
|
|
if(!GraphicsEnvironment.isHeadless())
|
|
{
|
|
Platform.runLater(() -> channel.setProcessing(false));
|
|
}
|
|
|
|
mChannelEventBroadcaster.broadcast(new ChannelEvent(channel,
|
|
ChannelEvent.Event.NOTIFICATION_PROCESSING_START_REJECTED, TUNER_UNAVAILABLE_DESCRIPTION));
|
|
|
|
throw new ChannelException("No Tuner Available");
|
|
}
|
|
|
|
ProcessingChain processingChain = new ProcessingChain(channel, mAliasModel);
|
|
|
|
//Certain decoders aggregate the decode events in the parent channel that also includes any events produced
|
|
//by the traffic channels. Establish listener registration depending on if this channel is a traffic channel
|
|
//and the request contains the parent event history, or if this is a parent channel and the request contains
|
|
//the traffic channel event history.
|
|
if(request.hasParentDecodeEventHistory())
|
|
{
|
|
processingChain.getDecodeEventHistory().addListener(request.getParentDecodeEventHistory());
|
|
}
|
|
else if(request.hasChildDecodeEventHistory())
|
|
{
|
|
request.getChildDecodeEventHistory().addListener(processingChain.getDecodeEventHistory());
|
|
}
|
|
|
|
//Register to receive event bus requests/notifications
|
|
processingChain.getEventBus().register(ChannelProcessingManager.this);
|
|
|
|
mChannelEventBroadcaster.addListener(processingChain);
|
|
|
|
/* Register global listeners */
|
|
for(Listener<AudioSegment> listener : mAudioSegmentListeners)
|
|
{
|
|
processingChain.addAudioSegmentListener(listener);
|
|
}
|
|
|
|
for(Listener<IDecodeEvent> listener : mDecodeEventListeners)
|
|
{
|
|
processingChain.addDecodeEventListener(listener);
|
|
}
|
|
|
|
//Add a listener to detect source error state that indicates the channel should be shutdown.
|
|
//Note: processing chain will only add this once.
|
|
processingChain.addSourceEventListener(mSourceErrorListener);
|
|
|
|
//Register this manager to receive channel events from traffic channel manager modules within
|
|
//the processing chain
|
|
processingChain.addChannelEventListener(this);
|
|
|
|
//Register channel to receive frequency correction events to show in the spectral display (hack!)
|
|
processingChain.addFrequencyChangeListener(channel);
|
|
|
|
/* Processing Modules */
|
|
List<Module> modules = DecoderFactory.getModules(mChannelMapModel, channel, mAliasModel, mUserPreferences,
|
|
request.getTrafficChannelManager(), request.getChannelDescriptor());
|
|
processingChain.addModules(modules);
|
|
|
|
//Post preload data from the request to the event bus. Modules that can handle preload data will annotate
|
|
//their processor method with @Subscribe to receive each specific preload data content class.
|
|
for(PreloadDataContent preloadDataContent: request.getPreloadDataContents())
|
|
{
|
|
processingChain.getEventBus().post(preloadDataContent);
|
|
}
|
|
|
|
//Setup event logging
|
|
List<Module> loggers = mEventLogManager.getLoggers(channel);
|
|
|
|
if(!loggers.isEmpty())
|
|
{
|
|
processingChain.addModules(loggers);
|
|
}
|
|
|
|
//Add recorders
|
|
processingChain.addModules(RecorderFactory.getRecorders(mUserPreferences, channel));
|
|
|
|
//Set the samples source
|
|
processingChain.setSource(source);
|
|
|
|
//Inject the channel identifier for traffic channels and preload user identifiers
|
|
if(channel.isTrafficChannel())
|
|
{
|
|
if(request.hasChannelDescriptor() && request.hasIdentifierCollection())
|
|
{
|
|
for(int timeslot = 0; timeslot < request.getChannelDescriptor().getTimeslotCount(); timeslot++)
|
|
{
|
|
DecoderLogicalChannelNameIdentifier identifier =
|
|
DecoderLogicalChannelNameIdentifier.create(request.getChannelDescriptor().toString(),
|
|
request.getChannelDescriptor().getProtocol());
|
|
IdentifierUpdateNotification notification = new IdentifierUpdateNotification(identifier,
|
|
IdentifierUpdateNotification.Operation.ADD, timeslot);
|
|
processingChain.getChannelState().updateChannelStateIdentifiers(notification);
|
|
|
|
//Inject scramble parameters
|
|
for(Identifier scrambleParameters: request.getIdentifierCollection()
|
|
.getIdentifiers(Form.SCRAMBLE_PARAMETERS))
|
|
{
|
|
//Broadcast scramble parameters to both timeslots
|
|
IdentifierUpdateNotification scrambleNotification = new IdentifierUpdateNotification(scrambleParameters,
|
|
IdentifierUpdateNotification.Operation.ADD, timeslot);
|
|
processingChain.getChannelState().updateChannelStateIdentifiers(scrambleNotification);
|
|
}
|
|
}
|
|
}
|
|
|
|
for(Identifier userIdentifier : request.getIdentifierCollection().getIdentifiers(IdentifierClass.USER))
|
|
{
|
|
if(request.getChannelDescriptor().getTimeslotCount() > 1)
|
|
{
|
|
//Only broadcast an identifier update for the timeslot specified in the originating collection
|
|
IdentifierUpdateNotification notification = new IdentifierUpdateNotification(userIdentifier,
|
|
IdentifierUpdateNotification.Operation.ADD, request.getIdentifierCollection().getTimeslot());
|
|
processingChain.getChannelState().updateChannelStateIdentifiers(notification);
|
|
}
|
|
else
|
|
{
|
|
//Only broadcast an identifier update for the timeslot specified in the originating collection
|
|
IdentifierUpdateNotification notification = new IdentifierUpdateNotification(userIdentifier,
|
|
IdentifierUpdateNotification.Operation.ADD, 0);
|
|
processingChain.getChannelState().updateChannelStateIdentifiers(notification);
|
|
}
|
|
}
|
|
}
|
|
|
|
if(addProcessingChain(channel, processingChain))
|
|
{
|
|
processingChain.start();
|
|
|
|
if(GraphicsEnvironment.isHeadless())
|
|
{
|
|
channel.setProcessing(true);
|
|
}
|
|
else
|
|
{
|
|
//This has to be done on the FX event thread when the playlist editor is constructed
|
|
Platform.runLater(() -> channel.setProcessing(true));
|
|
}
|
|
|
|
mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_START));
|
|
}
|
|
else
|
|
{
|
|
mLog.warn("Channel [" + channel.getName() + "] processing chain not added because it already exists");
|
|
processingChain.removeEventLoggingModules();
|
|
processingChain.removeRecordingModules();
|
|
processingChain.removeFrequencyChangeListener(channel);
|
|
channel.resetFrequencyCorrection();
|
|
mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP));
|
|
mChannelEventBroadcaster.removeListener(processingChain);
|
|
processingChain.getEventBus().unregister(ChannelProcessingManager.this);
|
|
processingChain.dispose();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Thread-safe add processing chain and add channel metadata to channel metadata model.
|
|
* @param channel for the processing chain
|
|
* @param processingChain to add
|
|
* @return true if processing chain was added or false if it was not added due to there already
|
|
* being a processing chain registered for that channel.
|
|
*/
|
|
private boolean addProcessingChain(Channel channel, ProcessingChain processingChain)
|
|
{
|
|
boolean added = false;
|
|
|
|
mLock.lock();
|
|
|
|
try
|
|
{
|
|
if(!mProcessingChainsMap.containsKey(channel))
|
|
{
|
|
added = true;
|
|
mProcessingChainsMap.put(channel, processingChain);
|
|
getChannelMetadataModel().add(new ChannelAndMetadata(channel, processingChain.getChannelState().getChannelMetadata()));
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
mLock.unlock();
|
|
}
|
|
|
|
return added;
|
|
}
|
|
|
|
/**
|
|
* Thread-safe remove processing chain.
|
|
* @param channel for identifying the processing chain
|
|
* @return the removed processing chain or null
|
|
*/
|
|
private ProcessingChain removeProcessingChain(Channel channel)
|
|
{
|
|
ProcessingChain removed = null;
|
|
|
|
mLock.lock();
|
|
|
|
try
|
|
{
|
|
removed = mProcessingChainsMap.remove(channel);
|
|
|
|
if(removed != null)
|
|
{
|
|
for(ChannelMetadata channelMetadata: removed.getChannelState().getChannelMetadata())
|
|
{
|
|
getChannelMetadataModel().remove(channelMetadata);
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
mLock.unlock();
|
|
}
|
|
|
|
return removed;
|
|
}
|
|
|
|
/**
|
|
* Stops the channel/processing chain.
|
|
*
|
|
* @param channel to stop
|
|
*/
|
|
private void stopProcessing(Channel channel) throws ChannelException
|
|
{
|
|
ProcessingChain processingChain = removeProcessingChain(channel);
|
|
|
|
if(processingChain != null)
|
|
{
|
|
if(GraphicsEnvironment.isHeadless())
|
|
{
|
|
channel.setProcessing(false);
|
|
}
|
|
else
|
|
{
|
|
//This has to be done on the FX event thread when the playlist editor is constructed
|
|
Platform.runLater(() -> channel.setProcessing(false));
|
|
}
|
|
|
|
try
|
|
{
|
|
processingChain.stop();
|
|
processingChain.removeEventLoggingModules();
|
|
processingChain.removeRecordingModules();
|
|
|
|
//Deregister channel from receive frequency correction events to show in the spectral display (hack!)
|
|
processingChain.removeFrequencyChangeListener(channel);
|
|
channel.resetFrequencyCorrection();
|
|
|
|
//Notify all processing chains that this channel is shutting down so that if this is a traffic channel,
|
|
//the owning parent channel's traffic channel manager can cleanup it's accounting.
|
|
mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP));
|
|
mChannelEventBroadcaster.removeListener(processingChain);
|
|
|
|
//Unregister for event bus requests and notifications
|
|
processingChain.getEventBus().unregister(ChannelProcessingManager.this);
|
|
processingChain.dispose();
|
|
}
|
|
catch(Exception e)
|
|
{
|
|
mLog.error("Error during shutdown of processing chain for channel [" + channel.getName() + "}", e);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Stops all currently processing channels to prepare for shutdown.
|
|
*/
|
|
public void shutdown()
|
|
{
|
|
List<ScheduledFuture<?>> delayedTasks = new ArrayList<>(mDelayedChannelStartTasks);
|
|
|
|
for(ScheduledFuture<?> delayedTask: delayedTasks)
|
|
{
|
|
delayedTask.cancel(true);
|
|
mDelayedChannelStartTasks.remove(delayedTask);
|
|
}
|
|
|
|
List<Channel> channelsToStop = new ArrayList<>(mProcessingChainsMap.keySet());
|
|
|
|
for(Channel channel : channelsToStop)
|
|
{
|
|
try
|
|
{
|
|
stopProcessing(channel);
|
|
}
|
|
catch(ChannelException ce)
|
|
{
|
|
mLog.error("Error stopping channel [" + channel.getName() + "] - " + ce.getMessage());
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Process a request to convert a currently processing standard channel type to a traffic channel type.
|
|
* @param request from the currently processing channel's processing chain event bus
|
|
*/
|
|
@Subscribe
|
|
public void convertToTrafficChannel(ChannelConversionRequest request)
|
|
{
|
|
//Update the channel to processing chain map.
|
|
ProcessingChain processingChain = mProcessingChainsMap.remove(request.getCurrentChannel());
|
|
|
|
if(processingChain != null)
|
|
{
|
|
//Remove the traffic channel manager from this processing chain. Reuse or reinsertion of the traffic
|
|
//channel manager to another processing chain is handled separately.
|
|
processingChain.removeTrafficChannelManager();
|
|
|
|
//Update processing flag for each configuration.
|
|
Platform.runLater(() -> {
|
|
request.getCurrentChannel().setProcessing(false);
|
|
request.getTrafficChannel().setProcessing(true);
|
|
});
|
|
|
|
mProcessingChainsMap.put(request.getTrafficChannel(), processingChain);
|
|
mChannelMetadataModel.updateChannelMetadataToChannelMap(processingChain.getChannelState().getChannelMetadata(),
|
|
request.getTrafficChannel());
|
|
|
|
//Post a change notification so that processing chain modules can reconfigure
|
|
processingChain.channelConfigurationChanged(new ChannelConfigurationChangeNotification(request.getTrafficChannel()));
|
|
}
|
|
else
|
|
{
|
|
mLog.warn("Request to convert to traffic channel ignored - no processing chain was found");
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Adds a message listener that will be added to all channels to receive
|
|
* any messages.
|
|
*/
|
|
public void addAudioSegmentListener(Listener<AudioSegment> listener)
|
|
{
|
|
mAudioSegmentListeners.add(listener);
|
|
}
|
|
|
|
/**
|
|
* Removes a message listener.
|
|
*/
|
|
public void removeAudioSegmentListener(Listener<AudioSegment> listener)
|
|
{
|
|
mAudioSegmentListeners.remove(listener);
|
|
}
|
|
|
|
/**
|
|
* Adds a message listener that will be added to all channels to receive
|
|
* any messages.
|
|
*/
|
|
public void addDecodeEventListener(Listener<IDecodeEvent> listener)
|
|
{
|
|
mDecodeEventListeners.add(listener);
|
|
}
|
|
|
|
/**
|
|
* Removes a message listener.
|
|
*/
|
|
public void removeDecodeEventListener(Listener<IDecodeEvent> listener)
|
|
{
|
|
mDecodeEventListeners.remove(listener);
|
|
}
|
|
|
|
/**
|
|
* Adds a listener to receive channel events from this manager
|
|
*/
|
|
public void addChannelEventListener(Listener<ChannelEvent> listener)
|
|
{
|
|
mChannelEventBroadcaster.addListener(listener);
|
|
}
|
|
|
|
/**
|
|
* Removes the listener from receiving channel events from this manager
|
|
*/
|
|
public void removeChannelEventListener(Listener<ChannelEvent> listener)
|
|
{
|
|
mChannelEventBroadcaster.removeListener(listener);
|
|
}
|
|
|
|
/**
|
|
* Task to scheduling attempt to start a channel after previous attempts failed for lack of tuner channel
|
|
*/
|
|
public class DelayedChannelStartTask implements Runnable
|
|
{
|
|
private ChannelStartProcessingRequest mRequest;
|
|
|
|
public DelayedChannelStartTask(ChannelStartProcessingRequest request)
|
|
{
|
|
mRequest = request;
|
|
}
|
|
|
|
@Override
|
|
public void run()
|
|
{
|
|
try
|
|
{
|
|
mDelayedChannelStartTasks.remove(this);
|
|
startChannelRequest(mRequest);
|
|
}
|
|
catch(Throwable t)
|
|
{
|
|
mLog.error("Error executing persistent channel start task");
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Monitors all channels for an error in the source event that would require the
|
|
* channel's processing chain to be stopped
|
|
*/
|
|
private class ChannelSourceEventErrorListener implements Listener<SourceEvent>
|
|
{
|
|
@Override public void receive(SourceEvent sourceEvent)
|
|
{
|
|
if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE && sourceEvent.getSource() != null)
|
|
{
|
|
Channel toShutdown = null;
|
|
|
|
mLock.lock();
|
|
|
|
try
|
|
{
|
|
for(Map.Entry<Channel,ProcessingChain> entry: mProcessingChainsMap.entrySet())
|
|
{
|
|
if(entry.getValue().hasSource(sourceEvent.getSource()))
|
|
{
|
|
toShutdown = entry.getKey();
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
mLock.unlock();
|
|
}
|
|
|
|
if(toShutdown != null)
|
|
{
|
|
if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE)
|
|
{
|
|
mLog.warn("Channel source error detected - stopping channel [" + toShutdown.getName() + "]");
|
|
}
|
|
else
|
|
{
|
|
mLog.warn("Source event error - stopping channel [" + toShutdown.getName() + "]");
|
|
}
|
|
|
|
try
|
|
{
|
|
stopProcessing(toShutdown);
|
|
}
|
|
catch(ChannelException ce)
|
|
{
|
|
mLog.error("Error stopping channel [" + (toShutdown != null ? toShutdown.getName() : "unknown") +
|
|
"] with source error - " + ce.getMessage());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Creates a diagnostic report.
|
|
* @return report text.
|
|
*/
|
|
public String getDiagnosticInformation()
|
|
{
|
|
StringBuilder sb = new StringBuilder();
|
|
sb.append("Channel Processing Manager - Diagnostics Report\n\n");
|
|
sb.append(DIVIDER);
|
|
sb.append("\tChannel to Processing Chain Map Contents\n");
|
|
Map<Channel,ProcessingChain> mapCopy = new HashMap<>(mProcessingChainsMap);
|
|
for(Map.Entry<Channel,ProcessingChain> entry: mapCopy.entrySet())
|
|
{
|
|
sb.append("\n\n--------------- CHANNEL:PROCESSING CHAIN MAP ENTRY --------------------\n");
|
|
try
|
|
{
|
|
Channel channel = entry.getKey();
|
|
sb.append("\tChannel: " + channel).append("\n");
|
|
sb.append("\t\tSource Configuration: " + channel.getSourceConfiguration()).append("\n");
|
|
ProcessingChain chain = entry.getValue();
|
|
sb.append("\tProcessing Chain - Processing: ").append(chain.isProcessing()).append("\n");
|
|
AbstractChannelState state = chain.getChannelState();
|
|
sb.append("Channel State: ").append(state.getClass()).append("\n");
|
|
for(ChannelMetadata metadata: state.getChannelMetadata())
|
|
{
|
|
sb.append(metadata.getDescription()).append("\n");
|
|
}
|
|
|
|
Source source = chain.getSource();
|
|
|
|
if(source != null)
|
|
{
|
|
sb.append("Channel Source Class: " + source.getClass()).append("\n");
|
|
sb.append("\t\tTo String:").append(source).append("\n");
|
|
sb.append("\t\tHash:").append(Integer.toHexString(source.hashCode()).toUpperCase()).append("\n");
|
|
}
|
|
else
|
|
{
|
|
sb.append("Channel Source: (null)\n");
|
|
}
|
|
}
|
|
catch(Throwable t)
|
|
{
|
|
sb.append("\tError while logging diagnostics of map entry - " + t.getMessage()).append("\n");
|
|
}
|
|
}
|
|
|
|
return sb.toString();
|
|
}
|
|
}
|