WindhoverLabs / phoebus / #94

Committed 18 Oct 2023 08:23PM UTC coverage: 16.596% (+0.05%) from 16.544%

Build # #94

Build Type

Committed by

Commit Message

-Remove app-commander-parameter-export from build.

Pull Request Pull Request #90: WIP - 88 out of memory issues

Run Details

217 of 217 new or added lines in 9 files covered. (100.0%)

17830 of 107434 relevant lines covered (16.6%)

0.17 hits per line

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

49.24

/core/commander-core/src/main/java/com/windhoverlabs/yamcs/core/CMDR_YamcsInstance.java

package com.windhoverlabs.yamcs.core;

import com.windhoverlabs.pv.yamcs.YamcsPV;
import com.windhoverlabs.pv.yamcs.YamcsSubscriptionService;
import com.windhoverlabs.yamcs.core.YamcsWebSocketClient.TmStatistics;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.function.Consumer;
import java.util.logging.Logger;
import javafx.collections.FXCollections;
import javafx.collections.ObservableList;
import org.yamcs.TmPacket;
import org.yamcs.client.CommandSubscription;
import org.yamcs.client.EventSubscription;
import org.yamcs.client.LinkSubscription;
import org.yamcs.client.MessageListener;
import org.yamcs.client.PacketSubscription;
import org.yamcs.client.Page;
import org.yamcs.client.YamcsClient;
import org.yamcs.client.archive.ArchiveClient;
import org.yamcs.client.mdb.MissionDatabaseClient.ListOptions;
import org.yamcs.client.processor.ProcessorClient;
import org.yamcs.mdb.ProcessingStatistics;
import org.yamcs.protobuf.Commanding.CommandHistoryEntry;
// import org.yamcs.protobuf.Event;
import org.yamcs.protobuf.CreateEventRequest;
import org.yamcs.protobuf.GetServerInfoResponse;
import org.yamcs.protobuf.GetServerInfoResponse.CommandOptionInfo;
import org.yamcs.protobuf.Mdb.ParameterInfo;
import org.yamcs.protobuf.ProcessorInfo;
import org.yamcs.protobuf.Pvalue.ParameterValue;
import org.yamcs.protobuf.SubscribeCommandsRequest;
import org.yamcs.protobuf.SubscribeEventsRequest;
import org.yamcs.protobuf.TmPacketData;
// import org.yamcs.protobuf.TmStatistics;
import org.yamcs.protobuf.links.LinkInfo;
import org.yamcs.protobuf.links.SubscribeLinksRequest;
import org.yamcs.utils.TimeEncoding;

public class CMDR_YamcsInstance extends YamcsObject<YamcsObject<?>> {
  public static final Logger logger = Logger.getLogger(CMDR_YamcsInstance.class.getPackageName());
  public static String OBJECT_TYPE = "instance";
  private ProcessorClient yamcsProcessor = null;
  private YamcsSubscriptionService paramSubscriptionService;
  private EventSubscription eventSubscription;
  private CommandSubscription commandsSubscription;
  private LinkSubscription linkSubscription;
  private MissionDatabase missionDatabase;
  //  private EventSubscription eventSubscription;
  private ArchiveClient yamcsArchiveClient;
  private CMDR_YamcsInstanceState instanceState;

  // TODO:Not sure if we want to have this on every instance and their server...just want it to work
  // for now.
  // Useful for "special" command link arguments such as cop1Bypass
  private HashMap<String, CommandOptionInfo> extraCommandArgs =
      new HashMap<String, CommandOptionInfo>();

  private ObservableList<CommandOption> optionsList = FXCollections.observableArrayList();
  private ObservableList<CMDR_Event> events = FXCollections.observableArrayList();
  private ObservableList<CommandHistoryEntry> commands = FXCollections.observableArrayList();
  private HashSet<String> commandsSet = new HashSet<String>();

  public ObservableList<CommandHistoryEntry> getCommands() {
    return commands;
  }

  private ObservableList<LinkInfo> links = FXCollections.observableArrayList();
  private ObservableList<TmStatistics> packets = FXCollections.observableArrayList();

  public static final Logger log = Logger.getLogger(CMDR_YamcsInstance.class.getPackageName());

  public ObservableList<TmStatistics> getPackets() {
    return packets;
  }

  private HashMap<String, LinkInfo> linksMap = new HashMap<String, LinkInfo>();

  public HashMap<String, LinkInfo> getLinksMap() {
    return linksMap;
  }

  private HashMap<String, Boolean> activeInLinks = new HashMap<String, Boolean>();

  private HashMap<String, Instant> LastUpdateLinks = new HashMap<String, Instant>();

  public HashMap<String, Boolean> getActiveInLinks() {
    return activeInLinks;
  }

  private HashMap<String, Boolean> activeOutLinks = new HashMap<String, Boolean>();
  private ScheduledThreadPoolExecutor timer;
  private YamcsWebSocketClient statsWS;

  public ObservableList<LinkInfo> getLinks() {
    return links;
  }

  public CMDR_YamcsInstanceState getInstanceState() {
    return instanceState;
  }

  // Make this class generic?
  public class CommandOption {
    private String id;
    private String value;

    public CommandOption(String newId, String value) {
      this.id = newId;
      this.value = value;
    }

    public String getValue() {
      return this.value;
    }

    public String getId() {
      return this.id;
    }

    public void setValue(String newValue) {
      this.value = newValue;
    }
  }

  public ObservableList<CommandOption> getOptionsList() {
    return optionsList;
  }

  public HashMap<String, CommandOptionInfo> getExtraCommandArgs() {
    return extraCommandArgs;
  }

