Java 类io.netty.handler.codec.http.HttpResponseStatus 实例源码

@Override
public void handle(@NotNull RoutingContext context) {
  context.vertx().<Buffer>executeBlocking(future -> {
    try (final ByteArrayOutputStream output = new ByteArrayOutputStream()) {
      new ProtobufFormatter(registry.metricFamilySamples()).write(output);
      future.complete(Buffer.buffer(output.toByteArray()));
    } catch (IOException e) {
      future.fail(e);
    }
  }, false, result -> {
    if (result.succeeded()) {
      context.response()
          .setStatusCode(HttpResponseStatus.OK.code())
          .putHeader("Content-Type", ProtobufFormatter.CONTENT_TYPE)
          .end(result.result());
    } else {
      context.fail(result.cause());
    }
  });
}

@Override
public FullHttpResponse get(ChannelHandlerContext channelHandlerContext, QueryDecoder queryDecoder, PathProvider path, HttpRequest httpRequest) throws Exception
{
    CloudNet.getLogger().debug("HTTP Request from " + channelHandlerContext.channel().remoteAddress());

    StringBuilder stringBuilder = new StringBuilder();

    try (InputStream inputStream = WebsiteDocumentation.class.getClassLoader().getResourceAsStream("files/api-doc.txt");
         BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)))
    {
        String input;
        while ((input = bufferedReader.readLine()) != null)
        {
            stringBuilder.append(input).append(System.lineSeparator());
        }
    }

    String output = stringBuilder.substring(0);
    ByteBuf byteBuf = Unpooled.wrappedBuffer(output.getBytes(StandardCharsets.UTF_8));
    FullHttpResponse fullHttpResponse = new DefaultFullHttpResponse(httpRequest.getProtocolVersion(), HttpResponseStatus.OK, byteBuf);
    fullHttpResponse.headers().set("Content-Type", "text/plain");
    return fullHttpResponse;
}

@SuppressWarnings("unchecked")
void refreshConfig() {
  HttpHeaders headers = new HttpHeaders();
  headers.add("Content-Type", "application/json;charset=UTF-8");
  headers.add("Authorization", token);
  HttpEntity<String> entity = new HttpEntity<>(headers);
  ResponseEntity<String> exchange = rest.exchange(composeAPI(), HttpMethod.GET, entity, String.class);
  if (HttpResponseStatus.OK.code() == exchange.getStatusCode().value()) {
    try {
      Map<String, Object> body = JsonUtils.OBJ_MAPPER.readValue(exchange.getBody(),
          new TypeReference<Map<String, Object>>() {
          });
      refreshConfigItems((Map<String, Object>) body.get("configurations"));
    } catch (IOException e) {
      LOGGER.error("JsonObject parse config center response error: ", e);
    }
  } else {
    LOGGER.error("fetch configuration failed, error code:{} for {}",
        exchange.getStatusCodeValue(),
        exchange.getBody());
  }
}

private void refreshMembers(MemberDiscovery memberDiscovery) {
  if (CONFIG_CENTER_CONFIG.getAutoDiscoveryEnabled()) {
    String configCenter = memberDiscovery.getConfigServer();
    IpPort ipPort = NetUtils.parseIpPortFromURI(configCenter);
    clientMgr.findThreadBindClientPool().runOnContext(client -> {
      HttpClientRequest request = client.get(ipPort.getPort(), ipPort.getHostOrIp(), URIConst.MEMBERS, rsp -> {
        if (rsp.statusCode() == HttpResponseStatus.OK.code()) {
          rsp.bodyHandler(buf -> {
            memberDiscovery.refreshMembers(buf.toJsonObject());
          });
        }
      });
      SignRequest signReq = createSignRequest(request.method().toString(),
          configCenter + URIConst.MEMBERS, new HashMap<>(), null);
      if (ConfigCenterConfig.INSTANCE.getToken() != null) {
        request.headers().add("X-Auth-Token", ConfigCenterConfig.INSTANCE.getToken());
      }
      authHeaderProviders.forEach(provider -> request.headers().addAll(provider.getSignAuthHeaders(signReq)));
      request.end();
    });
  }
}

