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

private boolean writeResponse(HttpObject currentObj, ChannelHandlerContext ctx) {
    // Decide whether to close the connection or not.
    boolean keepAlive = HttpHeaders.isKeepAlive(request);

    // Build the response object.
    FullHttpResponse response = new DefaultFullHttpResponse(
        HTTP_1_1, currentObj.getDecoderResult().isSuccess() ? OK : BAD_REQUEST,
        Unpooled.copiedBuffer(buf.toString(), CharsetUtil.UTF_8));

    response.headers().set(CONTENT_TYPE, "application/json");

    if (keepAlive) {
        // Add 'Content-Length' header only for a keep-alive connection.
        response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
        // Add keep alive header as per:
        // - http://www.w3.org/Protocols/HTTP/1.1/draft-ietf-http-v11-spec-01.html#Connection
        response.headers().set(CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
    }

    // Write the response.
    ctx.write(response);

    return keepAlive;
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    LOGGER.info("[Client ({})] => [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                msg);

    if (msg instanceof FullHttpRequest) {
        String streamId = ((HttpRequest) msg).headers().get(
                HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text());
        if (streamId == null) {
            throw new IllegalStateException("No streamId");
        }
        streams.offer(streamId);
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handle message: " + msg.getClass());
    }

    outboundChannel.writeAndFlush(msg);
}

private void debugRequestInfo(HttpObject httpObject, String key) {
    if (m_debugInfo && httpObject instanceof HttpRequest) {
        if (key != null) {
            LOGGER.debug("Cache Key: " + key);
        }
        if (httpObject instanceof FullHttpRequest) {
            FullHttpRequest req = (FullHttpRequest) httpObject;
            HttpHeaders headers = req.headers();
            LOGGER.debug("Headers:");
            for (Iterator<Entry<String, String>> it = headers.iterator(); it
                    .hasNext();) {
                Entry<String, String> entry = it.next();
                LOGGER.debug("\t" + entry.getKey() + ":\t"
                        + entry.getValue());
            }
            ByteBuf content = req.content();
            int length = content.readableBytes();
            LOGGER.debug("Content Length: " + length);
            if (length != 0) {
                LOGGER.debug("Content: "
                        + content.toString(Charset.forName("UTF-8")));
            }
        }
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx, HttpObject msg)
        throws Exception {
    if (msg instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) msg;

        LOGGER.info("Response status: " + response.getStatus());
        if (response.getStatus().equals(OK)) {
            LOGGER.info("Operation is successful");
        } else {
            LOGGER.error("Operation is failed");
        }
    }

    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;

        System.out.print(content.content().toString(CharsetUtil.UTF_8));
        System.out.flush();

    }

}

@Test
public void doChannelRead_checks_for_fully_send_responses_but_does_nothing_else_if_msg_is_not_HttpRequest_or_HttpContent() {
    // given
    HttpObject msgMock = mock(HttpObject.class);

    // when
    PipelineContinuationBehavior result = handler.doChannelRead(ctxMock, msgMock);

    // then
    verify(ctxMock).channel();
    verifyNoMoreInteractions(ctxMock);
    verify(stateMock).isResponseSendingLastChunkSent();
    verifyNoMoreInteractions(stateMock);
    verifyNoMoreInteractions(msgMock);
    assertThat(result).isEqualTo(PipelineContinuationBehavior.CONTINUE);
}

@Test
public void doChannelRead_does_nothing_if_msg_is_not_HttpRequest() {
    // given
    String pathTemplate = "/some/path/with/{id}";
    Collection<String> pathTemplates = new ArrayList<String>() {{ add(pathTemplate); }};
    doReturn(pathTemplates).when(matcherMock).matchingPathTemplates();
    HttpObject msg = mock(HttpObject.class);

    // when
    PipelineContinuationBehavior result = handlerSpy.doChannelRead(ctxMock, msg);

    // then
    verify(handlerSpy).doChannelRead(ctxMock, msg);
    verifyNoMoreInteractions(handlerSpy);
    verifyNoMoreInteractions(requestInfoMock);
    verifyNoMoreInteractions(stateMock);
    assertThat(result).isEqualTo(PipelineContinuationBehavior.CONTINUE);
}

public static Bootstrap createNettyHttpClientBootstrap() {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(new NioEventLoopGroup())
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new HttpClientCodec());
                     p.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
                     p.addLast("clientResponseHandler", new SimpleChannelInboundHandler<HttpObject>() {
                         @Override
                         protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
                             throw new RuntimeException("Client response handler was not setup before the call");
                         }
                     });
                 }
             });

    return bootstrap;
}

