}
duration = (Integer)v *1000L;
}
parameterCacheConfig = new ParameterCacheConfig(enabled, cacheAll, duration);
}
private static String key(String instance, String name) {
return instance+"."+name;
}
public CommandHistoryPublisher getCommandHistoryPublisher() {
return commandHistoryPublisher;
}
public ParameterRequestManagerImpl getParameterRequestManager() {
return parameterRequestManager;
}
public ContainerRequestManager getContainerRequestManager() {
return containerRequestManager;
}
public XtceTmProcessor getTmProcessor() {
return tmProcessor;
}
/**
* starts processing by invoking the start method for all the associated components
*
*/
@Override
public void doStart() {
if(tmPacketProvider!=null) {
tmPacketProvider.startAsync();
}
if(tmProcessor!=null) {
tmProcessor.startAsync();
}
if(commandReleaser!=null) {
commandReleaser.startAsync();
commandReleaser.awaitRunning();
commandingManager.startAsync();
commandingManager.awaitRunning();
CommandQueueManager cqm = commandingManager.getCommandQueueManager();
cqm.startAsync();
cqm.awaitRunning();
}
if(commandHistoryRequestManager!=null) {
commandHistoryRequestManager.startAsync();
startIfNecessary(commandHistoryProvider);
commandHistoryRequestManager.awaitRunning();
commandHistoryProvider.awaitRunning();
}
for(ParameterProvider pprov: parameterProviders) {
pprov.startAsync();
}
parameterRequestManager.start();
if(tmPacketProvider!=null) {
tmPacketProvider.awaitRunning();
}
if(tmProcessor!=null) {
tmProcessor.awaitRunning();
}
notifyStarted();
propagateProcessorStateChange();
}
private void startIfNecessary(Service service) {
if(service.state()==State.NEW) {
service.startAsync();
}
}
public void pause() {
((ArchiveTmPacketProvider)tmPacketProvider).pause();
propagateProcessorStateChange();
}
public void resume() {
((ArchiveTmPacketProvider)tmPacketProvider).resume();
propagateProcessorStateChange();
}
private void propagateProcessorStateChange() {
listeners.forEach(l -> l.processorStateChanged(this));
}
public void seek(long instant) {
getTmProcessor().resetStatistics();
((ArchiveTmPacketProvider)tmPacketProvider).seek(instant);
propagateProcessorStateChange();
}
public void changeSpeed(ReplaySpeed speed) {
((ArchiveTmPacketProvider)tmPacketProvider).changeSpeed(speed);
propagateProcessorStateChange();
}
/**
* @return the tcUplinker
*/
public CommandReleaser getCommandReleaser() {
return commandReleaser;
}
/**
* @return the tmPacketProvider
*/
public TmPacketProvider getTmPacketProvider() {
return tmPacketProvider;
}
public String getName() {
return name;
}
/**
* @return the type
*/
public String getType() {
return type;
}
public String getCreator() {
return creator;
}
public void setCreator(String creator) {
this.creator = creator;
}
public int getConnectedClients() {
return connectedClients.size();
}
public static YProcessor getInstance(String yamcsInstance, String name) {
return instances.get(key(yamcsInstance,name));
}
/**
* Increase with one the number of connected clients to the named processor and return the processor.
* @param name
* @return the processor with the given name
* @throws YProcessorException
*/
public static YProcessor connect(String yamcsInstance, String name, YProcessorClient s) throws YProcessorException {
YProcessor ds=instances.get(key(yamcsInstance,name));
if(ds==null) throw new YProcessorException("There is no processor named '"+name+"'");
ds.connect(s);
return ds;
}
/**
* Increase with one the number of connected clients
*/
public synchronized void connect(YProcessorClient s) throws YProcessorException {
log.debug("Session "+name+" has one more user: " +s);
if(quitting) throw new YProcessorException("This processor has been closed");
connectedClients.add(s);
}
/**
* Disconnects a client from this processor. If the processor has no more clients, quit.
*
*/
public void disconnect(YProcessorClient s) {
if(quitting) return;
boolean hasToQuit=false;
synchronized(this) {
connectedClients.remove(s);
log.info("Processor "+name+" has one less user: connectedUsers: "+connectedClients.size());
if((connectedClients.isEmpty())&&(!persistent)) {
hasToQuit=true;
}
}
if(hasToQuit) stopAsync();
}
public static Collection getProcessors() {
return instances.values();
}
public static Collection getProcessors(String instance) {
List processors = new ArrayList<>();
for (YProcessor processor : instances.values()) {
if (instance.equals(processor.getInstance())) {
processors.add(processor);
}
}
return instances.values();
}
/**
* Closes the processor by stoping the tm/pp and tc
* It can be that there are still clients connected, but they will not get any data and new clients can not connect to
* these processors anymore. Once it is closed, you can create a processor with the same name which will make it maybe a bit
* confusing :(
*
*/
@Override
public void doStop() {
if(quitting)return;
log.info("Processor "+name+" quitting");
quitting=true;
instances.remove(key(yamcsInstance,name));
for(ParameterProvider p:parameterProviders) {
p.stopAsync();
}
if(commandReleaser!=null) commandReleaser.stopAsync();
log.info("Processor "+name+" is out of business");
listeners.forEach(l -> l.yProcessorClosed(this));
synchronized(this) {
for(YProcessorClient s:connectedClients) {
s.yProcessorQuit();
}
}
if(getState() == ServiceState.RUNNING || getState() == ServiceState.STOPPING)
notifyStopped();
}
public static void addProcessorListener(YProcessorListener processorListener) {
listeners.add(processorListener);
}
public static void removeProcessorListener(YProcessorListener processorListener) {
listeners.remove(processorListener);
}
public boolean isPersistent() {
return persistent;
}
public void setPersistent(boolean systemSession) {
this.persistent = systemSession;
}
public boolean isSynchronous() {
return synchronous;
}
public boolean hasCommanding() {
return (commandingManager!=null);
}
public void setSynchronous(boolean synchronous) {
this.synchronous = synchronous;
}
public boolean isReplay() {
if(tmPacketProvider==null) return false;
return tmPacketProvider.isArchiveReplay();
}
/**
* valid only if isArchiveReplay returns true
* @return
*/
public ReplayRequest getReplayRequest() {
return ((ArchiveTmPacketProvider)tmPacketProvider).getReplayRequest();
}
/**
* valid only if isArchiveReplay returns true
* @return
*/
public ReplayState getReplayState() {
return ((ArchiveTmPacketProvider)tmPacketProvider).getReplayState();
}
public ServiceState getState() {
return ServiceState.valueOf(state().name());
}
public CommandingManager getCommandingManager() {
return commandingManager;
}
@Override
public String toString() {
return "name: "+name+" type: "+type+" connectedClients:"+connectedClients.size();
}
/**
*
* @return the yamcs instance this processor is part of
*/
public String getInstance() {
return yamcsInstance;
}
public XtceDb getXtceDb() {
return xtcedb;
}
public CommandHistoryRequestManager getCommandHistoryManager() {
return commandHistoryRequestManager;
}
public boolean hasAlarmChecker() {
return checkAlarms;
}
public boolean hasAlarmServer() {
return alarmServerEnabled;
}
public ParameterCacheConfig getPameterCacheConfig () {
return parameterCacheConfig;
}
public ParameterCache getParameterCache() {
return parameterRequestManager.getParameterCache();
}
public ScheduledThreadPoolExecutor getTimer() {
return timer;
}
/**
* Returns the processor time
*
* for realtime processors it is the mission time or simulation time
* for replay processors it is the replay time
* @return
*/
public long getCurrentTime() {
if(isReplay()) {
return ((ArchiveTmPacketProvider)tmPacketProvider).getReplayTime();
} else {
return timeService.getMissionTime();
}
}
public void quit() {
stopAsync();
awaitTerminated();
}
public void start() {
startAsync();
awaitRunning();
}
public void notifyStateChange() {
propagateProcessorStateChange();
}
} |