private StoreResponse toStoreResponse(HttpResponseStatus httpResponseStatus,
        HttpResponseHeaders httpResponseHeaders, InputStream contentInputStream) throws IOException {

    List<Entry<String, String>> headerEntries = httpResponseHeaders.entries();

    String[] headers = new String[headerEntries.size()];
    String[] values = new String[headerEntries.size()];

    int i = 0;

    for(Entry<String, String> headerEntry: headerEntries) {
        headers[i] = headerEntry.getKey();
        values[i] = headerEntry.getValue();
        i++;
    }

    StoreResponse storeResponse = new StoreResponse(
            headers,
            values,
            httpResponseStatus.code(),
            contentInputStream);

    return storeResponse;
}

private static JsonObject createErrorResponse(
    RoutingContext routingContext, Throwable exception) {
  // by default ...
  int status = 500;

  // intercept specific exception types and assign the relevant HTTP status code
  if (InvalidRequestException.class.isAssignableFrom(exception.getClass())
      || SubscriptionUnsupportedException.class.isAssignableFrom(exception.getClass())) {
    status = HttpResponseStatus.BAD_REQUEST.code();
  } else if (InvalidCredentialsException.class.isAssignableFrom(exception.getClass())) {
    status = HttpResponseStatus.UNAUTHORIZED.code();
  } else if (AccessDeniedException.class.isAssignableFrom(exception.getClass())) {
    status = HttpResponseStatus.FORBIDDEN.code();
  }

  return new JsonObject()
      .put("timestamp", LocalDateTime.now().toString())
      .put("statusCode", status)
      .put("statusMessage", HttpResponseStatus.valueOf(status).reasonPhrase())
      .put("path", routingContext.request().uri())
      .put("message", orDefault(exception.getMessage(), ""))
      .put("stackTrace", displayable(exception.getStackTrace()));
}

@Override
public RoleClient create(Throwable cause) {
    LOGGER.error("RoleClient find {}, by using {}", cause.getMessage(), RESOURCE_ID);
    if (null != fallback) {
        return fallback;
    }

    responseObjectDTO = new HashMap();
    Map result = new HashMap();
    result.put("totalElements", 0l);
    responseObjectDTO.put("data", result);
    responseObjectDTO.put("code", HttpResponseStatus.BAD_GATEWAY.code());
    responseObjectDTO.put("msg", "RoleClient is not available");

    return fallback = new RoleClient() {
        @Override
        public Map roles() {
            LOGGER.error("roles() invoke fail");
            return responseObjectDTO;
        }
    };
}

