test_class.file: mesos-rxjava-client/src/test/java/com/mesosphere/mesos/rx/java/MesosClientTest.java
test_class.identifier: MesosClientTest
test_case.body: @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_case.class_method_signature: MesosClientTest.testMesosStreamIdIsSavedForSuccessfulSubscribeCall()
test_case.full_signature: @Test public void testMesosStreamIdIsSavedForSuccessfulSubscribeCall()
test_case.identifier: testMesosStreamIdIsSavedForSuccessfulSubscribeCall
test_case.invocations: verifyResponseOk mediaType add headers add headers mediaType create call isEqualTo assertThat get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testMesosStreamIdIsSavedForSuccessfulSubscribeCall()
focal_class.declarator: LOGGER = LoggerFactory.getLogger(MesosClient.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOGGER = LoggerFactory.getLogger(MesosClient.class);
focal_class.type: Logger
focal_class.var_name: LOGGER
focal_class.declarator: MESOS_STREAM_ID = "Mesos-Stream-Id"
focal_class.modifier: private static final
focal_class.original_string: private static final String MESOS_STREAM_ID = "Mesos-Stream-Id";
focal_class.type: String
focal_class.var_name: MESOS_STREAM_ID
focal_class.declarator: exec = Executors.newSingleThreadExecutor(
          new DefaultThreadFactory("stream-monitor-thread", true)
      )
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final ExecutorService exec = Executors.newSingleThreadExecutor(
          new DefaultThreadFactory("stream-monitor-thread", true)
      );
focal_class.type: ExecutorService
focal_class.var_name: exec
focal_class.declarator: mesosUri
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final URI mesosUri;
focal_class.type: URI
focal_class.var_name: mesosUri
focal_class.declarator: receiveCodec
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final MessageCodec<Receive> receiveCodec;
focal_class.type: MessageCodec<Receive>
focal_class.var_name: receiveCodec
focal_class.declarator: subscribe
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final Send subscribe;
focal_class.type: Send
focal_class.var_name: subscribe
focal_class.declarator: streamProcessor
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final Function<Observable<Receive>, Observable<Optional<SinkOperation<Send>>>> streamProcessor;
focal_class.type: Function<Observable<Receive>, Observable<Optional<SinkOperation<Send>>>>
focal_class.var_name: streamProcessor
focal_class.declarator: createPost
focal_class.modifier: @NotNull
      // @VisibleForTesting
      final
focal_class.original_string: @NotNull
      // @VisibleForTesting
      final Func1<Send, Observable<HttpClientRequest<ByteBuf>>> createPost;
focal_class.type: Func1<Send, Observable<HttpClientRequest<ByteBuf>>>
focal_class.var_name: createPost
focal_class.declarator: userAgent
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final UserAgent userAgent;
focal_class.type: UserAgent
focal_class.var_name: userAgent
focal_class.declarator: mesosStreamId = new AtomicReference<>(null)
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final AtomicReference<String> mesosStreamId = new AtomicReference<>(null);
focal_class.type: AtomicReference<String>
focal_class.var_name: mesosStreamId
focal_class.declarator: backpressureTransformer
focal_class.modifier: @NotNull
      private final
focal_class.original_string: @NotNull
      private final Observable.Transformer<byte[], byte[]> backpressureTransformer;
focal_class.type: Observable.Transformer<byte[], byte[]>
focal_class.var_name: backpressureTransformer
focal_class.file: mesos-rxjava-client/src/main/java/com/mesosphere/mesos/rx/java/MesosClient.java
focal_class.identifier: MesosClient
focal_class.class_method_signature: MesosClient.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
      )
focal_class.full_signature:   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
      )
focal_class.identifier: MesosClient
focal_class.parameters: (
          @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
      )
focal_class.signature:  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
      )
