#119

Committed 22 Sep 2023 10:58PM UTC coverage: 0.0%. Remained the same

Build # #119

Build Type

push

Committed by

Commit Message

Merge pull request #43 from WindhoverLabs/udp_fixes

Udp fixes

Run Details

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

0 of 6276 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/UdpStreamOutProvider.java

package com.windhoverlabs.yamcs.tctm;

import com.google.common.util.concurrent.RateLimiter;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.yamcs.AbstractYamcsService;
import org.yamcs.ConfigurationException;
import org.yamcs.InitException;
import org.yamcs.Spec;
import org.yamcs.YConfiguration;
import org.yamcs.events.EventProducer;
import org.yamcs.events.EventProducerFactory;
import org.yamcs.parameter.SystemParametersProducer;
import org.yamcs.tctm.Link;
import org.yamcs.utils.DataRateMeter;
import org.yamcs.yarch.ColumnDefinition;
import org.yamcs.yarch.DataType;
import org.yamcs.yarch.Stream;
import org.yamcs.yarch.StreamSubscriber;
import org.yamcs.yarch.Tuple;
import org.yamcs.yarch.TupleDefinition;
import org.yamcs.yarch.YarchDatabase;
import org.yamcs.yarch.YarchDatabaseInstance;

/**
 * Receives telemetry fames via UDP. One UDP datagram = one TM frame.
 *
 * @author nm
 */