@Test
public void canHandleFailureForGetDataset() {
  String id = "aDatasetId";

  Exception exception = new RuntimeException("boom!");

  when(datasetDao.get(id)).thenThrow(exception);

  String endpoint = "dataset/" + id;
  HttpResponse response = read(endpoint);

  assertThat(response.getStatusCode(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
  assertThat(
      response.getStatusMessage(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.reasonPhrase()));

  assertThatErrorResponseIsCorrect(
      response.getPayload(), exception.getMessage(), endpoint, INTERNAL_SERVER_ERROR);
}

public static void startDownTask(TaskInfo taskInfo, HttpRequest httpRequest,
    HttpResponse httpResponse, Channel clientChannel) {
  HttpHeaders httpHeaders = httpResponse.headers();
  HttpDownInfo httpDownInfo = new HttpDownInfo(taskInfo, httpRequest);
  HttpDownServer.DOWN_CONTENT.put(taskInfo.getId(), httpDownInfo);
  httpHeaders.clear();
  httpResponse.setStatus(HttpResponseStatus.OK);
  httpHeaders.set(HttpHeaderNames.CONTENT_TYPE, "text/html");
  String host = HttpDownServer.isDev() ? "localhost"
      : ((InetSocketAddress) clientChannel.localAddress()).getHostString();
  String js =
      "<script>window.top.location.href='http://" + host + ":" + HttpDownServer.VIEW_SERVER_PORT
          + "/#/tasks/new/" + httpDownInfo
          .getTaskInfo().getId()
          + "';</script>";
  HttpContent content = new DefaultLastHttpContent();
  content.content().writeBytes(js.getBytes());
  httpHeaders.set(HttpHeaderNames.CONTENT_LENGTH, js.getBytes().length);
  clientChannel.writeAndFlush(httpResponse);
  clientChannel.writeAndFlush(content);
  clientChannel.close();
}

@Test
public void willCreateANewUserIfNoneExists() {
  String userName = "aName";
  String password = "aPassword";

  when(authenticationDao.exists(userName)).thenReturn(false);

  String payload = "{ \"username\": \"" + userName + "\", \"password\": \"" + password + "\" }";

  HttpResponse response = post("login", payload);

  assertThat(response.getStatusCode(), is(HttpResponseStatus.OK.code()));
  assertThat(response.getStatusMessage(), is(HttpResponseStatus.OK.reasonPhrase()));
  // we expect the session cookie to have been set
  assertThat(response.getHeaders(), hasKey("set-cookie"));
  // we expect to be redirected to the about page
  assertThat(response.getUrl(), Matchers.endsWith("/dragoman/about.hbs"));

  verify(authenticationDao, never()).isValid(userName, password);
  verify(authenticationDao, times(1)).createUser(userName, password);
}

private static FullHttpResponse getDino(XrpcRequest request, List<Dino> dinos) {
  try {
    DinoGetRequest getRequest =
        DinoGetRequest.parseFrom(CodedInputStream.newInstance(request.getData().nioBuffer()));
    Optional<Dino> dinoOptional =
        dinos.stream().filter(xs -> xs.getName().equals(getRequest.getName())).findFirst();

    if (dinoOptional.isPresent()) {
      DinoGetReply getReply = DinoGetReply.newBuilder().setDino(dinoOptional.get()).build();
      ByteBuf resp = request.getByteBuf();
      resp.ensureWritable(CodedOutputStream.computeMessageSizeNoTag(getReply), true);
      getReply.writeTo(new ByteBufOutputStream(resp));

      return Recipes.newResponse(
          HttpResponseStatus.OK,
          request.getByteBuf().writeBytes(resp),
          Recipes.ContentType.Application_Octet_Stream);
    }

  } catch (IOException e) {
    return Recipes.newResponseBadRequest("Malformed GetDino Request: " + e.getMessage());
  }

  return Recipes.newResponseOk("Dino not Found");
}

private void configResponseCodeMeters() {
  final Map<HttpResponseStatus, String> meterNamesByStatusCode = new ConcurrentHashMap<>(6);

  // Create the proper metrics containers
  final String NAME_PREFIX = "responseCodes.";
  meterNamesByStatusCode.put(HttpResponseStatus.OK, NAME_PREFIX + "ok");
  meterNamesByStatusCode.put(HttpResponseStatus.CREATED, NAME_PREFIX + "created");
  meterNamesByStatusCode.put(HttpResponseStatus.NO_CONTENT, NAME_PREFIX + "noContent");
  meterNamesByStatusCode.put(HttpResponseStatus.BAD_REQUEST, NAME_PREFIX + "badRequest");
  meterNamesByStatusCode.put(HttpResponseStatus.NOT_FOUND, NAME_PREFIX + "notFound");
  meterNamesByStatusCode.put(
      HttpResponseStatus.TOO_MANY_REQUESTS, NAME_PREFIX + "tooManyRequests");
  meterNamesByStatusCode.put(
      HttpResponseStatus.INTERNAL_SERVER_ERROR, NAME_PREFIX + "serverError");

  for (Map.Entry<HttpResponseStatus, String> entry : meterNamesByStatusCode.entrySet()) {
    ctx.getMetersByStatusCode().put(entry.getKey(), metricRegistry.meter(entry.getValue()));
  }
}

/**
 * Returns a full HTTP response with the specified status, content type, and custom headers.
 *
 * <p>Headers should be specified as a map of strings. For example, to allow CORS, add the
 * following key and value: "access-control-allow-origin", "http://foo.example"
 *
 * <p>If content type or content length are passed in as custom headers, they will be ignored.
 * Instead, content type will be as specified by the parameter contentType and content length will
 * be the length of the parameter contentLength.
 */
public static FullHttpResponse newResponse(
    HttpResponseStatus status,
    ByteBuf payload,
    ContentType contentType,
    Map<String, String> customHeaders) {
  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, payload);

  if (customHeaders != null) {
    for (Map.Entry<String, String> entry : customHeaders.entrySet()) {
      response.headers().set(entry.getKey(), entry.getValue());
    }
  }

  response.headers().set(CONTENT_TYPE, contentType.value);
  response.headers().setInt(CONTENT_LENGTH, payload.readableBytes());

  return response;
}

@Override
public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
    String path = request.getDecodedPath().substring(1);
    String[] pathArray = path.split("/");

    if (pathArray.length > 0 && options.getBasePath().equals(pathArray[0])) {
        Observable<TusResponse> tResp = handlerFactory.makeHandler(request).handle();
        return tResp.map(tr -> { 
            tr.getHeaders().stream()
            .forEach(h -> response.setHeader(h.getName(), h.getValue()));
            response.setStatus(HttpResponseStatus.valueOf(tr.getStatusCode()));
            return null;
        });
    }
    return http404(response);
}

