【发布时间】：2015-11-06 01:02:54
【问题描述】：
我正在使用 RxJava 和 RxNetty 为 Apache Mesos 的新 HTTP 调度程序 API 编写客户端。
我已经成功地创建了与 RxNetty 的连接，并从生成的分块流中创建了一个Observable<Event>。
现在我正在尝试对一个接收器进行建模，该接收器可用于将调用发送回 Mesos，以声明/拒绝资源提供、确认任务状态更新等。
要发送到 Mesos 的消息是 Call，我需要能够为进入 Sink 的每个 Call 提供 onCompleted 或 onError。这是因为 Mesos 对发送给它的 Call 执行同步验证。
我实际上是在尝试允许以下内容：
final MesosSchedulerClient client = new MesosSchedulerClient();
final Observable<Event> events = client.openEventStream(subscribeCall);
final Observable<Observable<Call>> ackCalls = events
.filter(event -> event.getType() == Event.Type.UPDATE && event.getUpdate().getStatus().hasUuid())
.zipWith(frameworkIDObservable, (Event e, AtomicReference<FrameworkID>> fwId) -> {
final TaskStatus status = e.getUpdate().getStatus();
final Call ackCall = ackUpdate(fwId.get(), status.getUuid(), status.getAgentId(), status.getTaskId());
return Observable.just(ackCall)
.doOnComplete(() -> { ... })
.doOnError((e) -> { ... });
});
client.sink(ackCalls);
现在我想出了一个自定义对象[1]，它扩展了 Subject 并为onCompleted 指定了Call 和Action0，为onError 指定了Action1<Throwable>。不过，如果可能的话，我更愿意使用 RxJava 中的现有结构。我想出的示例用法[2]。
任何指导将不胜感激。
【问题讨论】：
-
我对 mesos 不熟悉，也不完全了解您想要实现的目标。扩展主题似乎是不必要的，因为您不接受订阅者；相反，您可能想要使用 Subscribers.create() 或扩展 Subscriber。
-
感谢@akarnokd 的指导，我将研究如何创建订阅者。
-
谢谢@akarnokd 我改用订阅者了。
标签： java client rx-java event-stream