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