  public ArchiveClient getYamcsArchiveClient() {
    return yamcsArchiveClient;
  }

  public ObservableList<CMDR_Event> getEvents() {
    return events;
  }

  public ProcessorClient getYamcsProcessor() {
    return yamcsProcessor;
  }

  public CMDR_YamcsInstance(String name) {
    super(name);
  }

  @Override
  public ObservableList<YamcsObject<?>> getItems() {
    return FXCollections.emptyObservableList();
  }

  @Override
  public void createAndAddChild(String name) {
    throw new IllegalStateException("CMDR_YamcsInstance does not allow child items");
  }

  @Override
  public String getObjectType() {
    return OBJECT_TYPE;
  }

  protected void initProcessorClient(YamcsClient yamcsClient) {
    yamcsProcessor = yamcsClient.createProcessorClient(getName(), "realtime");
  }

  protected void initYamcsSubscriptionService(
      YamcsClient yamcsClient, String serverName, String procesor) {
    paramSubscriptionService =
        new YamcsSubscriptionService(
            yamcsClient.createParameterSubscription(), serverName, this.getName(), procesor);
  }

  protected void initLinkSubscription(YamcsClient yamcsClient, String serverName) {
    linkSubscription = yamcsClient.createLinkSubscription();
    linkSubscription.addMessageListener(
        linkEvent -> {
          switch (linkEvent.getType()) {
            case REGISTERED:
            case UPDATED:
              {
                var link = linkEvent.getLinkInfo();
                LinkInfo linkFromList = null;

                LastUpdateLinks.put(link.getName(), Instant.now());

                linksMap.put(link.getName(), link);

                boolean linkExistsInlList = false;

                for (var l : links) {
                  if (l != null) {
                    if (l.getName().equals(link.getName())) {
                      linkFromList = l;
                      linkExistsInlList = true;
                    }
                  }
                }

                if (linkExistsInlList) {
                  links.remove(linkFromList);
                }
                links.add(linksMap.get(link.getName()));
              }

              break;
            case UNREGISTERED:
              //               TODO but not currently sent by Yamcs
          }
        });

    linkSubscription.sendMessage(SubscribeLinksRequest.newBuilder().setInstance(getName()).build());
  }

  protected void initEventSubscription(YamcsClient yamcsClient, String serverName) {
    eventSubscription = yamcsClient.createEventSubscription();
    eventSubscription.addMessageListener(
        event -> {
          events.add(
              new CMDR_Event(
                  event.getMessage(),
                  Instant.ofEpochSecond(
                      event.getGenerationTime().getSeconds(), event.getGenerationTime().getNanos()),
                  event.getSeverity(),
                  event.getType(),
                  Instant.ofEpochSecond(
                      event.getReceptionTime().getSeconds(), event.getReceptionTime().getNanos()),
                  event.getSource(),
                  this.getName()));
        });

    yamcsArchiveClient = yamcsClient.createArchiveClient(getName());

    eventSubscription.sendMessage(
        SubscribeEventsRequest.newBuilder().setInstance(getName()).build());
  }

  protected void initCommandSubscription(YamcsClient yamcsClient) {
    commandsSubscription = yamcsClient.createCommandSubscription();
    commandsSubscription.addMessageListener(
        command -> {
          var commandId = command.getId();
          if (!commandsSet.contains(commandId)) {
            commandsSet.add(commandId);
            commands.add(command);
          }
        });
    var p = getRealtimeProcessor(yamcsClient);
    if (p == null) {
      log.info("Failed to initialize Commands subscription. No realtime processor found.");
      return;
    }
    commandsSubscription.sendMessage(
        SubscribeCommandsRequest.newBuilder()
            .setInstance(getName())
            .setProcessor(p.getName())
            .setIgnorePastCommands(false)
            .build());
  }

  private MissionDatabase loadMissionDatabase(YamcsClient client) {
    var missionDatabase = new MissionDatabase();

    var mdbClient = client.createMissionDatabaseClient(getName());

    try {
      var page = mdbClient.listParameters(ListOptions.limit(500)).get();
      page.iterator().forEachRemaining(missionDatabase::addParameter);
      while (page.hasNextPage()) {
        page = page.getNextPage().get();
        page.iterator().forEachRemaining(missionDatabase::addParameter);
      }

      var commandPage = mdbClient.listCommands(ListOptions.limit(200)).get();
      commandPage.iterator().forEachRemaining(missionDatabase::addCommand);
      while (commandPage.hasNextPage()) {
        commandPage = commandPage.getNextPage().get();
        commandPage.iterator().forEachRemaining(missionDatabase::addCommand);
      }
    } catch (Exception e) {
      e.printStackTrace();
      //          throw new Exception("Failed to load mission database", e);
    }
    return missionDatabase;
  }

