WindhoverLabs / yamcs-cfs / #91

Build # #91

Build Type

push

Committed by

Commit Message

-Add ReplayToUDPService. WIP

Run Details

37 of 37 new or added lines in 1 file covered. (100.0%)

0 of 4442 relevant lines covered (0.0%)

0.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/main/java/com/windhoverlabs/yamcs/archive/ReplayToUDPService.java

package com.windhoverlabs.yamcs.archive;

import com.google.protobuf.Timestamp;
import com.google.protobuf.util.Timestamps;
import java.text.ParseException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import org.yamcs.AbstractYamcsService;
import org.yamcs.ConfigurationException;
import org.yamcs.InitException;
import org.yamcs.Processor;
import org.yamcs.ProcessorException;
import org.yamcs.ProcessorFactory;
import org.yamcs.ValidationException;
import org.yamcs.YConfiguration;
import org.yamcs.archive.ReplayOptions;
import org.yamcs.client.ParameterSubscription;
import org.yamcs.logging.Log;
import org.yamcs.protobuf.Yamcs.EndAction;
import org.yamcs.protobuf.Yamcs.ReplayRequest;

public class ReplayToUDPService extends AbstractYamcsService {
  private ReplayOptions replayOptions;
  private String processorName;
  private Processor processor;

  private String start;
  private boolean reverse; // TODO: Add to conifg
  private String stop;
  private Timestamp startTimeStamp;
  private Timestamp stopTimeStamp;

  private ParameterSubscription subscription;

  private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
  private AtomicBoolean subscriptionDirty = new AtomicBoolean(false);