@Test
public void testConcurrentPatch() throws InterruptedException {
    HttpClientRequest<ByteBuf, ByteBuf> post = post(100L);

    Observable<byte[]> slowContent = Observable.just("hello ".getBytes()).repeat()
            .zipWith(Observable.interval(50, TimeUnit.MILLISECONDS).startWith(0L), (data, nop) -> data).take(10);
    Observable<byte[]> fastContent = Observable.just("goodbye ".getBytes()).repeat()
            .zipWith(Observable.interval(10, TimeUnit.MILLISECONDS).startWith(0L), (data, nop) -> data).take(10);

    Iterator<HttpClientResponse<ByteBuf>> iterator = post.map(this::getLocation)
            .flatMap(location -> Observable.merge(
                    patch(location, 0 , slowContent),
                    patch(location, 0, fastContent).delay(120, TimeUnit.MILLISECONDS)))
            .toBlocking().getIterator();

    // the first response should be the failure
    assertThat(iterator.next()).isNotNull()
            .extracting(HttpClientResponse::getStatus).containsExactly(HttpResponseStatus.BAD_REQUEST);

    // the second one should be sucessfull
    assertThat(iterator.next()).isNotNull()
            .extracting(HttpClientResponse::getStatus).containsExactly(HttpResponseStatus.NO_CONTENT);
}

@Test
public void shouldHandlerRequestAndResponse() {
    inboundChannel.pipeline().addLast(handler);

    DefaultFullHttpRequest req = new DefaultFullHttpRequest(
            HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
    inboundChannel.write(req);

    assertEquals(1, inboundChannel.outboundMessages().size());

    Object outboundReq = inboundChannel.outboundMessages().poll();
    assertTrue(outboundReq instanceof ByteBuf);
    assertEquals("GET / HTTP/1.1\r\n\r\n", new String(readBytes((ByteBuf) outboundReq)));

    DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    assertFalse(inboundChannel.writeInbound(resp));

    assertEquals(1, outboundChannel.outboundMessages().size());
    assertEquals(resp, outboundChannel.outboundMessages().poll());

    resp.release();
}

@Override
protected ChannelHandler newNonSslHandler(ChannelHandlerContext context) {
    return new ChannelInboundHandlerAdapter() {

        private HttpResponseEncoder encoder = new HttpResponseEncoder();

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            LOG.trace("Received non-SSL request, returning redirect");
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.MOVED_PERMANENTLY, Unpooled.EMPTY_BUFFER);
            response.headers().set(Names.LOCATION, redirectAddress);
            LOG.trace(Constants.LOG_RETURNING_RESPONSE, response);
            encoder.write(ctx, response, ctx.voidPromise());
            ctx.flush();
        }
    };
}

