Использование RxJava для построения асинхронного API REST

Question

Глядя на RxJava для создания асинхронной поддержки наших API. Прямо сейчас мы используем аннотации jetty + JAX-RS @Path и не уверены . Каков правильный подход для привязки входящего вызова REST к API RxJava.

В основном это в контексте освобождения потока запроса до ответ от БД готов.

Посмотрел Vert.x но требует Java 7, и мы связаны прямо сейчас Java 6.

В поисках предложений относительно выше. каковы типичные подходы , которые требуется связать входящий HTTP-запрос с API-интерфейсом RxJava.

Answer 1

Нечто подобное должно работать для Jetty:

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

Вот пример того, что бы создать клиента Observable для 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); 
           } 
          } 
         }); 
      } 
     }); 
    } 
}