Блог

Мне нужно использовать сообщения из Websocket, но я должен выполнить некоторые логические действия перед использованием данных, поэтому я не могу использовать компонент Webscoket.

У меня есть Java-код, который выполняет аутентификацию в этом Websocket и подписывается на «Датчик» для получения данных.

Могу ли я создать компонент Camel, в котором я использую этот код, from() и каждый раз, когда я получаю новые данные onNext() , Camel запускает процесс?

 WebSocket webSocket = new WebSocket(uri, apiKey, (api, authenthication) -> {
      console.println("Authenticated successfully as "   authenthication.getUserName());
      String[] sensors = {sensorId};
      api.getMetrics(sensors).subscribe(metrics -> {
        Metric[] allMetrics = metrics.get(sensorId);
        Arrays.sort(allMetrics, (metric1, metric2) -> metric1.getId().compareTo(metric2.getId()));
        Metric firstMetric = allMetrics[0];
        console.println("Metric: "   firstMetric.getDisplayName());
        String metricId = firstMetric.getId();
        String[] metric = {metricId};
        api.getUnits(metric).subscribe(units -> {
          Unit unit = units.get(metric[0])[0];
          console.println("Unit: "   unit.getName());
          Instant now = Instant.now();
          Instant aMinuteAgo = now.minus(timeInterval, ChronoUnit.SECONDS);
          Date start = Date.from(aMinuteAgo);
          Date end = Date.from(now);
          api.getData(sensorId, metricId, unit.getId(), emptyMap(), start, end).subscribe(new DisposableObserver<Data>() {

            @Override
            public void onNext(Data data) {
              console.println("Data from last "   timeInterval   " seconds: ");
              console.println(data.getData());
            }

            @Override
            public void onComplete() {
              console.println("Data update:");
              Disposable subscription = api.subscribeData(sensors, metricId, unit.getId()).subscribe(updates -> {
                console.println(updates.getData());
              });

              ScheduledExecutorService scheduler = newSingleThreadScheduledExecutor(daemonThreadFactory);
              scheduler.schedule(subscription::dispose, cancelDelay, SECONDS);
            }

            @Override
            public void onError(Throwable error) {
              error.printStackTrace();
            }
          });
        });
      });
    });
    console.println("Connection was closed by server.");
  }