public void smokeTest(String contentType) throws Exception {
    Web3 web3Mock = Mockito.mock(Web3.class);
    String mockResult = "output";
    Mockito.when(web3Mock.web3_sha3(Mockito.anyString())).thenReturn(mockResult);
    CorsConfiguration mockCorsConfiguration = Mockito.mock(CorsConfiguration.class);
    Mockito.when(mockCorsConfiguration.hasHeader()).thenReturn(true);
    Mockito.when(mockCorsConfiguration.getHeader()).thenReturn("*");

    int randomPort = 9999;//new ServerSocket(0).getLocalPort();

    List<ModuleDescription> filteredModules = Collections.singletonList(new ModuleDescription("web3", "1.0", true, Collections.emptyList(), Collections.emptyList()));
    JsonRpcWeb3FilterHandler filterHandler = new JsonRpcWeb3FilterHandler("*");
    JsonRpcWeb3ServerHandler serverHandler = new JsonRpcWeb3ServerHandler(web3Mock, filteredModules);
    JsonRpcNettyServer server = new JsonRpcNettyServer(InetAddress.getLoopbackAddress(), randomPort, 0, Boolean.TRUE, mockCorsConfiguration, filterHandler, serverHandler);
    server.start();

    HttpURLConnection conn = sendJsonRpcMessage(randomPort, contentType);
    JsonNode jsonRpcResponse = OBJECT_MAPPER.readTree(conn.getInputStream());

    assertThat(conn.getResponseCode(), is(HttpResponseStatus.OK.code()));
    assertThat(jsonRpcResponse.at("/result").asText(), is(mockResult));
    server.stop();
}

public static int writeBack(Channel channel, boolean isSuccess, String resultStr, boolean isKeepAlive) {
    ByteBuf content = Unpooled.copiedBuffer(resultStr, Constants.DEFAULT_CHARSET);
    HttpResponseStatus status;
    if (isSuccess) {
        status = HttpResponseStatus.OK;
    } else {
        status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
    }
    FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, status, content);
    //logger.info("result str:{}", resultStr);
    res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
    HttpHeaders.setContentLength(res, content.readableBytes());
    try {
        ChannelFuture f = channel.writeAndFlush(res);
        if (isKeepAlive) {
            HttpHeaders.setKeepAlive(res, true);
        } else {
            HttpHeaders.setKeepAlive(res, false);//set keepalive closed
            f.addListener(ChannelFutureListener.CLOSE);
        }
    } catch (Exception e2) {
        logger.warn("Failed to send HTTP response to remote, cause by:", e2);
    }

    return content.readableBytes();
}

public DataStore(Configuration conf) throws QonduitException {

        try {
            final BaseConfiguration apacheConf = new BaseConfiguration();
            Configuration.Accumulo accumuloConf = conf.getAccumulo();
            apacheConf.setProperty("instance.name", accumuloConf.getInstanceName());
            apacheConf.setProperty("instance.zookeeper.host", accumuloConf.getZookeepers());
            final ClientConfiguration aconf = new ClientConfiguration(Collections.singletonList(apacheConf));
            final Instance instance = new ZooKeeperInstance(aconf);
            connector = instance
                    .getConnector(accumuloConf.getUsername(), new PasswordToken(accumuloConf.getPassword()));
        } catch (Exception e) {
            throw new QonduitException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error creating DataStoreImpl",
                    e.getMessage(), e);
        }
    }

public void refreshConfig(String configcenter) {
  clientMgr.findThreadBindClientPool().runOnContext(client -> {
    String path = URIConst.ITEMS + "?dimensionsInfo=" + StringUtils.deleteWhitespace(serviceName);
    IpPort ipPort = NetUtils.parseIpPortFromURI(configcenter);
    HttpClientRequest request = client.get(ipPort.getPort(), ipPort.getHostOrIp(), path, rsp -> {
      if (rsp.statusCode() == HttpResponseStatus.OK.code()) {
        rsp.bodyHandler(buf -> {
          try {
            parseConfigUtils
                .refreshConfigItems(JsonUtils.OBJ_MAPPER.readValue(buf.toString(),
                    new TypeReference<LinkedHashMap<String, Map<String, String>>>() {
                    }));
          } catch (IOException e) {
            LOGGER.error("config refresh result parse fail", e);
          }
        });
      } else {
        LOGGER.error("fetch config fail");
      }
    });
    Map<String, String> headers = new HashMap<>();
    headers.put("x-domain-name", tenantName);
    if (ConfigCenterConfig.INSTANCE.getToken() != null) {
      headers.put("X-Auth-Token", ConfigCenterConfig.INSTANCE.getToken());
    }
    request.headers().addAll(headers);
    authHeaderProviders.forEach(provider -> request.headers()
        .addAll(provider.getSignAuthHeaders(createSignRequest(request.method().toString(),
            configcenter + path, headers, null))));
    request.exceptionHandler(e -> {
      LOGGER.error("config refresh fail {}", e.getMessage());
    });
    request.end();
  });
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, StrictTransportResponse msg) throws Exception {
    QonduitException e = new QonduitException(HttpResponseStatus.NOT_FOUND.code(),
            "Returning HTTP Strict Transport Security response", null, null);
    e.addResponseHeader(HSTS_HEADER_NAME, hstsMaxAge);
    // Don't call sendHttpError from here, throw an error instead and let
    // the exception handler catch it.
    throw e;
}

