Vert.x - 带有 DataInputStreams 的 GraphQL 订阅

Posted 2023-03-08

【中文标题】Vert.x - 带有 DataInputStreams 的 GraphQL 订阅

【英文标题】：Vert.x - GraphQL Subscriptions with DataInputStreams

【发布时间】：2020-02-12 09:58:00

【问题描述】：

我有通过 DataInputStream 连接到的第 3 方代码。第 3 方代码在生成信息时不断吐出信息。当遇到感兴趣的事情时，我想将其传递给 GraphQL 订阅

在这种情况下，我不确定如何将第 3 方代码连接到服务器端 GraphQL 订阅代码。任何建议，将不胜感激。

一些概念代码如下：

public void liveStream(DataInputStream in) 
  // Sit and constantly watch input stream and report when messages come in
  while(true) 
    SomeMessage message = readFromInputStream(in);
    System.out.println("Received Message Type:" + message.getType());

    // Convert SomeMessage into the appropriate class based on its type
    if (message.getType() == "foo") 
      Foo foo = convertMessageToFoo(message);
     else if (message.getType() == "bar") 
      Bar bar = convertMessageToBar(message);
     else if (howeverManyMoreOfThese) 
      // Keep converting to different objects
    
         


// The client code will eventually trigger this method when 
// the GraphQL Subscription query is sent over
VertxDataFetcher<Publisher<SomeClassTBD>> myTestDataFetcher() 
  return new VertxDataFetcher<> (env, future) -> 
    try 
      future.complete(myTest());
     catch(Exception e) 
      future.fail(e);
    
  );

【问题讨论】：

这里是一个订阅github.com/vert-x3/vertx-examples/tree/master/…HTH的Vert.x GraphQL服务器示例

【参考方案1】：

好的，我使用 executorService 将我的 liveStream 代码包装在 ObservableOnSubscribe 中，并且我正在取回所有数据。我想我现在可以将它直接传递到前端，或者创建单独的发布者来处理特定的对象类型，并让 graphql 订阅指向它们各自的发布者。

ExecutorService executor = Executors.newSingleThreadExecutor;

ObservableOnSubscribe<SomeClassTBD> handler = emitter ->
  executor.submit(() -> 
    try 
      //liveStream code here
      emitter.onComplete();
    
    catch(Exception e) 
      emitter.onError(e);
    
    finally 
      // Cleanup here
    
  );
  Observable<SomeClassTBD> = Observable.create(handler);