focal_class.class_method_signature: MesosClient.openStream()
focal_class.full_signature: @NotNull public AwaitableSubscription openStream()
focal_class.identifier: openStream
focal_class.modifiers: @NotNull public
focal_class.parameters: ()
focal_class.return: AwaitableSubscription
focal_class.signature: AwaitableSubscription openStream()
focal_class.class_method_signature: MesosClient.resolveMesosUri(final @NotNull URI mesosUri)
focal_class.full_signature: @NotNull private static URI resolveMesosUri(final @NotNull URI mesosUri)
focal_class.identifier: resolveMesosUri
focal_class.modifiers: @NotNull private static
focal_class.parameters: (final @NotNull URI mesosUri)
focal_class.return: URI
focal_class.signature: URI resolveMesosUri(final @NotNull URI mesosUri)
focal_class.class_method_signature: MesosClient.getUriFromRedirectResponse(final @NotNull URI mesosUri, @NotNull final HttpClientResponse<ByteBuf> redirectResponse)
focal_class.full_signature: @NotNull // @VisibleForTesting static URI getUriFromRedirectResponse(final @NotNull URI mesosUri, @NotNull final HttpClientResponse<ByteBuf> redirectResponse)
focal_class.identifier: getUriFromRedirectResponse
focal_class.modifiers: @NotNull // @VisibleForTesting static
focal_class.parameters: (final @NotNull URI mesosUri, @NotNull final HttpClientResponse<ByteBuf> redirectResponse)
focal_class.return: URI
focal_class.signature: URI getUriFromRedirectResponse(final @NotNull URI mesosUri, @NotNull final HttpClientResponse<ByteBuf> redirectResponse)
focal_class.class_method_signature: MesosClient.resolveRelativeUri(final @NotNull URI mesosUri, final String location)
focal_class.full_signature: @NotNull // @VisibleForTesting static URI resolveRelativeUri(final @NotNull URI mesosUri, final String location)
focal_class.identifier: resolveRelativeUri
focal_class.modifiers: @NotNull // @VisibleForTesting static
focal_class.parameters: (final @NotNull URI mesosUri, final String location)
focal_class.return: URI
focal_class.signature: URI resolveRelativeUri(final @NotNull URI mesosUri, final String location)
focal_class.class_method_signature: MesosClient.createRedirectUri(@NotNull final URI uri)
focal_class.full_signature: @NotNull // @VisibleForTesting static String createRedirectUri(@NotNull final URI uri)
focal_class.identifier: createRedirectUri
focal_class.modifiers: @NotNull // @VisibleForTesting static
focal_class.parameters: (@NotNull final URI uri)
focal_class.return: String
focal_class.signature: String createRedirectUri(@NotNull final URI uri)
focal_class.class_method_signature: MesosClient.verifyResponseOk(
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_class.full_signature: @NotNull // @VisibleForTesting static Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> verifyResponseOk(
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_class.identifier: verifyResponseOk
focal_class.modifiers: @NotNull // @VisibleForTesting static
focal_class.parameters: (
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_class.return: Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>>
focal_class.signature: Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> verifyResponseOk(
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_class.class_method_signature: MesosClient.getPort(@NotNull final URI uri)
focal_class.full_signature: static int getPort(@NotNull final URI uri)
focal_class.identifier: getPort
focal_class.modifiers: static
focal_class.parameters: (@NotNull final URI uri)
focal_class.return: int
focal_class.signature: int getPort(@NotNull final URI uri)
focal_class.class_method_signature: MesosClient.curryCreatePost(
          @NotNull final URI mesosUri,
          @NotNull final MessageCodec<Send> sendCodec,
          @NotNull final MessageCodec<Receive> receiveCodec,
          @NotNull final UserAgent userAgent,
          @NotNull final AtomicReference<String> mesosStreamId
      )
focal_class.full_signature: @NotNull // @VisibleForTesting static 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
      )
focal_class.identifier: curryCreatePost
focal_class.modifiers: @NotNull // @VisibleForTesting static
focal_class.parameters: (
          @NotNull final URI mesosUri,
          @NotNull final MessageCodec<Send> sendCodec,
          @NotNull final MessageCodec<Receive> receiveCodec,
          @NotNull final UserAgent userAgent,
          @NotNull final AtomicReference<String> mesosStreamId
      )
focal_class.return: Func1<Send, Observable<HttpClientRequest<ByteBuf>>>
focal_class.signature: 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
      )
focal_method.body: @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");
                      }
                  });
              }
          };
      }
focal_method.class_method_signature: MesosClient.verifyResponseOk(
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_method.full_signature: @NotNull // @VisibleForTesting static Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> verifyResponseOk(
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_method.identifier: verifyResponseOk
focal_method.invocations: getStatus code get getHeaders equals contains getHeaders get getHeaders compareAndSet getContent getHeaders flatMap attemptToReadErrorResponse entries withMessage format warn
focal_method.modifiers: @NotNull // @VisibleForTesting static
focal_method.parameters: (
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )
focal_method.return: Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>>
focal_method.signature: Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> verifyResponseOk(
          @NotNull final Send subscription,
          @NotNull final AtomicReference<String> mesosStreamId,
          @NotNull final String receiveMediaType
      )