/**
 * Tells if HTTP response server failed to accept the request.
 * 
 * @param status instance of {@code HttpResponseStatus}
 * @return {@code true} if it is bad request; otherwise,
 * {@code false}
 */
public static boolean isInvalid(HttpResponseStatus status) {
  if (status.equals(HttpResponseStatus.BAD_REQUEST) ||
      status.equals(HttpResponseStatus.NOT_IMPLEMENTED)) {
    return true;
  } else {
    return false;
  }
}

public void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
   FullHttpResponse response = new DefaultFullHttpResponse(
         HttpVersion.HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
   response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain; charset=UTF-8");

   // Close the connection as soon as the error message is sent.
   ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

/**
 * When file timestamp is the same as what the browser is sending up, send a "304 Not Modified"
 *
 * @param ctx
 *            Context
 */
public void sendNotModified(ChannelHandlerContext ctx) {
   FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_MODIFIED);
   setDateHeader(response);

   // Close the connection as soon as the error message is sent.
   ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

/**
 * Transforms the rxNetty's client response Observable to DocumentServiceResponse Observable.
 * 
 * 
 * Once the the customer code subscribes to the observable returned by the {@link AsyncDocumentClient} CRUD APIs,
 * the subscription goes up till it reaches the source rxNetty's observable, and at that point the HTTP invocation will be made.
 * 
 * @param clientResponseObservable
 * @param request
 * @return {@link Observable}
 */
private Observable<DocumentServiceResponse> toDocumentServiceResponse(Observable<HttpClientResponse<ByteBuf>> clientResponseObservable, 
        RxDocumentServiceRequest request) {

    return clientResponseObservable.flatMap(clientResponse -> {

        // header key/value pairs
        HttpResponseHeaders httpResponseHeaders = clientResponse.getHeaders();
        HttpResponseStatus httpResponseStatus = clientResponse.getStatus();

        Observable<InputStream> inputStreamObservable;

        if (request.getOperationType() == OperationType.Delete) {
            // for delete we don't expect any body
            inputStreamObservable = Observable.just(null);
        } else {
            // transforms the observable<ByteBuf> to Observable<InputStream>
            inputStreamObservable = toInputStream(clientResponse.getContent());
        }

        Observable<StoreResponse> storeResponseObservable = inputStreamObservable
                .map(contentInputStream -> {
                    try {
                        // If there is any error in the header response this throws exception
                        validateOrThrow(request, httpResponseStatus, httpResponseHeaders, contentInputStream);

                        // transforms to Observable<StoreResponse>
                        return toStoreResponse(httpResponseStatus, httpResponseHeaders, contentInputStream);
                    } catch (Exception e) {
                        throw Exceptions.propagate(e);
                    }
                });

        return storeResponseObservable;

    }).map(storeResponse -> new DocumentServiceResponse(storeResponse));
}

@Test
public void canHandleFailureForGetAllDatasets() {
  Throwable exception = new RuntimeException("boom!");

  when(datasetDao.getAll(anyString())).thenReturn(Observable.error(exception));

  String endpoint = "datasets";
  HttpResponse response = read(endpoint);
  assertThat(response.getStatusCode(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
  assertThat(
      response.getStatusMessage(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.reasonPhrase()));

  assertThatErrorResponseIsCorrect(
      response.getPayload(), exception.getMessage(), endpoint, INTERNAL_SERVER_ERROR);
}

@Test
public void canGetDataset() {
  Dataset dataset = aPersistedDataset();

  when(datasetDao.get(dataset.getId())).thenReturn(dataset);

  HttpResponse response = read("dataset/" + dataset.getId());
  assertThat(response.getStatusCode(), is(HttpResponseStatus.OK.code()));
  assertThat(response.getStatusMessage(), is(HttpResponseStatus.OK.reasonPhrase()));

  Dataset actual = viewTransformer.transform(Dataset.class, response.getPayload());
  assertThat(actual, is(dataset));
}

@Test
public void canCreateDataset(@Random Dataset dataset) {
  String name = "aName";
  String source = "aSource";
  String subscriptionControlField = "aSubscriptionControlField";
  String subscriptionControlFieldPattern = "aSubscriptionControlFieldPattern";

  when(datasetDao.write(any(Dataset.class))).thenReturn(dataset);

  String payload =
      "{ \"name\": \""
          + name
          + "\", \"source\": \""
          + source
          + "\", "
          + "\"subscriptionControlField\": \""
          + subscriptionControlField
          + "\", "
          + "\"subscriptionControlFieldPattern\": \""
          + subscriptionControlFieldPattern
          + "\" }";

  HttpResponse response = post("dataset", payload);
  assertThat(response.getStatusCode(), is(HttpResponseStatus.OK.code()));
  assertThat(response.getStatusMessage(), is(HttpResponseStatus.OK.reasonPhrase()));

  Dataset actual = viewTransformer.transform(Dataset.class, response.getPayload());
  assertThat(actual, is(dataset));

  ArgumentCaptor<Dataset> datasetCaptor = ArgumentCaptor.forClass(Dataset.class);
  verify(datasetDao).write(datasetCaptor.capture());

  Dataset submittedDataset = datasetCaptor.getValue();
  assertThat(submittedDataset.getName(), is(name));
  assertThat(submittedDataset.getOwner(), is(applicationConfiguration.getCannedUserName()));
  assertThat(submittedDataset.getSource(), is(source));
  assertThat(submittedDataset.getSubscriptionControlField(), is(subscriptionControlField));
  assertThat(
      submittedDataset.getSubscriptionControlFieldPattern(), is(subscriptionControlFieldPattern));
}

@Override
public void putResponseBodyInString(String payload, int retCode, String encoding) {

    // this.response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(retCode),
    // Unpooled.copiedBuffer(payload, CharsetUtil.UTF_8));
    this.response.setStatus(HttpResponseStatus.valueOf(retCode));

    this.response.content().writeBytes(Unpooled.copiedBuffer(payload, Charset.forName(encoding)));

    ctx.write(response);

    // write end marker
    ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE);
}

@Test
public void dodgyBodyWillBeRejectedAsABadRequestWhenAttemptingToCreateDataset() {
  String endpoint = "dataset";
  HttpResponse response = post(endpoint, "{ \"foo\": \"whatever\" }");
  assertThat(response.getStatusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
  assertThat(response.getStatusMessage(), is(HttpResponseStatus.BAD_REQUEST.reasonPhrase()));

  assertThatErrorResponseIsCorrect(
      response.getPayload(), "Failed to deserialise request body", endpoint, BAD_REQUEST);

  verify(datasetDao, never()).write(any(Dataset.class));
}

@Test
public void dodgyBodyWillBeRejectedAsABadRequestWhenAttemptingToUpdateDataset() {
  String endpoint = "dataset";
  HttpResponse response = put(endpoint, "{ \"foo\": \"whatever\" }");
  assertThat(response.getStatusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
  assertThat(response.getStatusMessage(), is(HttpResponseStatus.BAD_REQUEST.reasonPhrase()));

  assertThatErrorResponseIsCorrect(
      response.getPayload(), "Failed to deserialise request body", endpoint, BAD_REQUEST);

  verify(datasetDao, never()).write(any(Dataset.class));
}

private void assertThatErrorResponseIsCorrect(
    String response,
    String exceptionMessage,
    String endpoint,
    HttpResponseStatus httpResponseStatus) {
  JsonObject errorResponse = new JsonObject(response);

  assertThat(errorResponse.getString("timestamp"), notNullValue());
  assertThat(errorResponse.getInteger("statusCode"), is(httpResponseStatus.code()));
  assertThat(errorResponse.getString("statusMessage"), is(httpResponseStatus.reasonPhrase()));
  MatcherAssert.assertThat(
      errorResponse.getString("path"), Matchers.is(WebServerUtils.withApplicationName(endpoint)));
  assertThat(errorResponse.getString("message"), containsString(exceptionMessage));
  assertThat(errorResponse.getString("stackTrace"), notNullValue());
}

private boolean ifConnectSuccess(ChannelHandlerContext ctx, FullHttpRequest request) {
    if (!request.decoderResult().isSuccess()) {
        sendError(ctx, HttpResponseStatus.BAD_REQUEST);
        return false;
    }
    if (request.method() != HttpMethod.GET) {
        sendError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED);
        return false;
    }
    return true;
}

@Disabled
@Test
public void canLogout() {
  HttpResponse response = post("logout", null);

  assertThat(response.getStatusCode(), is(HttpResponseStatus.FORBIDDEN.code()));
  assertThat(response.getStatusMessage(), is(HttpResponseStatus.UNAUTHORIZED.reasonPhrase()));
}

/**
 * Reads the header part of response from remote HTTP server. Tests
 * the validity of this connection.
 * 
 * @param ctx handler context of this channel
 * @param msg received message
 */
public void channelRead(ChannelHandlerContext ctx, Object msg) {
  if (msg instanceof HttpResponse) {
    HttpResponse response = (HttpResponse) msg;
    HttpResponseStatus status = response.getStatus();
    try {
      if (HTTPResponseCode.isMoved(status)) {
        URI uri = URI.create(
            response.headers().get(HttpHeaders.Names.LOCATION));
        builder.setUri(URI.create(uri.endpoint()));
        code = ActionCode.Redirect;
        throw new HTTPException(builder.getHost() + " " + status.toString());
      } else if (HTTPResponseCode.isNotFound(status)) {
        code = ActionCode.NotFound;
        throw new HTTPException(builder.getHost() + " " + status.toString());
      } else if (HTTPResponseCode.isInvalid(status)) {
        code = ActionCode.Bad;
        throw new HTTPException(
            builder.getHost() + " HEADER method unsupported");
      } else if (HTTPResponseCode.isOK(status)) try {
        // Valid HTTP server found
        code = ActionCode.OK;
        builder.setKeepAlive(!response.headers()
            .get("Connection").equalsIgnoreCase("close"));
      } catch (NullPointerException npe) {
        // No connection header.
        builder.setKeepAlive(true);
      }
    } catch (HTTPException e) {
      System.err.println(e.getMessage());
    }
  }
  if (msg instanceof HttpContent) {
    endTest(ctx);
  }
}

@Test
public void testOptionsStatusCode() {
    serverRule.getHttpClient().createOptions("/files")
    .doOnNext(resp-> logger.info(resp.toString()))
    .map(resp -> resp.getStatus())
    .toBlocking()
    .forEach(status -> assertEquals(HttpResponseStatus.NO_CONTENT, status));
}

@Test
public void testStatus() {
    Iterable<HttpClientResponse<ByteBuf>> response = serverRule.getHttpClient()
            .createPost("/files")
            .addHeader("Tus-Resumable", "1.0.0")
            .addHeader("Upload-Length", "100")
            .toBlocking()
            .toIterable();
    Iterator<HttpClientResponse<ByteBuf>> it = response.iterator();
    assertTrue(it.hasNext());
    assertEquals(HttpResponseStatus.CREATED, it.next().getStatus());
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    System.err.println(cause.getMessage());
    //cause.printStackTrace();
    if (ctx.channel().isActive()) {
        sendError(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR);
    }
}

public static void sendResponse(Channel channel, HttpResponseStatus resultCode, ContentTypes contentType,
        ByteBuf result, boolean isClose) {
    channel.eventLoop().execute(() -> {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, resultCode, result);
        response.headers().add(HttpHeaderNames.CONTENT_TYPE, contentType.getValue());
        response.headers().add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes() + "");
        channel.writeAndFlush(response).addListener((ChannelFutureListener) arg0 -> {
            if (isClose) {
                arg0.channel().close();
            }
        });
    });
}