public static CompletableFuture<NettyHttpClientResponse> setupNettyHttpClientResponseHandler(
    Channel ch, Consumer<ChannelPipeline> pipelineAdjuster
) {
    CompletableFuture<NettyHttpClientResponse> responseFromServerFuture = new CompletableFuture<>();
    ch.pipeline().replace("clientResponseHandler", "clientResponseHandler", new SimpleChannelInboundHandler<HttpObject>() {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg)
            throws Exception {
            if (msg instanceof FullHttpResponse) {
                // Store the proxyServer response for asserting on later.
                responseFromServerFuture.complete(new NettyHttpClientResponse((FullHttpResponse) msg));
            } else {
                // Should never happen.
                throw new RuntimeException("Received unexpected message type: " + msg.getClass());
            }
        }
    });

    if (pipelineAdjuster != null)
        pipelineAdjuster.accept(ch.pipeline());

    return responseFromServerFuture;
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (httpObject instanceof HttpRequest) {
        this.httpRequest = (HttpRequest) httpObject;
    }

    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;

        storeRequestContent(httpContent);

        if (httpContent instanceof LastHttpContent) {
            LastHttpContent lastHttpContent = (LastHttpContent) httpContent;
            trailingHeaders = lastHttpContent .trailingHeaders();
        }
    }

    return null;
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (httpObject instanceof HttpRequest) {
        HttpRequest httpRequest = (HttpRequest) httpObject;

        if (ProxyUtils.isCONNECT(httpRequest)) {
            Attribute<String> hostname = ctx.attr(AttributeKey.<String>valueOf(HttpsAwareFiltersAdapter.HOST_ATTRIBUTE_NAME));
            String hostAndPort = httpRequest.getUri();

            // CONNECT requests contain the port, even when using the default port. a sensible default is to remove the
            // default port, since in most cases it is not explicitly specified and its presence (in a HAR file, for example)
            // would be unexpected.
            String hostNoDefaultPort = BrowserMobHttpUtil.removeMatchingPort(hostAndPort, 443);
            hostname.set(hostNoDefaultPort);
        }
    }

    return null;
}

@Override
public HttpObject serverToProxyResponse(HttpObject httpObject) {
    HttpObject processedHttpObject = httpObject;

    for (HttpFilters filter : filters) {
        try {
            processedHttpObject = filter.serverToProxyResponse(processedHttpObject);
            if (processedHttpObject == null) {
                return null;
            }
        } catch (RuntimeException e) {
            log.warn("Filter in filter chain threw exception. Filter method may have been aborted.", e);
        }
    }

    return processedHttpObject;
}

@Override
public HttpObject proxyToClientResponse(HttpObject httpObject) {
    HttpObject processedHttpObject = httpObject;
    for (HttpFilters filter : filters) {
        try {
            processedHttpObject = filter.proxyToClientResponse(processedHttpObject);
            if (processedHttpObject == null) {
                return null;
            }
        } catch (RuntimeException e) {
            log.warn("Filter in filter chain threw exception. Filter method may have been aborted.", e);
        }
    }

    return processedHttpObject;
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (httpObject instanceof HttpRequest) {
        HttpRequest httpRequest = (HttpRequest) httpObject;

        String url = getFullUrl(httpRequest);

        for (BlacklistEntry entry : blacklistedUrls) {
            if (HttpMethod.CONNECT.equals(httpRequest.getMethod()) && entry.getHttpMethodPattern() == null) {
                // do not allow CONNECTs to be blacklisted unless a method pattern is explicitly specified
                continue;
            }

            if (entry.matches(url, httpRequest.getMethod().name())) {
                HttpResponseStatus status = HttpResponseStatus.valueOf(entry.getStatusCode());
                HttpResponse resp = new DefaultFullHttpResponse(httpRequest.getProtocolVersion(), status);
                HttpHeaders.setContentLength(resp, 0L);

                return resp;
            }
        }
    }

    return null;
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (credentialsByHostname.isEmpty()) {
        return null;
    }

    if (httpObject instanceof HttpRequest) {
        HttpRequest httpRequest = (HttpRequest) httpObject;

        // providing authorization during a CONNECT is generally not useful
        if (ProxyUtils.isCONNECT(httpRequest)) {
            return null;
        }

        String hostname = getHost(httpRequest);

        // if there is an entry in the credentials map matching this hostname, add the credentials to the request
        String base64CredentialsForHostname = credentialsByHostname.get(hostname);
        if (base64CredentialsForHostname != null) {
            httpRequest.headers().add(HttpHeaders.Names.AUTHORIZATION, "Basic " + base64CredentialsForHostname);
        }
    }

    return null;
}

