#157

Committed 25 Nov 2024 11:22PM UTC coverage: 0.0%. Remained the same

Build # #157

Build Type

push

Committed by

Commit Message

Merge 096a65ddf into e64813ef2

Run Details

0 of 18 new or added lines in 13 files covered. (0.0%)

1 existing line in 1 file now uncovered.

0 of 6640 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/tctm/StreamTcDataLink.java

package com.windhoverlabs.yamcs.tctm;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.yamcs.ConfigurationException;
import org.yamcs.YConfiguration;
import org.yamcs.YamcsServer;
import org.yamcs.commanding.PreparedCommand;
import org.yamcs.tctm.AbstractThreadedTcDataLink;
import org.yamcs.tctm.GenericCommandPostprocessor;
import org.yamcs.yarch.ColumnDefinition;
import org.yamcs.yarch.DataType;
import org.yamcs.yarch.Stream;
import org.yamcs.yarch.Tuple;
import org.yamcs.yarch.TupleDefinition;
import org.yamcs.yarch.YarchDatabase;
import org.yamcs.yarch.YarchDatabaseInstance;

/**
 * Sends packets out a YAMCS Stream.
 *
 * @author nm
 */
public class StreamTcDataLink extends AbstractThreadedTcDataLink {

  protected String streamName;
  protected Stream stream;

  static TupleDefinition gftdef;

  static final String RECTIME_CNAME = "rectime";
  static final String DATA_CNAME = "data";

  static {
    gftdef = new TupleDefinition();
    gftdef.addColumn(new ColumnDefinition(RECTIME_CNAME, DataType.TIMESTAMP));
    gftdef.addColumn(new ColumnDefinition(DATA_CNAME, DataType.BINARY));
  }

  @Override
  public void init(String yamcsInstance, String name, YConfiguration config)
      throws ConfigurationException {
    super.init(yamcsInstance, name, config);
    timeService = YamcsServer.getTimeService(yamcsInstance);

    this.streamName = config.getString("out_stream");

    YarchDatabaseInstance ydb = YarchDatabase.getInstance(yamcsInstance);
    this.stream = getStream(ydb, streamName);
  }

  @Override
  public String getDetailedStatus() {
    return String.format("OK");
  }

  @Override
  protected void initPostprocessor(String instance, YConfiguration config) {
    Map<String, Object> m = null;
    if (config == null) {
      m = new HashMap<>();
      config = YConfiguration.wrap(m);
    } else if (!config.containsKey("commandPostprocessorClassName")) {
      m = config.getRoot();
    }
    if (m != null) {
      log.warn(
          "Please set the commandPostprocessorClassName for the StreamTcDataLink; in the future versions it will default to GenericCommandPostprocessor");
      m.put("commandPostprocessorClassName", GenericCommandPostprocessor.class.getName());
    }
    super.initPostprocessor(instance, config);
  }

  @Override
  public void uplinkCommand(PreparedCommand pc) throws IOException {

    byte[] binary = postprocess(pc);
    if (binary == null) {
      return;
    }

    this.stream.emitTuple(new Tuple(this.gftdef, Arrays.asList(pc.getGenerationTime(), binary)));

    dataOut(1, binary.length);
    ackCommand(pc.getCommandId());
  }

  @Override
  protected Status connectionStatus() {
    return Status.OK;
  }

  @Override
  public void shutDown() {}

  @Override
  protected void startUp() {}

  private static Stream getStream(YarchDatabaseInstance ydb, String streamName) {
    Stream stream = ydb.getStream(streamName);
    if (stream == null) {
      try {
        ydb.execute("create stream " + streamName + gftdef.getStringDefinition());
        // ydb.execute("create stream " + streamName);
      } catch (Exception e) {
        throw new ConfigurationException(e);
      }
      stream = ydb.getStream(streamName);
    }
    return stream;
  }

  @Override
  protected void doHousekeeping() {
    if (!isRunningAndEnabled()) {
      return;
    }
  }
}

