MesosClient { @NotNull static <Send, Receive> Func1<Send, Observable<HttpClientRequest<ByteBuf>>> curryCreatePost( @NotNull final URI mesosUri, @NotNull final MessageCodec<Send> sendCodec, @NotNull final MessageCodec<Receive> receiveCodec, @NotNull final UserAgent userAgent, @NotNull final AtomicReference<String> mesosStreamId ) { return (Send s) -> { final byte[] bytes = sendCodec.encode(s); HttpClientRequest<ByteBuf> request = HttpClientRequest.createPost(mesosUri.getPath()) .withHeader("User-Agent", userAgent.toString()) .withHeader("Content-Type", sendCodec.mediaType()) .withHeader("Accept", receiveCodec.mediaType()); final String streamId = mesosStreamId.get(); if (streamId != null) { request = request.withHeader(MESOS_STREAM_ID, streamId); } final String userInfo = mesosUri.getUserInfo(); if (userInfo != null) { request = request.withHeader( HttpHeaderNames.AUTHORIZATION.toString(), String.format("Basic %s", Base64.getEncoder().encodeToString(userInfo.getBytes())) ); } return just( request .withContent(bytes) ); }; } 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 testRequestUriFromPassedUri() throws Exception { final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost( URI.create("http: StringMessageCodec.UTF8_STRING, StringMessageCodec.UTF8_STRING, new UserAgent( literal("testing", "latest") ), new AtomicReference<>(null) ); final HttpClientRequest<ByteBuf> request = createPost.call("something") .toBlocking() .first(); assertThat(request.getUri()).isEqualTo("/glavin/api/v1/scheduler"); }
@Test public void testBasicAuthHeaderAddedToRequestWhenUserInfoPresentInUri() throws Exception { final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost( URI.create("http: StringMessageCodec.UTF8_STRING, StringMessageCodec.UTF8_STRING, new UserAgent( literal("testing", "latest") ), new AtomicReference<>(null) ); final HttpClientRequest<ByteBuf> request = createPost.call("something") .toBlocking() .first(); final String authHeaderName = HttpHeaderNames.AUTHORIZATION.toString(); final Map<String, String> headers = headersToMap(request.getHeaders()); assertThat(headers).containsKeys(authHeaderName); final String authorization = headers.get(authHeaderName); assertThat(authorization).isEqualTo("Basic dGVzdHVzZXI6dGVzdHBhc3N3b3Jk"); final String base64UserPass = authorization.substring("Basic ".length()); final String userPass = new String(Base64.getDecoder().decode(base64UserPass)); assertThat(userPass).isEqualTo("testuser:testpassword"); }
@Test public void testMesosStreamIdAddedToRequestWhenNonNull() throws Exception { final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost( URI.create("http: StringMessageCodec.UTF8_STRING, StringMessageCodec.UTF8_STRING, new UserAgent( literal("testing", "latest") ), new AtomicReference<>("streamId") ); final HttpClientRequest<ByteBuf> request = createPost.call("something") .toBlocking() .first(); final Map<String, String> headers = headersToMap(request.getHeaders()); assertThat(headers).containsKeys("Mesos-Stream-Id"); assertThat(headers.get("Mesos-Stream-Id")).isEqualTo("streamId"); }
@Test public void testMesosStreamIdNotPresentWhenNull() throws Exception { final Func1<String, Observable<HttpClientRequest<ByteBuf>>> createPost = MesosClient.curryCreatePost( URI.create("http: StringMessageCodec.UTF8_STRING, StringMessageCodec.UTF8_STRING, new UserAgent( literal("testing", "latest") ), new AtomicReference<>(null) ); final HttpClientRequest<ByteBuf> request = createPost.call("something") .toBlocking() .first(); final Map<String, String> headers = headersToMap(request.getHeaders()); assertThat(headers).doesNotContainKeys("Mesos-Stream-Id"); }