rx-java - 使用RxJava构建异步REST API

Question

研究 RxJava 来为我们的 API 构建异步支持。现在我们使用 jetty + JAX-RS @Path 注释，但不确定将传入的 REST api 调用与 RxJava API 联系起来的正确方法是什么？

基本上，这是在释放请求线程直到DB 的响应已准备好。

查看了 Vert.x，但这需要 java 7，而我们现在只能使用 java 6。

寻找有关上述内容的建议。典型的方法是什么需要将传入的 http 请求绑定(bind)到 RxJava API。

Answer 1

下面是一个为 JAX-RS 创建 Customer Observable 的示例:

public class ApiService {
    Client client;

    public ApiService() {
        client = ClientBuilder.newClient();
    }

    public Observable<Customer> createCustomerObservable(final int customerId) {
        return Observable.create(new Observable.OnSubscribe<Customer>() {
            @Override
            public void call(final Subscriber<? super Customer> subscriber) {
                client
                        .target("http://domain.com/customers/{id}")
                        .resolveTemplate("id", customerId)
                        .request()
                        .async()
                        .get(new InvocationCallback<Customer>() {
                            @Override
                            public void completed(Customer customer) {
                                // Do something
                                if (!subscriber.isUnsubscribed()) {
                                    subscriber.onNext(customer);
                                    subscriber.onCompleted();
                                }
                            }

                            @Override
                            public void failed(Throwable throwable) {
                                // Process error
                                if (!subscriber.isUnsubscribed()) {
                                    subscriber.onError(throwable);
                                }
                            }
                        });
            }
        });
    }
}