使用RxNetty访问Meetup的流API

本文将涉及多个主题：响应式编程，HTTP，解析JSON以及与社交API集成。完全在一个用例中：我们将通过非夸张的RxNetty库实时加载和处理新的metup.com事件，结合Netty框架的强大功能和RxJava库的灵活性。 Meetup提供了公开可用的流API ，可实时推送世界各地注册的每一个Meetup。只需浏览至stream.meetup.com/2/open_events并观察JSON块如何缓慢地出现在屏幕上。每当有人创建新事件时，自包含的JSON就会从服务器推送到您的浏览器。这意味着这样的请求永无止境，相反，只要需要，我们就会不断接收部分数据。我们已经在将Twitter4J变成RxJava的Observable中研究了类似的情况。每个新的Meetup事件都会发布一个独立的JSON文档，与此类似（省略很多细节）：

{ "id" : "219088449","name" : "Silver Wings Brunch","time" : 1421609400000,"mtime" : 1417814004321,"duration" : 900000,"rsvp_limit" : 0,"status" : "upcoming","event_url" : "http://www.meetup.com/Laguna-Niguel-Social-Networking-Meetup/events/219088449/","group" : { "name" : "Former Flight Attendants South Orange and North San Diego Co","state" : "CA"...},"venue" : { "address_1" : "26860 Ortega Highway","city" : "San Juan Capistrano","country" : "US"...},"venue_visibility" : "public","visibility" : "public","yes_rsvp_count" : 1...
}

每当我们长时间轮询的HTTP连接（带有Transfer-Encoding: chunked响应标头）推送此类JSON时，我们都希望对其进行解析并以某种方式进一步传递。我们讨厌回调，因此RxJava似乎是一个合理的选择（认为： Observable<Event> ）。

步骤1：使用RxNetty接收原始数据

我们不能使用普通的HTTP客户端，因为它们专注于请求-响应语义。这里没有任何响应，我们只是永远保持打开的连接，并在数据到达时使用它们。 RxJava具有开箱即用的RxApacheHttp库，但它假定为text/event-stream内容类型。相反，我们将使用底层的通用RxNetty库。它是Netty（duh！）的包装，并且能够实现任意的 TCP / IP（包括HTTP）以及UDP客户端和服务器。如果您不了解Netty，则它是基于数据包的，而不是面向流的，因此我们可以预期每次Meetup推送都会有一个Netty事件。该API当然不是简单明了的，但是一旦您使用它，它就会变得有意义：

HttpClient<ByteBuf, ByteBuf> httpClient = RxNetty.<ByteBuf, ByteBuf>newHttpClientBuilder("stream.meetup.com", 443).pipelineConfigurator(new HttpClientPipelineConfigurator<>()).withSslEngineFactory(DefaultFactories.trustAll()).build();final Observable<HttpClientResponse> responses = httpClient.submit(HttpClientRequest.createGet("/2/open_events"));
final Observable byteBufs = responses.flatMap(AbstractHttpContentHolder::getContent);
final Observable chunks = byteBufs.map(content -> content.toString(StandardCharsets.UTF_8));

首先，我们创建HttpClient并设置SSL（请注意，关于服务器证书的trustAll()可能不是最佳的生产设置）。稍后，我们submit() GET请求并接收Observable<HttpClientResponse<ByteBuf>>作为回报。 ByteBuf是Netty对通过网络发送或接收的一堆字节的抽象。该观察结果将立即告诉我们从Meetup收到的每条数据。从响应中提取ByteBuf ，我们将其转换为包含上述JSON的String 。到目前为止，一切正常。

步骤2：将数据包与JSON文档对齐

Netty非常强大，因为它不会掩盖泄漏抽象所固有的复杂性。每次通过TCP / IP线路接收到某些内容时，都会通知我们。您可能会相信，当服务器发送100字节时，客户端的Netty会通知我们有关这100字节的信息。但是，TCP / IP堆栈可以自由地拆分和合并您通过有线发送的数据，特别是因为它假定是流，因此如何将其拆分为数据包无关紧要。 Netty的文档中对此警告做了很大的解释。对我们意味着什么？当Meetup发送单个事件时，我们可能仅收到一个可观察到的chunks String 。但是同样可以将其划分为任意数量的数据包，因此chunks将发出多个String 。更糟糕的是，如果Meetup接连发送两个事件，则它们可能适合一个数据包。在这种情况下， chunks将发出一个带有两个独立JSON文档的String 。事实上，我们不能假设JSON字符串和收到的网络数据包之间有任何对齐。我们所知道的是，代表事件的各个JSON文档由换行符分隔。令人惊讶的是， RxJavaString官方附加组件RxJavaString提供了一种精确的方法：

