原文:http://blog.mygraphql.com/wordpress/?p=106
订阅(Subscriptions)
订阅查询(Subscription Queries)
Graphql 订阅(subscriptions)使你可以让你订阅响应式数据源(reactive
source) 。当有新数据时,会发送给订阅者。
可以阅读 http://graphql.org/blog/subsc...
来了解订阅的背景知识。
假设你有一个股票服务。可以用这个 graphql 语句来订阅它的数据:
subscription StockCodeSubscription {
stockQuotes(stockCode:"IBM') {
dateTime
stockCode
stockPrice
stockPriceChange
}
}
股票价格变化时,graphql 订阅 可以把 ExecutionResult
对象以流的方式传送给订阅者。和其它 graphql 查询一样,只会发送指定的字段
。
不同的是,一开始的查询结果是一个响应式流(reactive-streams)Publisher(流发布者)
对象。通过对象可以获取未来的数据。
你需要使用 SubscriptionExecutionStrategy
策略作为执行策略(execution
strategy)。因为它支持 reactive-streams APIs.
GraphQL graphQL = GraphQL
.newGraphQL(schema)
.subscriptionExecutionStrategy(new SubscriptionExecutionStrategy())
.build();
ExecutionResult executionResult = graphQL.execute(query);
Publisher<ExecutionResult> stockPriceStream = executionResult.getData();
这里的 Publisher<ExecutionResult>
就是流事件的发布者【译注:原文
publisher of a stream of events】。你需要编写你自己的流处理代码,如:
GraphQL graphQL = GraphQL
.newGraphQL(schema)
.subscriptionExecutionStrategy(new SubscriptionExecutionStrategy())
.build();
String query = "" +
" subscription StockCodeSubscription {\n" +
" stockQuotes(stockCode:\"IBM') {\n" +
" dateTime\n" +
" stockCode\n" +
" stockPrice\n" +
" stockPriceChange\n" +
" }\n" +
" }\n";
ExecutionResult executionResult = graphQL.execute(query);
Publisher<ExecutionResult> stockPriceStream = executionResult.getData();
AtomicReference<Subscription> subscriptionRef = new AtomicReference<>();
stockPriceStream.subscribe(new Subscriber<ExecutionResult>() {
@Override
public void onSubscribe(Subscription s) {
subscriptionRef.set(s);
s.request(1);
}
@Override
public void onNext(ExecutionResult er) {
//
// process the next stock price
//
processStockPriceChange(er.getData());
//
// ask the publisher for one more item please
//
subscriptionRef.get().request(1);
}
@Override
public void onError(Throwable t) {
//
// The upstream publishing data source has encountered an error
// and the subscription is now terminated. Real production code needs
// to decide on a error handling strategy.
//
}
@Override
public void onComplete() {
//
// the subscription has completed. There is not more data
//
}
});
需要编写 reactive-streams 代码去消费一源源不断的ExecutionResults
。你可以在 http://www.reactive-streams.org/ 中看到更
reactive-streams 代码的编写细节。
>><<RxJava
是这个流行的 reactive-streams 实现。在
http://reactivex.io/intro.html 中可以看到更多创建Publishers 数据 和
Subscriptions 数据的细节。
graphql-java 只是产出一个流对象。它不关心如何在网络上用 web sockets
或其它手段发送流数据 。虽然这很重要,但不是作为基础 graphql-java
库应该做的。
我们编写了一个 websockets 的(基于 Jetty)
模拟股票报价的示例应用。它使用了 RxJava。
详见 https://github.com/graphql-ja...
关于订阅服务的 Data Fetchers
订阅字段的 DataFetcher
的职责是生成一个 Publisher
。这个 Publisher
输出的每一个对象,将会通过 graphql 查询来映射。然后作为执行结果返回。
你会像这样子去编写Data Fetcher:
DataFetcher<Publisher<StockInfo>> publisherDataFetcher = new DataFetcher<Publisher<StockInfo>>() {
@Override
public Publisher<StockInfo> get(DataFetchingEnvironment environment) {
String stockCodeArg = environment.getArgument("stockCode");
return buildPublisherForStockCode(stockCodeArg);
}
};
如何获取流事件,就由你的 reactive code 来决定 了。graphql-java
会帮助你从流对象中获取 graphql 字段(fields)。像一般的 graphql
查询一样。