@Override
@SuppressWarnings("unchecked")
public void start() {
    List<Handler<HttpObject>> httpHandlerList = (List<Handler<HttpObject>>) getHandlerList();
    DefaultHttpProxyServer.bootstrap()
            .withPort(port).withFiltersSource(new HttpFiltersSourceAdapter() {
        @Override
        public HttpFilters filterRequest(HttpRequest originalRequest) {
            return new HttpFiltersAdapter(originalRequest) {
                @Override
                public HttpResponse clientToProxyRequest(HttpObject httpObject) {
                    httpHandlerList.forEach(httpObjectHandler -> httpObjectHandler.client2Proxy(httpObject));
                    return null;
                }
            };
        }
    }).start();
}

@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject)
    throws Exception {
  // initial request
  if (LOG.isDebugEnabled()) {
    LOG.debug(String.format("%s: Reading from client %s", System.currentTimeMillis(), httpObject));
  }
  if (httpObject instanceof HttpRequest) {
    HttpRequest initialRequest = (HttpRequest) httpObject;
    if (_channelHandlerDelegate == null) {
      _channelHandlerDelegate =
          HandlerDelegateFactory.create(initialRequest, _channelMediator, _connectionFlowRegistry);
      _channelHandlerDelegate.onCreate();
    }
  }
  _channelHandlerDelegate.onRead(httpObject);
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg)
        throws Exception {
    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;
        RandomAccessFile output = null;
        FileChannel oc = null;
        try {
            output = new RandomAccessFile(file, "rw");
            oc = output.getChannel();
            oc.position(oc.size());
            ByteBuf buffer = content.content();
            for (int i = 0, len = buffer.nioBufferCount(); i < len; i++) {
                oc.write(buffer.nioBuffers()[i]);
            }
        } finally {
            IOUtils.closeQuietly(oc);
            IOUtils.closeQuietly(output);
        }
        if (content instanceof LastHttpContent) {
            ctx.close();
        }
    }
}

private void startProxyServerWithFilterAnsweringStatusCode(int statusCode) {
    final HttpResponseStatus status = HttpResponseStatus.valueOf(statusCode);
    HttpFiltersSource filtersSource = new HttpFiltersSourceAdapter() {
        @Override
        public HttpFilters filterRequest(HttpRequest originalRequest) {
            return new HttpFiltersAdapter(originalRequest) {
                @Override
                public HttpResponse clientToProxyRequest(HttpObject httpObject) {
                    return new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
                }
            };
        }
    };

    proxyServer = DefaultHttpProxyServer.bootstrap()
            .withPort(0)
            .withFiltersSource(filtersSource)
            .start();
}

@Override
public void handleReadFromClient(ChannelMediator channelMediator, HttpObject httpObject) {
  if (channelMediator == null) {
    throw new IllegalStateException("HRFC: ChannelMediator can't be null");
  }

  try {
    if (httpObject instanceof HttpRequest) {
      HttpRequest httpRequest = (HttpRequest) httpObject;
      _clientRequestBuilder.interpretHttpRequest(httpRequest);
      _clientRequestBuilder.addHeaders(httpRequest);
    }

    if (httpObject instanceof HttpContent) {
      _clientRequestBuilder.appendHttpContent((HttpContent) httpObject);
    }
  } catch (IOException e) {
    throw new RuntimeException("HRFC: Failed to record HttpContent", e);
  }

  channelMediator.writeToServer(httpObject);
}

@Override
public void onRead(HttpObject httpObject) {
  if (!_connectionFlowProcessor.isComplete()) {
    _channelReadCallback.write(httpObject);
    // Accroding to http://netty.io/wiki/reference-counted-objects.html
    // When an event loop reads data into a ByteBuf and triggers a channelRead() event with it,
    // it is the responsibility of the ChannelHandler in the corresponding pipeline to release the buffer.
    // Since this is the last ChannelHandler, it release the reference-counted after read. So we need to
    // retain to make sure it will not be released until we stored in scene.
    if(httpObject instanceof HttpContent){
      ((HttpContent)httpObject).retain();
    }
    return;
  }
  _channelMediator.readFromClientChannel(httpObject);
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (httpObject instanceof HttpRequest) {
        System.err.println("--->" + ((HttpRequest) httpObject).getUri());
    }

    return super.clientToProxyRequest(httpObject);
}

