Java 类org.eclipse.jetty.server.HttpChannel 实例源码

@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    MethodInterceptResult result) throws Throwable {
    HttpChannel httpChannel = (HttpChannel)allArguments[0];
    HttpServletRequest servletRequest = httpChannel.getRequest();

    ContextCarrier contextCarrier = new ContextCarrier();

    CarrierItem next = contextCarrier.items();
    while (next.hasNext()) {
        next = next.next();
        next.setHeadValue(servletRequest.getHeader(next.getHeadKey()));
    }

    AbstractSpan span = ContextManager.createEntrySpan(servletRequest.getRequestURI(), contextCarrier);
    Tags.URL.set(span, servletRequest.getRequestURL().toString());
    Tags.HTTP.METHOD.set(span, servletRequest.getMethod());
    span.setComponent(ComponentsDefine.JETTY_SERVER);
    SpanLayer.asHttp(span);
}

@Test
public void shouldSetSecureSiteURLWhenSiteUrlIsConfigured() throws URISyntaxException {
    final ServerConfigService serverConfigService = mock(ServerConfigService.class);
    when(serverConfigService.hasAnyUrlConfigured()).thenReturn(true);
    when(serverConfigService.siteUrlFor("http://url/go/admin?tab=oAuth", true)).thenReturn("https://url/go/admin?tab=oAuth");

    Request request = new Request(mock(HttpChannel.class), mock(HttpInput.class));
    request.setUri(new HttpURI("/go/admin?tab=oAuth"));
    request.setServerName("url");

    DeploymentContextWriter writer = new DeploymentContextWriter() {
        @Override protected BaseUrlProvider getBaseUrlProvider(HttpServletRequest req) {
            return serverConfigService;
        }
    };
    writer.writeSecureSiteUrl(request);
    assertThat(request.getAttribute("secure_site"), is("https://url/go/admin?tab=oAuth"));
    assertThat(request.getAttribute("force_ssl"), is("true"));
}

@Override
public HttpResponse serve(ServiceRequestContext ctx, HttpRequest req) throws Exception {
    final ArmeriaConnector connector = this.connector;

    final HttpResponseWriter res = HttpResponse.streaming();

    req.aggregate().handle(voidFunction((aReq, cause) -> {
        if (cause != null) {
            logger.warn("{} Failed to aggregate a request:", ctx, cause);
            res.close(HttpHeaders.of(HttpStatus.INTERNAL_SERVER_ERROR));
            return;
        }

        boolean success = false;
        try {
            final ArmeriaHttpTransport transport = new ArmeriaHttpTransport(req.method());
            final HttpChannel httpChannel = new HttpChannel(
                    connector,
                    connector.getHttpConfiguration(),
                    new ArmeriaEndPoint(hostname, connector.getScheduler(),
                                        ctx.localAddress(), ctx.remoteAddress()),
                    transport);

            fillRequest(ctx, aReq, httpChannel.getRequest());

            ctx.blockingTaskExecutor().execute(() -> invoke(ctx, res, transport, httpChannel));
            success = true;
        } finally {
            if (!success) {
                res.close();
            }
        }
    })).exceptionally(CompletionActions::log);

    return res;
}

private void invoke(ServiceRequestContext ctx, HttpResponseWriter res,
                    ArmeriaHttpTransport transport, HttpChannel httpChannel) {

    final Queue<HttpData> out = transport.out;
    try {
        server.handle(httpChannel);
        httpChannel.getResponse().getHttpOutput().flush();

        final Throwable cause = transport.cause;
        if (cause != null) {
            throw cause;
        }

        final HttpHeaders headers = toResponseHeaders(transport);
        res.write(headers);
        for (;;) {
            final HttpData data = out.poll();
            if (data == null || !res.write(data)) {
                break;
            }
        }
        res.close();
    } catch (Throwable t) {
        logger.warn("{} Failed to produce a response:", ctx, t);
        res.close();
    }
}

