JAX-RSクライアントを利用したWeb APIの非同期呼び出しと、RxJavaを使った待ち合わせ

ApiServer.java

import com.sun.net.httpserver.HttpServer;
import java.net.URI;
import org.glassfish.jersey.jdkhttp.JdkHttpServerFactory;
import org.glassfish.jersey.server.ResourceConfig;

/**
 * 簡易JAX-RSサーバ.
 * 
 * @author kawasima
 */
public class ApiServer {
    public static void main(String... args) {
        URI uri = URI.create("http://localhost:8888/rest/");
        
        ResourceConfig config = new ResourceConfig();
        config.register(ExampleResource.class);
        
        HttpServer httpServer = JdkHttpServerFactory.createHttpServer(uri, config);
    }
}

ExampleResource.java

import java.time.LocalDateTime;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.QueryParam;

/**
 * 0～9秒ランダムでウェイトするWeb API.
 * 
 * @author kawasima
 */
@Path("/")
public class ExampleResource {
    private static final Random RANDOM = new Random();
    @GET
    public String index(@QueryParam("seq") int seq) {
        LocalDateTime t1 = LocalDateTime.now();
        long waitingSeconds = RANDOM.nextInt(10);
        try {
            TimeUnit.SECONDS.sleep(waitingSeconds);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        return seq + "," + t1.toString() + "," + LocalDateTime.now() + "," + waitingSeconds;
    }
}

ParallelApi.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Response;
import rx.Observable;
import rx.Subscriber;

/**
 * 非同期APIを100同時リクエスト呼び出し, 結果を順次処理する.
 * 
 * @author kawasima
 */
public class ParallelApi {
    private static Observable<Response> call(WebTarget target, int seq) {
        Future<Response> f = target.queryParam("seq", seq).request().async().get();
        return Observable.from(f);
    }

    public static void main(String... args) {
       Client client = ClientBuilder.newClient();
        WebTarget target = client.target("http://localhost:8888/rest/");
        
        List<Observable<Response>> responses = new ArrayList<>();
        for (int i=0; i<100; i++) {
            responses.add(call(target, i));
        }
        
        System.err.println("subscribe!");
        StringBuilder sb = new StringBuilder();
        Observable.merge(responses).subscribe(new Subscriber<Response> () {
           @Override
           public void onNext(Response response) {
               if (response.getStatus() == 200) {
                   sb.append(response.readEntity(String.class))
                           .append('\n');
               } else {
                   throw new IllegalStateException(response.getStatusInfo().toString());
               }
           }

           @Override
           public void onCompleted() {
               System.out.println("Complete");
           }

           @Override
           public void onError(Throwable e) {
               System.out.println(e);
           }

        });
        System.out.println(sb);
        client.close();
    }
}

pom.dependencies

    <dependencies>
        <dependency>
            <groupId>io.reactivex</groupId>
            <artifactId>rxjava</artifactId>
            <version>1.0.7</version>
        </dependency>
        <dependency>
            <groupId>org.glassfish.jersey.core</groupId>
            <artifactId>jersey-client</artifactId>
            <version>2.16</version>
        </dependency>
        <dependency>
            <groupId>org.glassfish.jersey.core</groupId>
            <artifactId>jersey-server</artifactId>
            <version>2.16</version>
        </dependency>
        <dependency>
            <groupId>org.glassfish.jersey.containers</groupId>
            <artifactId>jersey-container-jdk-http</artifactId>
            <version>2.16</version>
        </dependency>
    </dependencies>