public class UdpStreamOutProvider extends AbstractYamcsService
    implements Link, StreamSubscriber, SystemParametersProducer {
  String host;
  int port;
  DatagramSocket socket;
  InetAddress address;
  Thread thread;
  RateLimiter rateLimiter;
  protected Stream stream;
  protected AtomicLong packetCount = new AtomicLong(0);
  DataRateMeter packetRateMeter = new DataRateMeter();
  DataRateMeter dataRateMeter = new DataRateMeter();
  protected String linkName;
  protected AtomicBoolean disabled = new AtomicBoolean(false);
  private boolean receiving;
  private boolean enabledAtStartup = true;
  protected EventProducer eventProducer;

  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));
  }

  /**
   * Creates a new UDP Frame Data Link
   *
   * @throws ConfigurationException if port is not defined in the configuration
   */
  public void init(String instance, String name, YConfiguration config) {
    try {
      super.init(instance, name, config);
    } catch (InitException e1) {
      // TODO Auto-generated catch block
      e1.printStackTrace();
    }
    host = config.getString("host");
    port = config.getInt("port");
    enabledAtStartup = config.getBoolean("enabledAtStartup");
    String streamName = config.getString("stream");
    this.linkName = name;

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

    this.stream.addSubscriber(this);

    try {
      address = InetAddress.getByName(host);
    } catch (UnknownHostException e) {
      throw new ConfigurationException("Cannot resolve host '" + host + "'", e);
    }
    if (config.containsKey("frameMaxRate")) {
      rateLimiter = RateLimiter.create(config.getDouble("frameMaxRate"), 1, TimeUnit.SECONDS);
    }
    receiving = true;

    eventProducer =
        EventProducerFactory.getEventProducer(instance, this.getClass().getSimpleName(), 10000);

    if (!enabledAtStartup) {
      disabled.getAndSet(true);
    }
  }

  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
  public YConfiguration getConfig() {
    return config;
  }

  @Override
  public String getName() {
    return linkName;
  }

  @Override
  public void resetCounters() {
    packetCount.set(0);
  }

  @Override
  public void doStart() {
    if (!isDisabled()) {
      try {
        socket = new DatagramSocket();
      } catch (SocketException e) {
        notifyFailed(e);
      }
    }
    notifyStarted();
  }

  @Override
  public void doStop() {
    socket.close();
    notifyStopped();
  }

  public boolean isRunningAndEnabled() {
    State state = state();
    return (state == State.RUNNING || state == State.STARTING) && !disabled.get();
  }

  /**
   * called when a new packet is received to update the statistics
   *
   * @param packetSize
   */
  protected void updateStats(int packetSize) {
    packetCount.getAndIncrement();
    packetRateMeter.mark(1);
    dataRateMeter.mark(packetSize);
  }

  @Override
  public void disable() {
    boolean b = disabled.getAndSet(true);
    if (!b) {
      try {
        /* TODO */
        // doDisable();
      } catch (Exception e) {
        disabled.set(false);
        log.warn("Failed to disable link", e);
      }
    }
  }

  @Override
  public void enable() {
    /* TODO */
  }

  @Override
  public long getDataInCount() {
    return 0;
  }

  @Override
  public long getDataOutCount() {
    return packetCount.get();
  }

  @Override
  public Status getLinkStatus() {
    if (isDisabled()) {
      return Status.DISABLED;
    }
    if (state() == State.FAILED) {
      return Status.FAILED;
    }

    return connectionStatus();
  }

  @Override
  public boolean isDisabled() {
    return disabled.get();
  }

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

  @Override
  public void onTuple(Stream arg0, Tuple tuple) {
    if (isRunningAndEnabled() && receiving) {
      byte[] pktData = tuple.getColumn(DATA_CNAME);
      if (pktData == null) {
        throw new ConfigurationException("no column named '%s' in the tuple", DATA_CNAME);
      } else {
        DatagramPacket dtg = new DatagramPacket(pktData, pktData.length, address, port);
        try {
          //                synchronized (socket)
          //                {
          //                socket.send(dtg);
          //                }
          if (!socket.isClosed()) {
            socket.send(dtg);
          }
          updateStats(pktData.length);
        } catch (IOException e) {
          eventProducer.sendInfo("Error sending datagram:" + e.toString());
          log.warn("Error sending datagram", e);
          //          notifyFailed(e);
          return;
        }
      }
    }
  }

  @Override
  public Spec getSpec() {
    // TODO Auto-generated method stub
    return super.getSpec();
  }

  //  Not using the regular YAMCS "doStop" since I don't think that's meant to be called by API
  // users
  //   and I don't know if it'll have any side effects.
  public void stopReceving() {
    closeSockets();
    receiving = false;
  }

  public void startReceving() {
    if (receiving) {
      eventProducer.sendInfo("Link is already receving.");
      return;
    }
    if (disabled.get()) {
      disabled.getAndSet(false);
    }
    try {
      socket = new DatagramSocket(port);
    } catch (SocketException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    receiving = true;
  }

  private void closeSockets() {
    //          synchronized (socket)
    //          {
    //            socket.close();
    //
    //          }
    if (socket != null) {
      socket.close();
    }
  }
}