  @Override
  public void init(String yamcsInstance, String serviceName, YConfiguration config)
      throws InitException {
    this.yamcsInstance = yamcsInstance;
    this.serviceName = serviceName;
    this.config = config;
    processorName = this.config.getString("", "ReplayToUDPService");

    start = this.config.getString("start");
    stop = this.config.getString("stop");
    log = new Log(getClass(), yamcsInstance);
    try {
      startTimeStamp = Timestamps.parse(start);
      stopTimeStamp = Timestamps.parse(stop);
    } catch (ParseException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }

    //    TODO:Add these options to YAML
    replayOptions =
        new ReplayOptions(
            ReplayRequest.newBuilder()
                .setStart(startTimeStamp)
                .setStop(stopTimeStamp)
                .setEndAction(EndAction.LOOP)
                .setAutostart(true)
                .build());

    try {
      processor =
          ProcessorFactory.create(
              yamcsInstance,
              processorName,
              "Archive",
              ReplayToUDPService.class.toString(),
              replayOptions);
    } catch (ProcessorException | ConfigurationException | ValidationException | InitException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
  }

  @Override
  protected void doStart() {
    // TODO Auto-generated method stub

    log.info("Starting new processor '{}'", processor.getName());
    processor.startAsync();
    processor.awaitRunning();
    processor.getTmPacketProvider();

    // Periodically check if the subscription needs a refresh
    // (PVs send individual events, so this bundles them)
    //    executor.scheduleWithFixedDelay(
    //        () -> {
    //          if (subscriptionDirty.getAndSet(false) && subscription != null) {
    //            Set<NamedObjectId> ids = getRequestedIdentifiers();
    //            // TODO:Make log level configurable
    //            //            log.info(String.format("Modifying subscription to %s", ids));
    //            subscription.sendMessage(
    //                SubscribeParametersRequest.newBuilder()
    //                    .setAction(Action.REPLACE)
    //                    .setSendFromCache(true)
    //                    .setAbortOnInvalid(false)
    //                    .setUpdateOnExpiration(true)
    //                    .addAllId(ids)
    //                    .setProcessor(currentProcessor)
    //                    .build());
    //          }
    //        },
    //        500,
    //        500,
    //        TimeUnit.MILLISECONDS);
    //  }
    notifyStarted();
  }

  @Override
  protected void doStop() {
    // TODO Auto-generated method stub
    notifyStopped();
  }
}

1	package com.windhoverlabs.yamcs.archive;
2
3	import com.google.protobuf.Timestamp;
4	import com.google.protobuf.util.Timestamps;
5	import java.text.ParseException;
6	import java.util.concurrent.Executors;
7	import java.util.concurrent.ScheduledExecutorService;
8	import java.util.concurrent.atomic.AtomicBoolean;
9	import org.yamcs.AbstractYamcsService;
10	import org.yamcs.ConfigurationException;
11	import org.yamcs.InitException;
12	import org.yamcs.Processor;
13	import org.yamcs.ProcessorException;
14	import org.yamcs.ProcessorFactory;
15	import org.yamcs.ValidationException;
16	import org.yamcs.YConfiguration;
17	import org.yamcs.archive.ReplayOptions;
18	import org.yamcs.client.ParameterSubscription;
19	import org.yamcs.logging.Log;
20	import org.yamcs.protobuf.Yamcs.EndAction;
21	import org.yamcs.protobuf.Yamcs.ReplayRequest;
22
23	public class ReplayToUDPService extends AbstractYamcsService {	×
24	private ReplayOptions replayOptions;
25	private String processorName;
26	private Processor processor;
27
28	private String start;
29	private boolean reverse; // TODO: Add to conifg
30	private String stop;
31	private Timestamp startTimeStamp;
32	private Timestamp stopTimeStamp;
33
34	private ParameterSubscription subscription;
35
36	private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();	×
37	private AtomicBoolean subscriptionDirty = new AtomicBoolean(false);	×
38
39	@Override
40	public void init(String yamcsInstance, String serviceName, YConfiguration config)
41	throws InitException {
42	this.yamcsInstance = yamcsInstance;	×
43	this.serviceName = serviceName;	×
44	this.config = config;	×
45	processorName = this.config.getString("", "ReplayToUDPService");	×
46
47	start = this.config.getString("start");	×
48	stop = this.config.getString("stop");	×
49	log = new Log(getClass(), yamcsInstance);	×
50	try {
51	startTimeStamp = Timestamps.parse(start);	×
52	stopTimeStamp = Timestamps.parse(stop);	×
53	} catch (ParseException e) {	×
54	// TODO Auto-generated catch block
55	e.printStackTrace();	×
56	}	×
57
58	// TODO:Add these options to YAML
59	replayOptions =	×
60	new ReplayOptions(
61	ReplayRequest.newBuilder()	×
62	.setStart(startTimeStamp)	×
63	.setStop(stopTimeStamp)	×
64	.setEndAction(EndAction.LOOP)	×
65	.setAutostart(true)	×
66	.build());	×
67
68	try {
69	processor =	×
70	ProcessorFactory.create(	×
71	yamcsInstance,
72	processorName,
73	"Archive",
74	ReplayToUDPService.class.toString(),	×
75	replayOptions);
76	} catch (ProcessorException \| ConfigurationException \| ValidationException \| InitException e) {	×
77	// TODO Auto-generated catch block
78	e.printStackTrace();	×
79	}	×
80	}	×
81
82	@Override
83	protected void doStart() {
84	// TODO Auto-generated method stub
85
86	log.info("Starting new processor '{}'", processor.getName());	×
87	processor.startAsync();	×
88	processor.awaitRunning();	×
89	processor.getTmPacketProvider();	×
90
91	// Periodically check if the subscription needs a refresh
92	// (PVs send individual events, so this bundles them)
93	// executor.scheduleWithFixedDelay(
94	// () -> {
95	// if (subscriptionDirty.getAndSet(false) && subscription != null) {
96	// Set<NamedObjectId> ids = getRequestedIdentifiers();
97	// // TODO:Make log level configurable
98	// // log.info(String.format("Modifying subscription to %s", ids));
99	// subscription.sendMessage(
100	// SubscribeParametersRequest.newBuilder()
101	// .setAction(Action.REPLACE)
102	// .setSendFromCache(true)
103	// .setAbortOnInvalid(false)
104	// .setUpdateOnExpiration(true)
105	// .addAllId(ids)
106	// .setProcessor(currentProcessor)
107	// .build());
108	// }
109	// },
110	// 500,
111	// 500,
112	// TimeUnit.MILLISECONDS);
113	// }
114	notifyStarted();	×
115	}	×
116
117	@Override
118	protected void doStop() {
119	// TODO Auto-generated method stub
120	notifyStopped();	×
121	}	×
122	}

WindhoverLabs / yamcs-cfs / #91

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous