class MesosClient {

    @NotNull
    // @VisibleForTesting
    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
    );

}