class MesosClient {

    @NotNull
    // @VisibleForTesting
    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) {
                        // this means that even though we got back a 200 it's not the sort of response we were expecting
                        // For example hitting an endpoint that returns an html document instead of a document of type
                        // `receiveMediaType`
                        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);
                        // This shouldn't actually ever happen, but it's here for completeness of the if-else tree
                        // that always has to result in an exception being thrown so the compiler is okay with this
                        // lambda
                        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
    );

}