io.reactivex.netty.protocol.http.server.HttpServer#shutdown ( )源码实例Demo

@Test
public void testStreamDoesNotRunWhenSubscribeFails_mesos5xxResponse() throws Throwable {
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClient(uri);

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up because subscribe will 500");
    } catch (Mesos5xxException e) {
        // expected
        final MesosClientErrorContext ctx = e.getContext();
        assertThat(ctx.getStatusCode()).isEqualTo(500);
    } finally {
        server.shutdown();
    }
}
 

@Test
public void testStreamDoesNotRunWhenSubscribeFails_mismatchContentType() throws Throwable {
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.OK);
        response.getHeaders().setHeader("Content-Type", "application/json");
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClient(uri);

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up because of content type mismatch");
    } catch (MesosException e) {
        // expected
        final MesosClientErrorContext ctx = e.getContext();
        assertThat(ctx.getStatusCode()).isEqualTo(200);
        assertThat(ctx.getMessage()).isEqualTo("Response had Content-Type \"application/json\" expected \"text/plain;charset=utf-8\"");
    } finally {
        server.shutdown();
    }
}
 

@Test
public void testStreamDoesNotRunWhenSubscribeFails_mesos4xxResponse() throws Throwable {
    final String errorMessage = "Error message that should come from the server";
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.BAD_REQUEST);
        final byte[] msgBytes = errorMessage.getBytes(StandardCharsets.UTF_8);
        response.getHeaders().setHeader("Content-Length", msgBytes.length);
        response.getHeaders().setHeader("Content-Type", "text/plain;charset=utf-8");
        response.writeBytes(msgBytes);
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClient(uri);

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up because subscribe will 400");
    } catch (Mesos4xxException e) {
        // expected
        final MesosClientErrorContext ctx = e.getContext();
        assertThat(ctx.getStatusCode()).isEqualTo(400);
        assertThat(ctx.getMessage()).isEqualTo(errorMessage);
    } finally {
        server.shutdown();
    }
}
 

@Test
@Ignore
public void testBurstyObservable_missingBackpressureException() throws Throwable {
    final String subscribedMessage = "{\"type\": \"SUBSCRIBED\",\"subscribed\": {\"framework_id\": {\"value\":\"12220-3440-12532-2345\"},\"heartbeat_interval_seconds\":15.0}";
    final String heartbeatMessage = "{\"type\":\"HEARTBEAT\"}";
    final byte[] hmsg = heartbeatMessage.getBytes(StandardCharsets.UTF_8);
    final byte[] hbytes = String.format("%d\n", heartbeatMessage.getBytes().length).getBytes(StandardCharsets.UTF_8);

    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.OK);
        response.getHeaders().setHeader("Content-Type", "text/plain;charset=utf-8");
        writeRecordIOMessage(response, subscribedMessage);
        for (int i = 0; i < 20000; i++) {
            response.writeBytes(hbytes);
            response.writeBytes(hmsg);
        }
        return response.flush();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClientForStreaming(uri).build();

    try {
        client.openStream().await();
        fail("Expect an exception to be propagated up due to backpressure");
    } catch (MissingBackpressureException e) {
        // expected
        e.printStackTrace();
        assertThat(e.getMessage()).isNullOrEmpty();
    } finally {
        server.shutdown();
    }
}
 

@Test
public void testBurstyObservable_unboundedBufferSucceeds() throws Throwable {
    msgNo = 0;
    final int numMessages = 20000;
    final String subscribedMessage = "{\"type\": \"SUBSCRIBED\",\"subscribed\": {\"framework_id\": {\"value\":\"12220-3440-12532-2345\"},\"heartbeat_interval_seconds\":15.0}";
    final String heartbeatMessage = "{\"type\":\"HEARTBEAT\"}";
    final RequestHandler<ByteBuf, ByteBuf> handler = (request, response) -> {
        response.setStatus(HttpResponseStatus.OK);
        response.getHeaders().setHeader("Content-Type", "text/plain;charset=utf-8");
        writeRecordIOMessage(response, subscribedMessage);
        for (int i = 0; i < numMessages; i++) {
            writeRecordIOMessage(response, heartbeatMessage);
        }
        return response.close();
    };
    final HttpServer<ByteBuf, ByteBuf> server = RxNetty.createHttpServer(0, handler);
    server.start();
    final URI uri = URI.create(String.format("http://localhost:%d/api/v1/scheduler", server.getServerPort()));
    final MesosClient<String, String> client = createClientForStreaming(uri)
            .onBackpressureBuffer()
            .build();

    try {
        client.openStream().await();
    } finally {
        // 20000 heartbeats PLUS 1 subscribe
        assertEquals("All heartbeats received (plus the subscribed)", 1 + numMessages, msgNo);
        server.shutdown();
    }
}