使用RxJava构建异步REST API

Question

查看RxJava为我们的API构建异步支持。 现在我们使用jetty + JAX-RS @Path注解，并且不确定 将传入的REST API调用绑定到RxJava API的正确方法是什么。

基本上这是在释放请求线程的情况下，直到 DB的响应已准备就绪。

看着Vert.x但需要Java 7和我们绑现在到Java 6

寻找关于上述建议。人们将传入的http请求绑定到RxJava API的典型方法 是什么。

Answer 1

像下面这样的东西应该为码头工作：

public class ApiService { 
    HttpClient httpClient; 

    public ApiService(HttpClient httpClient,) { 
     this.httpClient = httpClient; 
    } 

    public <RequestType, ResultType> Observable<ResultType> createApiObservable(final RequestType requestContent) { 
     return Observable.create(new Observable.OnSubscribe<ResultType>() { 
      @Override 
      public void call(final Subscriber<? super ResultType> subscriber) { 
       // Create the request content for your API. Your logic here... 
       ContentProvider contentProvider = serializeRequest(requestContent); 

       httpClient 
         .newRequest("http://domain.com/path") 
         .content(contentProvider) 
         .send(new Response.CompleteListener() { 
          @Override 
          void onComplete(Result result) { 
           // Pass along the error if one occurred. 
           if (result.isFailed()) { 
            subscriber.onError(result.getFailure()); 
            return; 
           } 

           // Convert the response data to the ResultType. Your logic here... 
           ResultType resultContent = parseResponse(result.getResponse()); 

           // Send the result to the subscriber. 
           subscriber.onNext(responseBytes); 
           subscriber.onCompleted(); 
          } 
         }); 
      } 
     }); 
    } 
} 

Answer 2

这里，将创建一个客户观察到的，JAX-RS一个例子：

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); 
           } 
          } 
         }); 
      } 
     }); 
    } 
}