原文：http://blog.mygraphql.com/wordpress/?p=106
订阅（Subscriptions）
订阅查询（Subscription Queries）
Graphql 订阅（subscriptions）使你可以让你订阅响应式数据源（reactive
source） 。当有新数据时，会发送给订阅者。
可以阅读 http://graphql.org/blog/subscriptions-in-graphql-and-relay/
来了解订阅的背景知识。
假设你有一个股票服务。可以用这个 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-java/graphql-java-subscription-example
关于订阅服务的 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
查询一样。