1	package com.windhoverlabs.yamcs.tctm;
2
3	import com.google.common.util.concurrent.RateLimiter;
4	import java.io.IOException;
5	import java.net.DatagramPacket;
6	import java.net.DatagramSocket;
7	import java.net.InetAddress;
8	import java.net.SocketException;
9	import java.net.UnknownHostException;
10	import java.util.concurrent.TimeUnit;
11	import java.util.concurrent.atomic.AtomicBoolean;
12	import java.util.concurrent.atomic.AtomicLong;
13	import org.yamcs.AbstractYamcsService;
14	import org.yamcs.ConfigurationException;
15	import org.yamcs.InitException;
16	import org.yamcs.Spec;
17	import org.yamcs.YConfiguration;
18	import org.yamcs.events.EventProducer;
19	import org.yamcs.events.EventProducerFactory;
20	import org.yamcs.parameter.SystemParametersProducer;
21	import org.yamcs.tctm.Link;
22	import org.yamcs.utils.DataRateMeter;
23	import org.yamcs.yarch.ColumnDefinition;
24	import org.yamcs.yarch.DataType;
25	import org.yamcs.yarch.Stream;
26	import org.yamcs.yarch.StreamSubscriber;
27	import org.yamcs.yarch.Tuple;
28	import org.yamcs.yarch.TupleDefinition;
29	import org.yamcs.yarch.YarchDatabase;
30	import org.yamcs.yarch.YarchDatabaseInstance;
31
32	/**
33	* Receives telemetry fames via UDP. One UDP datagram = one TM frame.
34	*
35	* @author nm
36	*/
37	public class UdpStreamOutProvider extends AbstractYamcsService	×
38	implements Link, StreamSubscriber, SystemParametersProducer {
39	String host;
40	int port;
41	DatagramSocket socket;
42	InetAddress address;
43	Thread thread;
44	RateLimiter rateLimiter;
45	protected Stream stream;
46	protected AtomicLong packetCount = new AtomicLong(0);	×
47	DataRateMeter packetRateMeter = new DataRateMeter();	×
48	DataRateMeter dataRateMeter = new DataRateMeter();	×
49	protected String linkName;
50	protected AtomicBoolean disabled = new AtomicBoolean(false);	×
51	private boolean receiving;
52	private boolean enabledAtStartup = true;	×
53	protected EventProducer eventProducer;
54
55	static TupleDefinition gftdef;
56
57	static final String RECTIME_CNAME = "rectime";
58	static final String DATA_CNAME = "data";
59
60	static {
61	gftdef = new TupleDefinition();	×
62	gftdef.addColumn(new ColumnDefinition(RECTIME_CNAME, DataType.TIMESTAMP));	×
63	gftdef.addColumn(new ColumnDefinition(DATA_CNAME, DataType.BINARY));	×
64	}	×
65
66	/**
67	* Creates a new UDP Frame Data Link
68	*
69	* @throws ConfigurationException if port is not defined in the configuration
70	*/
71	public void init(String instance, String name, YConfiguration config) {
72	try {
73	super.init(instance, name, config);	×
74	} catch (InitException e1) {	×
75	// TODO Auto-generated catch block
76	e1.printStackTrace();	×
77	}	×
78	host = config.getString("host");	×
79	port = config.getInt("port");	×
80	enabledAtStartup = config.getBoolean("enabledAtStartup");	×
81	String streamName = config.getString("stream");	×
82	this.linkName = name;	×
83
84	YarchDatabaseInstance ydb = YarchDatabase.getInstance(instance);	×
85	this.stream = getStream(ydb, streamName);	×
86
87	this.stream.addSubscriber(this);	×
88
89	try {
90	address = InetAddress.getByName(host);	×
91	} catch (UnknownHostException e) {	×
92	throw new ConfigurationException("Cannot resolve host '" + host + "'", e);	×
93	}	×
94	if (config.containsKey("frameMaxRate")) {	×
95	rateLimiter = RateLimiter.create(config.getDouble("frameMaxRate"), 1, TimeUnit.SECONDS);	×
96	}
97	receiving = true;	×
98
99	eventProducer =	×
100	EventProducerFactory.getEventProducer(instance, this.getClass().getSimpleName(), 10000);	×
101
102	if (!enabledAtStartup) {	×
103	disabled.getAndSet(true);	×
104	}
105	}	×
106
107	private static Stream getStream(YarchDatabaseInstance ydb, String streamName) {
108	Stream stream = ydb.getStream(streamName);	×
109	if (stream == null) {	×
110	try {
111	ydb.execute("create stream " + streamName + gftdef.getStringDefinition());	×
112	// ydb.execute("create stream " + streamName);
113	} catch (Exception e) {	×
114	throw new ConfigurationException(e);	×
115	}	×
116	stream = ydb.getStream(streamName);	×
117	}
118	return stream;	×
119	}
120
121	@Override
122	public YConfiguration getConfig() {
123	return config;	×
124	}
125
126	@Override
127	public String getName() {
128	return linkName;	×
129	}
130
131	@Override
132	public void resetCounters() {
133	packetCount.set(0);	×
134	}	×
135
136	@Override
137	public void doStart() {
138	if (!isDisabled()) {	×
139	try {
140	socket = new DatagramSocket();	×
141	} catch (SocketException e) {	×
142	notifyFailed(e);	×
143	}	×
144	}
145	notifyStarted();	×
146	}	×
147
148	@Override
149	public void doStop() {
150	socket.close();	×
151	notifyStopped();	×
152	}	×
153
154	public boolean isRunningAndEnabled() {
155	State state = state();	×
156	return (state == State.RUNNING \|\| state == State.STARTING) && !disabled.get();	×
157	}
158
159	/**
160	* called when a new packet is received to update the statistics
161	*
162	* @param packetSize
163	*/
164	protected void updateStats(int packetSize) {
165	packetCount.getAndIncrement();	×
166	packetRateMeter.mark(1);	×
167	dataRateMeter.mark(packetSize);	×
168	}	×
169
170	@Override
171	public void disable() {
172	boolean b = disabled.getAndSet(true);	×
173	if (!b) {	×
174	try {
175	/* TODO */
176	// doDisable();
177	} catch (Exception e) {
178	disabled.set(false);
179	log.warn("Failed to disable link", e);
180	}
181	}
182	}	×
183
184	@Override
185	public void enable() {
186	/* TODO */
187	}	×
188
189	@Override
190	public long getDataInCount() {
191	return 0;	×
192	}
193
194	@Override
195	public long getDataOutCount() {
196	return packetCount.get();	×
197	}
198
199	@Override
200	public Status getLinkStatus() {
201	if (isDisabled()) {	×
202	return Status.DISABLED;	×
203	}
204	if (state() == State.FAILED) {	×
205	return Status.FAILED;	×
206	}
207
208	return connectionStatus();	×
209	}
210
211	@Override
212	public boolean isDisabled() {
213	return disabled.get();	×
214	}
215
216	protected Status connectionStatus() {
217	return Status.OK;	×
218	}
219
220	@Override
221	public void onTuple(Stream arg0, Tuple tuple) {
222	if (isRunningAndEnabled() && receiving) {	×
223	byte[] pktData = tuple.getColumn(DATA_CNAME);	×
224	if (pktData == null) {	×
225	throw new ConfigurationException("no column named '%s' in the tuple", DATA_CNAME);	×
226	} else {
227	DatagramPacket dtg = new DatagramPacket(pktData, pktData.length, address, port);	×
228	try {
229	// synchronized (socket)
230	// {
231	// socket.send(dtg);
232	// }
233	if (!socket.isClosed()) {	×
234	socket.send(dtg);	×
235	}
236	updateStats(pktData.length);	×
237	} catch (IOException e) {	×
238	eventProducer.sendInfo("Error sending datagram:" + e.toString());	×
239	log.warn("Error sending datagram", e);	×
240	// notifyFailed(e);
241	return;	×
242	}	×
243	}
244	}
245	}	×
246
247	@Override
248	public Spec getSpec() {
249	// TODO Auto-generated method stub
250	return super.getSpec();	×
251	}
252
253	// Not using the regular YAMCS "doStop" since I don't think that's meant to be called by API
254	// users
255	// and I don't know if it'll have any side effects.
256	public void stopReceving() {
257	closeSockets();	×
258	receiving = false;	×
259	}	×
260
261	public void startReceving() {
262	if (receiving) {	×
263	eventProducer.sendInfo("Link is already receving.");	×
264	return;	×
265	}
266	if (disabled.get()) {	×
267	disabled.getAndSet(false);	×
268	}
269	try {
270	socket = new DatagramSocket(port);	×
271	} catch (SocketException e) {	×
272	// TODO Auto-generated catch block
273	e.printStackTrace();	×
274	}	×
275	receiving = true;	×
276	}	×
277
278	private void closeSockets() {
279	// synchronized (socket)
280	// {
281	// socket.close();
282	//
283	// }
284	if (socket != null) {	×
285	socket.close();	×
286	}
287	}	×
288	}

WindhoverLabs / yamcs-cfs / #119

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