1	package com.windhoverlabs.yamcs.tctm;
2
3	import java.io.IOException;
4	import java.util.Arrays;
5	import java.util.HashMap;
6	import java.util.Map;
7	import org.yamcs.ConfigurationException;
8	import org.yamcs.YConfiguration;
9	import org.yamcs.YamcsServer;
10	import org.yamcs.commanding.PreparedCommand;
11	import org.yamcs.tctm.AbstractThreadedTcDataLink;
12	import org.yamcs.tctm.GenericCommandPostprocessor;
13	import org.yamcs.yarch.ColumnDefinition;
14	import org.yamcs.yarch.DataType;
15	import org.yamcs.yarch.Stream;
16	import org.yamcs.yarch.Tuple;
17	import org.yamcs.yarch.TupleDefinition;
18	import org.yamcs.yarch.YarchDatabase;
19	import org.yamcs.yarch.YarchDatabaseInstance;
20
21	/**
22	* Sends packets out a YAMCS Stream.
23	*
24	* @author nm
25	*/
26	public class StreamTcDataLink extends AbstractThreadedTcDataLink {	×
27
28	protected String streamName;
29	protected Stream stream;
30
31	static TupleDefinition gftdef;
32
33	static final String RECTIME_CNAME = "rectime";
34	static final String DATA_CNAME = "data";
35
36	static {
37	gftdef = new TupleDefinition();	×
38	gftdef.addColumn(new ColumnDefinition(RECTIME_CNAME, DataType.TIMESTAMP));	×
39	gftdef.addColumn(new ColumnDefinition(DATA_CNAME, DataType.BINARY));	×
40	}	×
41
42	@Override
43	public void init(String yamcsInstance, String name, YConfiguration config)
44	throws ConfigurationException {
45	super.init(yamcsInstance, name, config);	×
46	timeService = YamcsServer.getTimeService(yamcsInstance);	×
47
48	this.streamName = config.getString("out_stream");	×
49
50	YarchDatabaseInstance ydb = YarchDatabase.getInstance(yamcsInstance);	×
51	this.stream = getStream(ydb, streamName);	×
52	}	×
53
54	@Override
55	public String getDetailedStatus() {
56	return String.format("OK");	×
57	}
58
59	@Override
60	protected void initPostprocessor(String instance, YConfiguration config) {
61	Map<String, Object> m = null;	×
62	if (config == null) {	×
63	m = new HashMap<>();	×
64	config = YConfiguration.wrap(m);	×
65	} else if (!config.containsKey("commandPostprocessorClassName")) {	×
66	m = config.getRoot();	×
67	}
68	if (m != null) {	×
69	log.warn(	×
70	"Please set the commandPostprocessorClassName for the StreamTcDataLink; in the future versions it will default to GenericCommandPostprocessor");
71	m.put("commandPostprocessorClassName", GenericCommandPostprocessor.class.getName());	×
72	}
73	super.initPostprocessor(instance, config);	×
74	}	×
75
76	@Override
77	public void uplinkCommand(PreparedCommand pc) throws IOException {
78
79	byte[] binary = postprocess(pc);	×
80	if (binary == null) {	×
81	return;	×
82	}
83
84	this.stream.emitTuple(new Tuple(this.gftdef, Arrays.asList(pc.getGenerationTime(), binary)));	×
85
NEW 86	dataOut(1, binary.length);	×
87	ackCommand(pc.getCommandId());	×
88	}	×
89
90	@Override
91	protected Status connectionStatus() {
92	return Status.OK;	×
93	}
94
95	@Override
96	public void shutDown() {}	×
97
98	@Override
99	protected void startUp() {}	×
100
101	private static Stream getStream(YarchDatabaseInstance ydb, String streamName) {
102	Stream stream = ydb.getStream(streamName);	×
103	if (stream == null) {	×
104	try {
105	ydb.execute("create stream " + streamName + gftdef.getStringDefinition());	×
106	// ydb.execute("create stream " + streamName);
107	} catch (Exception e) {	×
108	throw new ConfigurationException(e);	×
109	}	×
110	stream = ydb.getStream(streamName);	×
111	}
112	return stream;	×
113	}
114
115	@Override
116	protected void doHousekeeping() {
117	if (!isRunningAndEnabled()) {	×
118	return;	×
119	}
120	}	×
121	}

WindhoverLabs / yamcs-cfs / #157

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