@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    Object ret) throws Throwable {
    HttpChannel httpChannel = (HttpChannel)allArguments[0];
    HttpServletResponse servletResponse = httpChannel.getResponse();
    AbstractSpan span = ContextManager.activeSpan();
    if (servletResponse.getStatus() >= 400) {
        span.errorOccurred();
        Tags.STATUS_CODE.set(span, Integer.toString(servletResponse.getStatus()));
    }
    ContextManager.stopSpan();
    return ret;
}

@Override
protected Request getRequest(final Object[] args) {
    if (args == null || args.length < 1) {
        return null;
    }

    if (args[0] instanceof HttpChannel) {
        final HttpChannel<?> channel = (HttpChannel<?>) args[0];
        return channel.getRequest();
    }
    return null;
}

@Override
protected void handshake(final Runnable handler) throws Exception {
  /** Infinite timeout because the continuation is never resumed but only completed on close. */
  req.getAsyncContext().setTimeout(0L);
  /** Server sent events headers. */
  rsp.setStatus(HttpServletResponse.SC_OK);
  rsp.setHeader("Connection", "Close");
  rsp.setContentType("text/event-stream; charset=utf-8");
  rsp.flushBuffer();

  HttpChannel channel = rsp.getHttpChannel();
  Connector connector = channel.getConnector();
  Executor executor = connector.getExecutor();
  executor.execute(handler);
}

@Test
public void handshake() throws Exception {
  new MockUnit(Request.class, Response.class, HttpOutput.class, Runnable.class,
    AsyncContext.class, HttpChannel.class, Connector.class, Executor.class)
    .expect(httpOutput)
    .expect(unit -> {
      AsyncContext async = unit.get(AsyncContext.class);
      async.setTimeout(0L);

      Request req = unit.get(Request.class);
      expect(req.getAsyncContext()).andReturn(async);
    })
    .expect(unit -> {
      Response rsp = unit.get(Response.class);
      rsp.setStatus(200);
      rsp.setHeader("Connection", "Close");
      rsp.setContentType("text/event-stream; charset=utf-8");
      rsp.flushBuffer();

      HttpChannel channel = unit.get(HttpChannel.class);
      expect(rsp.getHttpChannel()).andReturn(channel);

      Connector connector = unit.get(Connector.class);
      expect(channel.getConnector()).andReturn(connector);

      Executor executor = unit.get(Executor.class);
      expect(connector.getExecutor()).andReturn(executor);

      executor.execute(unit.get(Runnable.class));
    })
    .run(unit -> {
      new JettySse(unit.get(Request.class), unit.get(Response.class))
        .handshake(unit.get(Runnable.class));
    });
}

/**
 * Handler method converting a Jetty HttpChannel into a Restlet Call.
 * 
 * @param channel
 *        The channel to handle.
 */
@Override
public void handle( HttpChannel channel ) throws IOException, ServletException
{
    try
    {
        helper.handle( new JettyServerCall( helper.getHelped(), channel, ensureHostHeader ) );
    }
    catch( Throwable e )
    {
        channel.getEndPoint().close();
        throw new IOException( "Restlet exception", e );
    }
}

@Override
public void handleAsync( HttpChannel channel ) throws IOException, ServletException
{
    // TODO: should we handle async differently?
    handle( channel );
}

/**
 * Constructor.
 * 
 * @param server
 *        The parent server.
 * @param channel
 *        The wrapped Jetty HTTP channel.
 * @param ensureHostHeader
 *        Whether to generate a Host header if not provided by the request
 */
public JettyServerCall( Server server, HttpChannel channel, boolean ensureHostHeader )
{
    super( server );
    this.channel = channel;
    this.ensureHostHeader = ensureHostHeader;
}

/**
 * Returns the wrapped Jetty HTTP channel.
 * 
 * @return The wrapped Jetty HTTP channel.
 */
public HttpChannel getChannel()
{
    return channel;
}