MesosClient { @NotNull static <Send> Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> verifyResponseOk( @NotNull final Send subscription, @NotNull final AtomicReference<String> mesosStreamId, @NotNull final String receiveMediaType ) { return resp -> { final HttpResponseStatus status = resp.getStatus(); final int code = status.code(); final String contentType = resp.getHeaders().get(HttpHeaderNames.CONTENT_TYPE); if (code == 200 && receiveMediaType.equals(contentType)) { if (resp.getHeaders().contains(MESOS_STREAM_ID)) { final String streamId = resp.getHeaders().get(MESOS_STREAM_ID); mesosStreamId.compareAndSet(null, streamId); } return resp.getContent(); } else { final HttpResponseHeaders headers = resp.getHeaders(); return ResponseUtils.attemptToReadErrorResponse(resp).flatMap(msg -> { final List<Map.Entry<String, String>> entries = headers.entries(); final MesosClientErrorContext context = new MesosClientErrorContext(code, msg, entries); if (code == 200) { throw new MesosException( subscription, context.withMessage( String.format( "Response had Content-Type \"%s\" expected \"%s\"", contentType, receiveMediaType ) ) ); } else if (400 <= code && code < 500) { throw new Mesos4xxException(subscription, context); } else if (500 <= code && code < 600) { throw new Mesos5xxException(subscription, context); } else { LOGGER.warn("Unhandled error: context = {}", context); throw new IllegalStateException("Unhandled error"); } }); } }; } MesosClient(
        @NotNull final URI mesosUri,
        @NotNull final Function<Class<?>, UserAgentEntry> applicationUserAgentEntry,
        @NotNull final MessageCodec<Send> sendCodec,
        @NotNull final MessageCodec<Receive> receiveCodec,
        @NotNull final Send subscribe,
        @NotNull final Function<Observable<Receive>, Observable<Optional<SinkOperation<Send>>>> streamProcessor,
        @NotNull final  Observable.Transformer<byte[], byte[]> backpressureTransformer
    ); @NotNull AwaitableSubscription openStream();  }
@Test public void testMesosStreamIdIsSavedForSuccessfulSubscribeCall() throws Exception { final AtomicReference<String> mesosStreamId = new AtomicReference<>(null); final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk( "Subscribe", mesosStreamId, StringMessageCodec.UTF8_STRING.mediaType() ); final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); nettyResponse.headers().add("Mesos-Stream-Id", "streamId"); nettyResponse.headers().add("Content-Type", StringMessageCodec.UTF8_STRING.mediaType()); final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>( nettyResponse, UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS) ); f.call(response); assertThat(mesosStreamId.get()).isEqualTo("streamId"); }
@Test public void testMesosStreamIdIsNotSavedForUnsuccessfulSubscribeCall() throws Exception { final AtomicReference<String> mesosStreamId = new AtomicReference<>(null); final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk( "Subscribe", mesosStreamId, StringMessageCodec.UTF8_STRING.mediaType() ); final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST); nettyResponse.headers().add("Mesos-Stream-Id", "streamId"); nettyResponse.headers().add("Content-Type", StringMessageCodec.UTF8_STRING.mediaType()); final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>( nettyResponse, UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS) ); try { f.call(response); } catch (Mesos4xxException e) { } assertThat(mesosStreamId.get()).isEqualTo(null); }
@Test public void testVerifyResponseOk_ensuresContentTypeOfResponseMatchesReceiveCodec() throws Exception { final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk( "Subscribe", new AtomicReference<>(), StringMessageCodec.UTF8_STRING.mediaType() ); final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); nettyResponse.headers().add("Content-Type", "text/html"); final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>( nettyResponse, UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS) ); try { f.call(response); } catch (MesosException e) { final String expected = String.format( "Response had Content-Type \"%s\" expected \"%s\"", "text/html", StringMessageCodec.UTF8_STRING.mediaType() ); assertThat(e.getContext().getMessage()).isEqualTo(expected); } }