  protected void initMDBParameterRDequest(YamcsClient yamcsClient, String serverName) {
    var mdb = yamcsClient.createMissionDatabaseClient(getName()).listParameters();
    Page<ParameterInfo> paramsPage = null;
    try {
      paramsPage = mdb.get();
    } catch (InterruptedException | ExecutionException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    var it = paramsPage.iterator();
    it.forEachRemaining(
        p -> {
          //          System.out.println("p-->" + p.getQualifiedName());

          for (var m : p.getType().getMemberList()) {
            //            System.out.println("p member-->" + m.getName());
          }
        });
    while (paramsPage.hasNextPage()) {
      //      var it = paramsPage.iterator();
      try {
        paramsPage = paramsPage.getNextPage().get();
      } catch (InterruptedException | ExecutionException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
      it = paramsPage.iterator();
      it.forEachRemaining(
          p -> {
            //            System.out.println("p-->" + p.getQualifiedName());

            for (var m : p.getType().getMemberList()) {
              //              System.out.println("p member-->" + m.getName());
            }
          });
    }
  }

  /**
   * Initializes all of the subscriptions to the servers such as event and parameter subscriptions.
   * Always call this AFTER the websocket connection to YAMCS has been established. Ideally inside
   * the connected() method of a org.yamcs.client.ConnectionListener. Otherwise, one might cause a
   * race between the time we "connect" via the websocket and the time we create these
   * subscriptions.
   *
   * @param yamcsClient
   * @param serverName
   */
  // TODO:This shoud return whether or not the instance activated successfully.
  public void activate(YamcsClient yamcsClient, String serverName) {
    initProcessorClient(yamcsClient);
    initYamcsSubscriptionService(yamcsClient, serverName, "realtime");
    initEventSubscription(yamcsClient, serverName);
    initLinkSubscription(yamcsClient, serverName);
    initMDBParameterRDequest(yamcsClient, serverName);
    initTMStats(yamcsClient);
    initCommandSubscription(yamcsClient);

    missionDatabase = loadMissionDatabase(yamcsClient);

    try {
      initCommandOptions(yamcsClient);
    } catch (InterruptedException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      return;
    } catch (ExecutionException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      return;
    }
    instanceState = CMDR_YamcsInstanceState.ACTIVATED;
  }

  public void subscribeTMStats(YamcsClient yamcsClient, Consumer<ProcessingStatistics> consumer) {
    //    TODO:Don't use the YAMCS thread pool. Use the Java one.
    //    timer = YamcsServer.getServer().getThreadPoolExecutor();
    //
    //    //    Make "realtime configurable"
    //
    //    timer.scheduleAtFixedRate(
    //        () -> {
    //          System.out.println("scheduleAtFixedRate1");
    //          YamcsServer.getServer();
    //          System.out.println("scheduleAtFixedRate2:" + YamcsServer.getServer());
    //          System.out.println("Instance:" + getName());
    //          var instance = YamcsServer.getServer().getInstance(getName());
    //
    //          System.out.println("scheduleAtFixedRate3:" + instance);
    //          YamcsServer.getServer().getInstance(getName()).getProcessor("realtime");
    //
    //          System.out.println("scheduleAtFixedRate4");
    //          ProcessingStatistics ps =
    //              YamcsServer.getServer()
    //                  .getInstance(getName())
    //                  .getProcessor("realtime")
    //                  .getTmProcessor()
    //                  .getStatistics();
    //          //           ps =
    //          //              YamcsServer.getServer()
    //          //                  .getInstance(getName())
    //          //                  .getProcessor("realtime")
    //          //                  .getTmProcessor()
    //          //                  .getStatistics();
    //          System.out.println("scheduleAtFixedRate5:" + ps);
    //          consumer.accept(ps);
    //        },
    //        1,
    //        1,
    //        TimeUnit.SECONDS);
    //          TODO:Add the API call to YMACS server side
  }

  public void initTMStats(YamcsClient yamcsClient) {
    var p = getRealtimeProcessor(yamcsClient);
    if (p == null) {
      log.info("Failed to initialize Stats subscription. No realtime processor found.");
      return;
    }
    statsWS =
        new YamcsWebSocketClient(
            stats -> {
              if (stats != null) {
                packets.clear();
                for (TmStatistics s : stats) {
                  packets.add(s);
                }
              }
            },
            yamcsClient.getHost(),
            yamcsClient.getPort(),
            getName(),
            p.getName());
    //
    //    subscribeTMStats(
    //        yamcsClient,
    //        stats -> {
    //          packets.clear();
    //          System.out.println("stats...");
    //          for (var s : stats.snapshot()) {
    //            packets.add(s);
    //          }
    //        });
  }

  private ProcessorInfo getRealtimeProcessor(YamcsClient yamcsClient) {
    List<ProcessorInfo> processors = null;
    ProcessorInfo realtimeP = null;

    try {
      processors = yamcsClient.listProcessors(getName()).get();
    } catch (InterruptedException | ExecutionException e) {
      // TODO Auto-generated catch block
      log.info(e.toString());
      return realtimeP;
    }

    for (var p : processors) {
      if (p.getName().equals("realtime")) {
        realtimeP = p;
        break;
      }
    }
    return realtimeP;
  }

  private void initPacketSubscription(YamcsClient yamcsClient) {
    PacketSubscription subscription = yamcsClient.createPacketSubscription();
    //    yamcsClient.createProcessorClient(OBJECT_TYPE, OBJECT_TYPE)
    subscription.addMessageListener(
        new MessageListener<TmPacketData>() {

          @Override
          public void onMessage(TmPacketData message) {
            TmPacket pwt =
                new TmPacket(
                    TimeEncoding.fromProtobufTimestamp(message.getReceptionTime()),
                    TimeEncoding.fromProtobufTimestamp(message.getGenerationTime()),
                    message.getSequenceNumber(),
                    message.getPacket().toByteArray());
            //            packetsTable.packetReceived(pwt);
          }

          @Override
          public void onError(Throwable t) {
            //            showError("Error subscribing: " + t.getMessage());
          }
        });

    //    subscription.sendMessage(SubscribeTMStatisticsRequest.newBuilder()
    //            .setInstance(getName())
    ////            .setStream(connectData.streamName)
    //            .build());
  }

  public MissionDatabase getMissionDatabase() {
    return missionDatabase;
  }

  private void initCommandOptions(YamcsClient yamcsClient)
      throws InterruptedException, ExecutionException {
    GetServerInfoResponse info = yamcsClient.getServerInfo().get();
    System.out.println("initCommandOptions-->1");
    for (CommandOptionInfo o : info.getCommandOptionsList()) {
      extraCommandArgs.put(o.getId(), o);

      // Eventually check the type and create Commandoption accordingly
      optionsList.add(new CommandOption(o.getId(), ""));
      System.out.println("initCommandOptions-->2" + optionsList);
    }
    System.out.println("initCommandOptions-->3");
  }

  public void deActivate(YamcsClient yamcsClient, String serverName) {
    // TODO:unInit resources...
    instanceState = CMDR_YamcsInstanceState.DEACTIVATED;
    if (eventSubscription != null) {
      eventSubscription.cancel(true);
      paramSubscriptionService.destroy();
    }

    statsWS.close();
  }

  public EventSubscription getEventSubscription() {
    return eventSubscription;
  }

  public void subscribePV(YamcsPV pv) {
    // TODO:Have to let the caller know whether were able to successfully subscribe
    // to this pv or not.
    paramSubscriptionService.register(pv);
  }

  /** Creates and publishes an event to YAMCS instance. */
  public void publishEvent(String message, YamcsClient yamcsClient) {
    yamcsClient.createEvent(
        CreateEventRequest.newBuilder()
            .setInstance(getName())
            .setMessage(message)
            .setSource("Commander")
            .build());
  }

  /** Creates and publishes an event to YAMCS instance. */
  public void publishEvent(CMDR_Event e, YamcsClient yamcsClient) {
    yamcsClient.createEvent(
        CreateEventRequest.newBuilder()
            .setInstance(getName())
            .setMessage(e.getMessage())
            .setSource(e.getSource())
            .setSeverity(e.getSeverity().toString())
            .build());
  }

  public ArrayList<String> getProcessors(YamcsClient yamcsClient) {

    ArrayList<String> processors = new ArrayList<String>();
    try {
      yamcsClient
          .listProcessors(getName())
          .get()
          .forEach(
              p -> {
                processors.add(p.getName());
              });
    } catch (InterruptedException | ExecutionException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }

    return processors;
  }

  public void switchProcessor(YamcsClient yamcsClient, String serverName, String processorName) {
    //          This seems redundant....
    paramSubscriptionService.destroy();
    initYamcsSubscriptionService(yamcsClient, serverName, processorName);
  }

  public void getParameters(
      YamcsClient yamcsClient,
      List<String> parameters,
      Instant start,
      Instant end,
      Consumer<ArrayList<Page<ParameterValue>>> consumer) {

    //    this.getYamcsArchiveClient().streamValues(parameters, consumer, start, end);
    ArrayList<Page<ParameterValue>> pages = new ArrayList<Page<ParameterValue>>();
    for (var p : parameters) {
      try {
        pages.add(this.getYamcsArchiveClient().listValues(p, start, end).get());
      } catch (InterruptedException | ExecutionException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
    }

    consumer.accept(pages);
  }

  public void getParameter(
      YamcsClient yamcsClient,
      String parameter,
      Instant start,
      Instant end,
      Consumer<ArrayList<Page<ParameterValue>>> consumer) {

    //    this.getYamcsArchiveClient().streamValues(parameters, consumer, start, end);
    ArrayList<Page<ParameterValue>> pages = new ArrayList<Page<ParameterValue>>();
    try {
      pages.add(this.getYamcsArchiveClient().listValues(parameter, start, end).get());
    } catch (InterruptedException | ExecutionException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }

    consumer.accept(pages);
  }

  public boolean isLinkActive(String linkName) {
    return Duration.between(Instant.now(), LastUpdateLinks.get(linkName)).toMillis() < 1000;
  }
}

1	package com.windhoverlabs.yamcs.core;
2
3	import com.windhoverlabs.pv.yamcs.YamcsPV;
4	import com.windhoverlabs.pv.yamcs.YamcsSubscriptionService;
5	import com.windhoverlabs.yamcs.core.YamcsWebSocketClient.TmStatistics;
6	import java.time.Duration;
7	import java.time.Instant;
8	import java.util.ArrayList;
9	import java.util.HashMap;
10	import java.util.HashSet;
11	import java.util.List;
12	import java.util.concurrent.ExecutionException;
13	import java.util.concurrent.ScheduledThreadPoolExecutor;
14	import java.util.function.Consumer;
15	import java.util.logging.Logger;
16	import javafx.collections.FXCollections;
17	import javafx.collections.ObservableList;
18	import org.yamcs.TmPacket;
19	import org.yamcs.client.CommandSubscription;
20	import org.yamcs.client.EventSubscription;
21	import org.yamcs.client.LinkSubscription;
22	import org.yamcs.client.MessageListener;
23	import org.yamcs.client.PacketSubscription;
24	import org.yamcs.client.Page;
25	import org.yamcs.client.YamcsClient;
26	import org.yamcs.client.archive.ArchiveClient;
27	import org.yamcs.client.mdb.MissionDatabaseClient.ListOptions;
28	import org.yamcs.client.processor.ProcessorClient;
29	import org.yamcs.mdb.ProcessingStatistics;
30	import org.yamcs.protobuf.Commanding.CommandHistoryEntry;
31	// import org.yamcs.protobuf.Event;
32	import org.yamcs.protobuf.CreateEventRequest;
33	import org.yamcs.protobuf.GetServerInfoResponse;
34	import org.yamcs.protobuf.GetServerInfoResponse.CommandOptionInfo;
35	import org.yamcs.protobuf.Mdb.ParameterInfo;
36	import org.yamcs.protobuf.ProcessorInfo;
37	import org.yamcs.protobuf.Pvalue.ParameterValue;
38	import org.yamcs.protobuf.SubscribeCommandsRequest;
39	import org.yamcs.protobuf.SubscribeEventsRequest;
40	import org.yamcs.protobuf.TmPacketData;
41	// import org.yamcs.protobuf.TmStatistics;
42	import org.yamcs.protobuf.links.LinkInfo;
43	import org.yamcs.protobuf.links.SubscribeLinksRequest;
44	import org.yamcs.utils.TimeEncoding;
45
46	public class CMDR_YamcsInstance extends YamcsObject<YamcsObject<?>> {
47	public static final Logger logger = Logger.getLogger(CMDR_YamcsInstance.class.getPackageName());	1✔
48	public static String OBJECT_TYPE = "instance";	1✔
49	private ProcessorClient yamcsProcessor = null;	1✔
50	private YamcsSubscriptionService paramSubscriptionService;
51	private EventSubscription eventSubscription;
52	private CommandSubscription commandsSubscription;
53	private LinkSubscription linkSubscription;
54	private MissionDatabase missionDatabase;
55	// private EventSubscription eventSubscription;
56	private ArchiveClient yamcsArchiveClient;
57	private CMDR_YamcsInstanceState instanceState;
58
59	// TODO:Not sure if we want to have this on every instance and their server...just want it to work
60	// for now.
61	// Useful for "special" command link arguments such as cop1Bypass
62	private HashMap<String, CommandOptionInfo> extraCommandArgs =	1✔
63	new HashMap<String, CommandOptionInfo>();
64
65	private ObservableList<CommandOption> optionsList = FXCollections.observableArrayList();	1✔
66	private ObservableList<CMDR_Event> events = FXCollections.observableArrayList();	1✔
67	private ObservableList<CommandHistoryEntry> commands = FXCollections.observableArrayList();	1✔
68	private HashSet<String> commandsSet = new HashSet<String>();	1✔
69
70	public ObservableList<CommandHistoryEntry> getCommands() {
71	return commands;	×
72	}
73
74	private ObservableList<LinkInfo> links = FXCollections.observableArrayList();	1✔
75	private ObservableList<TmStatistics> packets = FXCollections.observableArrayList();	1✔
76
77	public static final Logger log = Logger.getLogger(CMDR_YamcsInstance.class.getPackageName());	1✔
78
79	public ObservableList<TmStatistics> getPackets() {
80	return packets;	×
81	}
82
83	private HashMap<String, LinkInfo> linksMap = new HashMap<String, LinkInfo>();	1✔
84
85	public HashMap<String, LinkInfo> getLinksMap() {
86	return linksMap;	×
87	}
88
89	private HashMap<String, Boolean> activeInLinks = new HashMap<String, Boolean>();	1✔
90
91	private HashMap<String, Instant> LastUpdateLinks = new HashMap<String, Instant>();	1✔
92
93	public HashMap<String, Boolean> getActiveInLinks() {
94	return activeInLinks;	×
95	}
96
97	private HashMap<String, Boolean> activeOutLinks = new HashMap<String, Boolean>();	1✔
98	private ScheduledThreadPoolExecutor timer;
99	private YamcsWebSocketClient statsWS;
100
101	public ObservableList<LinkInfo> getLinks() {
102	return links;	×
103	}
104
105	public CMDR_YamcsInstanceState getInstanceState() {
106	return instanceState;	×
107	}
108
109	// Make this class generic?
110	public class CommandOption {
111	private String id;
112	private String value;
113
114	public CommandOption(String newId, String value) {	×
115	this.id = newId;	×
116	this.value = value;	×
117	}	×
118
119	public String getValue() {
120	return this.value;	×
121	}
122
123	public String getId() {
124	return this.id;	×
125	}
126
127	public void setValue(String newValue) {
128	this.value = newValue;	×
129	}	×
130	}
131
132	public ObservableList<CommandOption> getOptionsList() {
133	return optionsList;	×
134	}
135
136	public HashMap<String, CommandOptionInfo> getExtraCommandArgs() {
137	return extraCommandArgs;	×
138	}
139
140	public ArchiveClient getYamcsArchiveClient() {
141	return yamcsArchiveClient;	1✔
142	}
143
144	public ObservableList<CMDR_Event> getEvents() {
145	return events;	1✔
146	}
147
148	public ProcessorClient getYamcsProcessor() {
149	return yamcsProcessor;	1✔
150	}
151
152	public CMDR_YamcsInstance(String name) {
153	super(name);	1✔
154	}	1✔
155
156	@Override
157	public ObservableList<YamcsObject<?>> getItems() {
158	return FXCollections.emptyObservableList();	1✔
159	}
160
161	@Override
162	public void createAndAddChild(String name) {
163	throw new IllegalStateException("CMDR_YamcsInstance does not allow child items");	1✔
164	}
165
166	@Override
167	public String getObjectType() {
168	return OBJECT_TYPE;	1✔
169	}
170
171	protected void initProcessorClient(YamcsClient yamcsClient) {
172	yamcsProcessor = yamcsClient.createProcessorClient(getName(), "realtime");	1✔
173	}	1✔
174
175	protected void initYamcsSubscriptionService(
176	YamcsClient yamcsClient, String serverName, String procesor) {
177	paramSubscriptionService =	1✔
178	new YamcsSubscriptionService(
179	yamcsClient.createParameterSubscription(), serverName, this.getName(), procesor);	1✔
180	}	1✔
181
182	protected void initLinkSubscription(YamcsClient yamcsClient, String serverName) {
183	linkSubscription = yamcsClient.createLinkSubscription();	1✔
184	linkSubscription.addMessageListener(	1✔
185	linkEvent -> {
186	switch (linkEvent.getType()) {	1✔
187	case REGISTERED:
188	case UPDATED:
189	{
190	var link = linkEvent.getLinkInfo();	1✔
191	LinkInfo linkFromList = null;	1✔
192
193	LastUpdateLinks.put(link.getName(), Instant.now());	1✔
194
195	linksMap.put(link.getName(), link);	1✔
196
197	boolean linkExistsInlList = false;	1✔
198
199	for (var l : links) {	1✔
200	if (l != null) {	1✔
201	if (l.getName().equals(link.getName())) {	1✔
202	linkFromList = l;	×
203	linkExistsInlList = true;	×
204	}
205	}
206	}	1✔
207
208	if (linkExistsInlList) {	1✔
209	links.remove(linkFromList);	×
210	}
211	links.add(linksMap.get(link.getName()));	1✔
212	}
213
214	break;	1✔
215	case UNREGISTERED:
216	// TODO but not currently sent by Yamcs
217	}
218	});	1✔
219
220	linkSubscription.sendMessage(SubscribeLinksRequest.newBuilder().setInstance(getName()).build());	1✔
221	}	1✔
222
223	protected void initEventSubscription(YamcsClient yamcsClient, String serverName) {
224	eventSubscription = yamcsClient.createEventSubscription();	1✔
225	eventSubscription.addMessageListener(	1✔
226	event -> {
227	events.add(	×
228	new CMDR_Event(
229	event.getMessage(),	×
230	Instant.ofEpochSecond(	×
231	event.getGenerationTime().getSeconds(), event.getGenerationTime().getNanos()),	×
232	event.getSeverity(),	×
233	event.getType(),	×
234	Instant.ofEpochSecond(	×
235	event.getReceptionTime().getSeconds(), event.getReceptionTime().getNanos()),	×
236	event.getSource(),	×
237	this.getName()));	×
238	});	×
239
240	yamcsArchiveClient = yamcsClient.createArchiveClient(getName());	1✔
241
242	eventSubscription.sendMessage(	1✔
243	SubscribeEventsRequest.newBuilder().setInstance(getName()).build());	1✔
244	}	1✔
245
246	protected void initCommandSubscription(YamcsClient yamcsClient) {
247	commandsSubscription = yamcsClient.createCommandSubscription();	1✔
248	commandsSubscription.addMessageListener(	1✔
249	command -> {
250	var commandId = command.getId();	×
251	if (!commandsSet.contains(commandId)) {	×
252	commandsSet.add(commandId);	×
253	commands.add(command);	×
254	}
255	});	×
256	var p = getRealtimeProcessor(yamcsClient);	1✔
257	if (p == null) {	1✔
258	log.info("Failed to initialize Commands subscription. No realtime processor found.");	×
259	return;	×
260	}
261	commandsSubscription.sendMessage(	1✔
262	SubscribeCommandsRequest.newBuilder()	1✔
263	.setInstance(getName())	1✔
264	.setProcessor(p.getName())	1✔
265	.setIgnorePastCommands(false)	1✔
266	.build());	1✔
267	}	1✔
268
269	private MissionDatabase loadMissionDatabase(YamcsClient client) {
270	var missionDatabase = new MissionDatabase();	1✔
271
272	var mdbClient = client.createMissionDatabaseClient(getName());	1✔
273
274	try {
275	var page = mdbClient.listParameters(ListOptions.limit(500)).get();	1✔
276	page.iterator().forEachRemaining(missionDatabase::addParameter);	1✔
277	while (page.hasNextPage()) {	1✔
278	page = page.getNextPage().get();	×
279	page.iterator().forEachRemaining(missionDatabase::addParameter);	×
280	}
281
282	var commandPage = mdbClient.listCommands(ListOptions.limit(200)).get();	1✔
283	commandPage.iterator().forEachRemaining(missionDatabase::addCommand);	1✔
284	while (commandPage.hasNextPage()) {	1✔
285	commandPage = commandPage.getNextPage().get();	1✔
286	commandPage.iterator().forEachRemaining(missionDatabase::addCommand);	1✔
287	}
288	} catch (Exception e) {	×
289	e.printStackTrace();	×
290	// throw new Exception("Failed to load mission database", e);
291	}	1✔
292	return missionDatabase;	1✔
293	}
294
295	protected void initMDBParameterRDequest(YamcsClient yamcsClient, String serverName) {
296	var mdb = yamcsClient.createMissionDatabaseClient(getName()).listParameters();	1✔
297	Page<ParameterInfo> paramsPage = null;	1✔
298	try {
299	paramsPage = mdb.get();	1✔
300	} catch (InterruptedException \| ExecutionException e) {	×
301	// TODO Auto-generated catch block
302	e.printStackTrace();	×
303	}	1✔
304	var it = paramsPage.iterator();	1✔
305	it.forEachRemaining(	1✔
306	p -> {
307	// System.out.println("p-->" + p.getQualifiedName());
308
309	for (var m : p.getType().getMemberList()) {	1✔
310	// System.out.println("p member-->" + m.getName());
311	}	1✔
312	});	1✔
313	while (paramsPage.hasNextPage()) {	1✔
314	// var it = paramsPage.iterator();
315	try {
316	paramsPage = paramsPage.getNextPage().get();	×
317	} catch (InterruptedException \| ExecutionException e) {	×
318	// TODO Auto-generated catch block
319	e.printStackTrace();	×
320	}	×
321	it = paramsPage.iterator();	×
322	it.forEachRemaining(	×
323	p -> {
324	// System.out.println("p-->" + p.getQualifiedName());
325
326	for (var m : p.getType().getMemberList()) {	×
327	// System.out.println("p member-->" + m.getName());
328	}	×
329	});	×
330	}
331	}	1✔
332
333	/**
334	* Initializes all of the subscriptions to the servers such as event and parameter subscriptions.
335	* Always call this AFTER the websocket connection to YAMCS has been established. Ideally inside
336	* the connected() method of a org.yamcs.client.ConnectionListener. Otherwise, one might cause a
337	* race between the time we "connect" via the websocket and the time we create these
338	* subscriptions.
339	*
340	* @param yamcsClient
341	* @param serverName
342	*/
343	// TODO:This shoud return whether or not the instance activated successfully.
344	public void activate(YamcsClient yamcsClient, String serverName) {
345	initProcessorClient(yamcsClient);	1✔
346	initYamcsSubscriptionService(yamcsClient, serverName, "realtime");	1✔
347	initEventSubscription(yamcsClient, serverName);	1✔
348	initLinkSubscription(yamcsClient, serverName);	1✔
349	initMDBParameterRDequest(yamcsClient, serverName);	1✔
350	initTMStats(yamcsClient);	1✔
351	initCommandSubscription(yamcsClient);	1✔
352
353	missionDatabase = loadMissionDatabase(yamcsClient);	1✔
354
355	try {
356	initCommandOptions(yamcsClient);	1✔
357	} catch (InterruptedException e) {	×
358	// TODO Auto-generated catch block
359	e.printStackTrace();	×
360	return;	×
361	} catch (ExecutionException e) {	×
362	// TODO Auto-generated catch block
363	e.printStackTrace();	×
364	return;	×
365	}	1✔
366	instanceState = CMDR_YamcsInstanceState.ACTIVATED;	1✔
367	}	1✔
368
369	public void subscribeTMStats(YamcsClient yamcsClient, Consumer<ProcessingStatistics> consumer) {
370	// TODO:Don't use the YAMCS thread pool. Use the Java one.
371	// timer = YamcsServer.getServer().getThreadPoolExecutor();
372	//
373	// // Make "realtime configurable"
374	//
375	// timer.scheduleAtFixedRate(
376	// () -> {
377	// System.out.println("scheduleAtFixedRate1");
378	// YamcsServer.getServer();
379	// System.out.println("scheduleAtFixedRate2:" + YamcsServer.getServer());
380	// System.out.println("Instance:" + getName());
381	// var instance = YamcsServer.getServer().getInstance(getName());
382	//
383	// System.out.println("scheduleAtFixedRate3:" + instance);
384	// YamcsServer.getServer().getInstance(getName()).getProcessor("realtime");
385	//
386	// System.out.println("scheduleAtFixedRate4");
387	// ProcessingStatistics ps =
388	// YamcsServer.getServer()
389	// .getInstance(getName())
390	// .getProcessor("realtime")
391	// .getTmProcessor()
392	// .getStatistics();
393	// // ps =
394	// // YamcsServer.getServer()
395	// // .getInstance(getName())
396	// // .getProcessor("realtime")
397	// // .getTmProcessor()
398	// // .getStatistics();
399	// System.out.println("scheduleAtFixedRate5:" + ps);
400	// consumer.accept(ps);
401	// },
402	// 1,
403	// 1,
404	// TimeUnit.SECONDS);
405	// TODO:Add the API call to YMACS server side
406	}	×
407
408	public void initTMStats(YamcsClient yamcsClient) {
409	var p = getRealtimeProcessor(yamcsClient);	1✔
410	if (p == null) {	1✔
411	log.info("Failed to initialize Stats subscription. No realtime processor found.");	×
412	return;	×
413	}
414	statsWS =	1✔
415	new YamcsWebSocketClient(
416	stats -> {
417	if (stats != null) {	1✔
418	packets.clear();	×
419	for (TmStatistics s : stats) {	×
420	packets.add(s);	×
421	}	×
422	}
423	},	1✔
424	yamcsClient.getHost(),	1✔
425	yamcsClient.getPort(),	1✔
426	getName(),	1✔
427	p.getName());	1✔
428	//
429	// subscribeTMStats(
430	// yamcsClient,
431	// stats -> {
432	// packets.clear();
433	// System.out.println("stats...");
434	// for (var s : stats.snapshot()) {
435	// packets.add(s);
436	// }
437	// });
438	}	1✔
439
440	private ProcessorInfo getRealtimeProcessor(YamcsClient yamcsClient) {
441	List<ProcessorInfo> processors = null;	1✔
442	ProcessorInfo realtimeP = null;	1✔
443
444	try {
445	processors = yamcsClient.listProcessors(getName()).get();	1✔
446	} catch (InterruptedException \| ExecutionException e) {	×
447	// TODO Auto-generated catch block
448	log.info(e.toString());	×
449	return realtimeP;	×
450	}	1✔
451
452	for (var p : processors) {	1✔
453	if (p.getName().equals("realtime")) {	1✔
454	realtimeP = p;	1✔
455	break;	1✔
456	}
457	}	×
458	return realtimeP;	1✔
459	}
460
461	private void initPacketSubscription(YamcsClient yamcsClient) {
462	PacketSubscription subscription = yamcsClient.createPacketSubscription();	×
463	// yamcsClient.createProcessorClient(OBJECT_TYPE, OBJECT_TYPE)
464	subscription.addMessageListener(	×
465	new MessageListener<TmPacketData>() {	×
466
467	@Override
468	public void onMessage(TmPacketData message) {
469	TmPacket pwt =	×
470	new TmPacket(
471	TimeEncoding.fromProtobufTimestamp(message.getReceptionTime()),	×
472	TimeEncoding.fromProtobufTimestamp(message.getGenerationTime()),	×
473	message.getSequenceNumber(),	×
474	message.getPacket().toByteArray());	×
475	// packetsTable.packetReceived(pwt);
476	}	×
477
478	@Override
479	public void onError(Throwable t) {
480	// showError("Error subscribing: " + t.getMessage());
481	}	×
482	});
483
484	// subscription.sendMessage(SubscribeTMStatisticsRequest.newBuilder()
485	// .setInstance(getName())
486	//// .setStream(connectData.streamName)
487	// .build());
488	}	×
489
490	public MissionDatabase getMissionDatabase() {
491	return missionDatabase;	×
492	}
493
494	private void initCommandOptions(YamcsClient yamcsClient)
495	throws InterruptedException, ExecutionException {
496	GetServerInfoResponse info = yamcsClient.getServerInfo().get();	1✔
497	System.out.println("initCommandOptions-->1");	1✔
498	for (CommandOptionInfo o : info.getCommandOptionsList()) {	1✔
499	extraCommandArgs.put(o.getId(), o);	×
500
501	// Eventually check the type and create Commandoption accordingly
502	optionsList.add(new CommandOption(o.getId(), ""));	×
503	System.out.println("initCommandOptions-->2" + optionsList);	×
504	}	×
505	System.out.println("initCommandOptions-->3");	1✔
506	}	1✔
507
508	public void deActivate(YamcsClient yamcsClient, String serverName) {
509	// TODO:unInit resources...
510	instanceState = CMDR_YamcsInstanceState.DEACTIVATED;	1✔
511	if (eventSubscription != null) {	1✔
512	eventSubscription.cancel(true);	1✔
513	paramSubscriptionService.destroy();	1✔
514	}
515
516	statsWS.close();	1✔
517	}	1✔
518
519	public EventSubscription getEventSubscription() {
520	return eventSubscription;	1✔
521	}
522
523	public void subscribePV(YamcsPV pv) {
524	// TODO:Have to let the caller know whether were able to successfully subscribe
525	// to this pv or not.
526	paramSubscriptionService.register(pv);	×
527	}	×
528
529	/** Creates and publishes an event to YAMCS instance. */
530	public void publishEvent(String message, YamcsClient yamcsClient) {
531	yamcsClient.createEvent(	×
532	CreateEventRequest.newBuilder()	×
533	.setInstance(getName())	×
534	.setMessage(message)	×
535	.setSource("Commander")	×
536	.build());	×
537	}	×
538
539	/** Creates and publishes an event to YAMCS instance. */
540	public void publishEvent(CMDR_Event e, YamcsClient yamcsClient) {
541	yamcsClient.createEvent(	×
542	CreateEventRequest.newBuilder()	×
543	.setInstance(getName())	×
544	.setMessage(e.getMessage())	×
545	.setSource(e.getSource())	×
546	.setSeverity(e.getSeverity().toString())	×
547	.build());	×
548	}	×
549
550	public ArrayList<String> getProcessors(YamcsClient yamcsClient) {
551
552	ArrayList<String> processors = new ArrayList<String>();	×
553	try {
554	yamcsClient	×
555	.listProcessors(getName())	×
556	.get()	×
557	.forEach(	×
558	p -> {
559	processors.add(p.getName());	×
560	});	×
561	} catch (InterruptedException \| ExecutionException e) {	×
562	// TODO Auto-generated catch block
563	e.printStackTrace();	×
564	}	×
565
566	return processors;	×
567	}
568
569	public void switchProcessor(YamcsClient yamcsClient, String serverName, String processorName) {
570	// This seems redundant....
571	paramSubscriptionService.destroy();	×
572	initYamcsSubscriptionService(yamcsClient, serverName, processorName);	×
573	}	×
574
575	public void getParameters(
576	YamcsClient yamcsClient,
577	List<String> parameters,
578	Instant start,
579	Instant end,
580	Consumer<ArrayList<Page<ParameterValue>>> consumer) {
581
582	// this.getYamcsArchiveClient().streamValues(parameters, consumer, start, end);
583	ArrayList<Page<ParameterValue>> pages = new ArrayList<Page<ParameterValue>>();	×
584	for (var p : parameters) {	×
585	try {
586	pages.add(this.getYamcsArchiveClient().listValues(p, start, end).get());	×
587	} catch (InterruptedException \| ExecutionException e) {	×
588	// TODO Auto-generated catch block
589	e.printStackTrace();	×
590	}	×
591	}	×
592
593	consumer.accept(pages);	×
594	}	×
595
596	public void getParameter(
597	YamcsClient yamcsClient,
598	String parameter,
599	Instant start,
600	Instant end,
601	Consumer<ArrayList<Page<ParameterValue>>> consumer) {
602
603	// this.getYamcsArchiveClient().streamValues(parameters, consumer, start, end);
604	ArrayList<Page<ParameterValue>> pages = new ArrayList<Page<ParameterValue>>();	×
605	try {
606	pages.add(this.getYamcsArchiveClient().listValues(parameter, start, end).get());	×
607	} catch (InterruptedException \| ExecutionException e) {	×
608	// TODO Auto-generated catch block
609	e.printStackTrace();	×
610	}	×
611
612	consumer.accept(pages);	×
613	}	×
614
615	public boolean isLinkActive(String linkName) {
616	return Duration.between(Instant.now(), LastUpdateLinks.get(linkName)).toMillis() < 1000;	×
617	}
618	}

WindhoverLabs / phoebus / #94

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