@Override protected void channelRead0(
    ChannelHandlerContext context, HttpObject message) throws Exception {
  if (message instanceof HttpResponse) {
    receive((HttpResponse) message);
  }
  if (message instanceof HttpContent) {
    receive((HttpContent) message);
    if (message instanceof LastHttpContent) {
      release(this);
    }
  }
}

@Override protected void channelRead0(
    ChannelHandlerContext context, HttpObject message) throws Exception {
  if (message instanceof HttpResponse) {
    receive((HttpResponse) message);
  }
  if (message instanceof HttpContent) {
    receive((HttpContent) message);
    if (message instanceof LastHttpContent) {
      release(this);
    }
  }
}

public HttpProxyServer start() {

        return DefaultHttpProxyServer.bootstrap()
            .withAddress(new InetSocketAddress(host, port))
            .withFiltersSource(new HttpFiltersSourceAdapter() {

                @Override
                public HttpFilters filterRequest(HttpRequest originalRequest) {

                    return new HttpFiltersAdapter(originalRequest) {

                        final String mockedAddress = mockServer + ":" + mockPort;

                        @Override
                        public HttpResponse clientToProxyRequest(HttpObject httpObject) {

                            final HttpRequest request = (HttpRequest) httpObject;

                            if (request.getUri().contains("google")) {
                                request.setUri(mockedAddress);
                            }

                            super.clientToProxyRequest(request);

                            return null;
                        }
                    };
                }
            }).start();
    }

@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject)
        throws Exception {
    LOGGER.info("[Client ({})] <= [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                httpObject);

    outboundChannel.writeAndFlush(ReferenceCountUtil.retain(httpObject));

    if (httpObject instanceof HttpResponse) {
        currentRequest = null;
        delayOutboundHandler.next();
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    if (msg instanceof FullHttpRequest) {
        LOGGER.info("[Client ({})] => [Server ({})] : (PENDING) {}",
                    connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                    msg);
        HttpRequest request = (HttpRequest) msg;
        pendings.offer(new RequestPromise(request, promise));
        next();
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handled message: " + msg.getClass());
    } else {
        ctx.write(msg, promise);
    }
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    LOGGER.info("[Client ({})] => [Server ({})] : {}",
                connectionInfo.getClientAddr(), connectionInfo.getServerAddr(),
                msg);
    if (msg instanceof FullHttpRequest) {
        HttpMessage httpMessage = (HttpRequest) msg;
        httpMessage.headers().add(ExtensionHeaderNames.SCHEME.text(), "https");
    } else if (msg instanceof HttpObject) {
        throw new IllegalStateException("Cannot handle message: " + msg.getClass());
    }

    ctx.writeAndFlush(msg, promise);
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (httpObject instanceof HttpRequest) {
        System.err.println("--->" + ((HttpRequest) httpObject).getUri());
    }

    return super.clientToProxyRequest(httpObject);
}

@Override
public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) {
    if (LOGGER.isDebugEnabled()) {
        if (msg instanceof HttpResponse) {
            HttpResponse response = (HttpResponse) msg;
            LOGGER.debug("STATUS: " + response.getStatus());
            LOGGER.debug("VERSION: " + response.getProtocolVersion());
            if (!response.headers().isEmpty()) {
                for (String name : response.headers().names()) {
                    for (String value : response.headers().getAll(name)) {
                        LOGGER.debug("HEADER: " + name + " = " + value);
                    }
                }
            }
            if (HttpHeaders.isTransferEncodingChunked(response)) {
                LOGGER.debug("CHUNKED CONTENT {");
            } else {
                LOGGER.debug("CONTENT {");
            }
        }
        if (msg instanceof HttpContent) {
            HttpContent content = (HttpContent) msg;
            LOGGER.debug(content.content().toString(CharsetUtil.UTF_8));
            if (content instanceof LastHttpContent) {
                LOGGER.debug("} END OF CONTENT");
            }
        }
    }
}

@Override
public HttpObject serverToProxyResponse(HttpObject httpObject) {
    if (httpObject instanceof FullHttpResponse) {
        m_debugManager.debugResponse((FullHttpResponse) httpObject, this.m_ctx);
    }
    return httpObject;
}

public void init() {
    m_execService = Executors.newCachedThreadPool();
    m_scheduledService = Executors.newScheduledThreadPool(1);

    RequestKeyGenerator keyGenerator = new RequestKeyGenerator(
            m_appConfig);

    /*
     * m_policyManager and keyGenerator are used in the initialization of
     * NettyRequestProxyFilter, NettyResponseProxyFilter, please make sure
     * init this two first
     */
    m_policyManager = new DefaultPolicyManager(m_appConfig, keyGenerator,
            m_scheduledService);

    m_cacheManager = m_policyManager.getCacheManager();

    m_debugManager = new DebugManager(m_appConfig, m_policyManager);

    List<IHttpRequestProxyFilter<HttpObject, ChannelHandlerContext>> requestFilters = new ArrayList<>();
    requestFilters.add(new NettyRequestProxyFilter(m_policyManager,
            m_appConfig));
    List<IHttpResponseProxyFilter<HttpObject, ChannelHandlerContext>> responseFilters = new ArrayList<>();
    responseFilters.add(new NettyResponseProxyFilter(m_policyManager,
            m_execService));

    m_filterManager = new FilterManager<>(requestFilters, responseFilters);

}

public CacheHttpFilter(HttpRequest originalRequest,
        ChannelHandlerContext ctx, FilterManager<HttpObject, HttpObject, ChannelHandlerContext> filterManager, DebugManager debugManager) {
    super(originalRequest);
    this.m_ctx = ctx;
    m_filterManager = filterManager;
    m_debugManager = debugManager;
}

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    HttpResponse response = m_filterManager.filterRequest(
            httpObject, this.m_ctx);
    if (response != null) {
        m_debugManager.issueDebugRequest(
                (FullHttpRequest) httpObject, this.m_ctx, false);
    }
    return response;
}

