Apache Camel - создать пользовательский компонент / конечную точку?

Question

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

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

Могу ли я создать компонент 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.");
  }