Observable jsonChunks = StringObservable.split(chunks, "\n");

实际上，甚至还有更简单的StringObservable.byLine(chunks) ，但它使用的是平台相关的行尾。最好在官方文档中解释split()作用：

圣分裂

现在我们可以安全地解析jsonChunks发出的每个String了：

步骤3：解析JSON

有趣的是，这一步骤并不是那么简单。我承认，我排序的享受WSDL时间，因为我很容易，可预见生成如下web服务的合同Java模型。 JSON，特别是在JSON模式的边缘市场渗透方面，基本上是集成的“狂野西部”。通常，您会得到非正式的文档或请求和响应的样本。没有类型信息或格式，无论字段是否为必填项，等等。此外，由于我不情愿使用地图映射 （在那里，Clojure程序员），为了使用基于JSON的REST服务，我必须自己编写映射POJO。好吧，有解决方法。首先，我举了一个由Meetup流API生成的JSON的代表性示例，并将其放在src/main/json/meetup/event.json 。然后，我使用jsonschema2pojo-maven-plugin （也存在Gradle和Ant版本）。插件的名称令人困惑，它还可以与JSON示例（不仅是架构）一起使用以生成Java模型：

<plugin><groupId>org.jsonschema2pojo</groupId><artifactId>jsonschema2pojo-maven-plugin</artifactId><version>0.4.7</version><configuration><sourceDirectory>${basedir}/src/main/json/meetup</sourceDirectory><targetPackage>com.nurkiewicz.meetup.generated</targetPackage><includeHashcodeAndEquals>true</includeHashcodeAndEquals><includeToString>true</includeToString><initializeCollections>true</initializeCollections><sourceType>JSON</sourceType><useCommonsLang3>true</useCommonsLang3><useJodaDates>true</useJodaDates><useLongIntegers>true</useLongIntegers><outputDirectory>target/generated-sources</outputDirectory></configuration><executions><execution><id>generate-sources</id><phase>generate-sources</phase><goals><goal>generate</goal></goals></execution></executions>
</plugin>

此时，Maven将创建与Jackson兼容的Event.java ， Venue.java ， Group.java等：

private Event parseEventJson(String jsonStr) {try {return objectMapper.readValue(jsonStr, Event.class);} catch (IOException e) {throw new UncheckedIOException(e);}
}

很好，它很好：

final Observableevents = jsonChunks.map(this::parseEventJson);

步骤5：获利！！！

有了Observable<Event>我们可以实现一些非常有趣的用例。是否要查找刚刚创建的波兰所有聚会的名称？当然！

events.filter(event -> event.getVenue() != null).filter(event -> event.getVenue().getCountry().equals("pl")).map(Event::getName).forEach(System.out::println);

寻找统计信息每分钟创建多少个事件？没问题！

events.buffer(1, TimeUnit.MINUTES).map(List::size).forEach(count -> log.info("Count: {}", count));

或者，也许您想继续搜索将来最远的聚会，而不是寻找比已发现的聚会更近的聚会？

events.filter(event -> event.getTime() != null).scan(this::laterEventFrom).distinct().map(Event::getTime).map(Instant::ofEpochMilli).forEach(System.out::println);//...private Event laterEventFrom(Event first, Event second) {return first.getTime() > second.getTime() ?first :second;
}

此代码过滤掉未知时间的事件，发出当前事件或前一个事件（ scan() ），具体取决于后面的事件，过滤出重复事件并显示时间。这个运行了几分钟的小程序已经发现一个计划于2015年11月创建的聚会，而在撰写本文时，聚会是2014年12月。可能性是无止境的。

希望我能对如何轻松地将各种技术融合在一起有一个很好的了解：反应式编程以编写超快速的网络代码，无需样板代码的类型安全JSON解析，以及RxJava来快速处理事件流。请享用！