@Test
public void testServerToProxyResponse() {
    FullHttpResponse response = Mockito.mock(FullHttpResponse.class);
    HttpObject obj = debugHttpFilter.serverToProxyResponse(response);
    Assert.assertEquals(obj, response);
    Mockito.verify(m_debugManager, Mockito.times(1)).debugResponse(response, m_ctx);
}

@Test
public void testGetHttpResponse() {
    List<HttpObject> list = new ArrayList<>();
    list.add(mock(HttpResponse.class));

    assertNotNull(new ChunkResponsesSerializer().getHttpResponse(list));
}

@Test
public void getContent(){
    List<HttpObject> list = new ArrayList<>();
    list.add(mock(HttpResponse.class));
    list.add(mock(HttpContent.class));
    list.add(mock(LastHttpContent.class));
    list.add(mock(FullHttpResponse.class));

    ByteBuf[] content = new ChunkResponsesSerializer().getContent(list);
    assertEquals(3, content.length);
}

@Test
public void testClientToProxyRequestHttpObject() {
    HttpObject httpObject = mock(FullHttpRequest.class);
    when(m_filterManager.filterRequest(httpObject, m_ctx)).thenReturn(null);
    assertNull(m_filter.clientToProxyRequest(httpObject));

    when(m_filterManager.filterRequest(httpObject, m_ctx)).thenReturn(mock(HttpResponse.class));
    assertNotNull(m_filter.clientToProxyRequest(httpObject));
}

@Test
    public void testServerToProxyResponseHttpObject() {
        assertNull(m_filter.serverToProxyResponse(null));
        HttpObject httpObject = mock(HttpObject.class);
//      assertEquals(httpObject, m_filter.serverToProxyResponse(httpObject));

        FullHttpResponse fullHttpResponse = mock(FullHttpResponse.class);
        when(m_filterManager.filterResponse(fullHttpResponse, m_ctx))
                .thenReturn(mock(HttpResponse.class));
        assertTrue(m_filter.serverToProxyResponse(fullHttpResponse) instanceof HttpResponse);
    }

@Override protected void channelRead0(
    ChannelHandlerContext context, HttpObject message) throws Exception {
  if (message instanceof HttpResponse) {
    receive((HttpResponse) message);
  }
  if (message instanceof HttpContent) {
    receive((HttpContent) message);
    if (message instanceof LastHttpContent) {
      release(this);
    }
  }
}

@Override protected void channelRead0(
    ChannelHandlerContext context, HttpObject message) throws Exception {
  if (message instanceof HttpResponse) {
    receive((HttpResponse) message);
  }
  if (message instanceof HttpContent) {
    receive((HttpContent) message);
    if (message instanceof LastHttpContent) {
      release(this);
    }
  }
}