io.netty.handler.codec.http2.DefaultHttp2Connection #io.netty.handler.codec.http2.DefaultHttp2FrameReader源码实例Demo

static NettyClientHandler newHandler(
    ClientTransportLifecycleManager lifecycleManager,
    @Nullable KeepAliveManager keepAliveManager,
    int flowControlWindow,
    int maxHeaderListSize,
    Supplier<Stopwatch> stopwatchFactory,
    Runnable tooManyPingsRunnable,
    TransportTracer transportTracer,
    Attributes eagAttributes,
    String authority) {
  Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive");
  Http2HeadersDecoder headersDecoder = new GrpcHttp2ClientHeadersDecoder(maxHeaderListSize);
  Http2FrameReader frameReader = new DefaultHttp2FrameReader(headersDecoder);
  Http2FrameWriter frameWriter = new DefaultHttp2FrameWriter();
  Http2Connection connection = new DefaultHttp2Connection(false);
  WeightedFairQueueByteDistributor dist = new WeightedFairQueueByteDistributor(connection);
  dist.allocationQuantum(16 * 1024); // Make benchmarks fast again.
  DefaultHttp2RemoteFlowController controller =
      new DefaultHttp2RemoteFlowController(connection, dist);
  connection.remote().flowController(controller);

  return newHandler(
      connection,
      frameReader,
      frameWriter,
      lifecycleManager,
      keepAliveManager,
      flowControlWindow,
      maxHeaderListSize,
      stopwatchFactory,
      tooManyPingsRunnable,
      transportTracer,
      eagAttributes,
      authority);
}
 

/**
 * Must be called by subclasses to initialize the handler and channel.
 */
protected final void initChannel(Http2HeadersDecoder headersDecoder) throws Exception {
  content = Unpooled.copiedBuffer("hello world", UTF_8);
  frameWriter = mock(Http2FrameWriter.class, delegatesTo(new DefaultHttp2FrameWriter()));
  frameReader = new DefaultHttp2FrameReader(headersDecoder);

  channel = new FakeClockSupportedChanel();
  handler = newHandler();
  channel.pipeline().addLast(handler);
  ctx = channel.pipeline().context(handler);

  writeQueue = initWriteQueue();
}
 

private CapturingGrpcHttp2ConnectionHandler capturingGrpcHandler() {
  // Netty Boilerplate.  We don't really need any of this, but there is a tight coupling
  // between a Http2ConnectionHandler and its dependencies.
  Http2Connection connection = new DefaultHttp2Connection(true);
  Http2FrameWriter frameWriter = new DefaultHttp2FrameWriter();
  Http2FrameReader frameReader = new DefaultHttp2FrameReader(false);
  DefaultHttp2ConnectionEncoder encoder =
      new DefaultHttp2ConnectionEncoder(connection, frameWriter);
  DefaultHttp2ConnectionDecoder decoder =
      new DefaultHttp2ConnectionDecoder(connection, encoder, frameReader);

  return new CapturingGrpcHttp2ConnectionHandler(decoder, encoder, new Http2Settings());
}
 

static FakeGrpcHttp2ConnectionHandler newHandler() {
  DefaultHttp2Connection conn = new DefaultHttp2Connection(/*server=*/ false);
  DefaultHttp2ConnectionEncoder encoder =
      new DefaultHttp2ConnectionEncoder(conn, new DefaultHttp2FrameWriter());
  DefaultHttp2ConnectionDecoder decoder =
      new DefaultHttp2ConnectionDecoder(conn, encoder, new DefaultHttp2FrameReader());
  Http2Settings settings = new Http2Settings();
  return new FakeGrpcHttp2ConnectionHandler(
      /*channelUnused=*/ null, decoder, encoder, settings);
}
 

static NettyClientHandler newHandler(
    ClientTransportLifecycleManager lifecycleManager,
    @Nullable KeepAliveManager keepAliveManager,
    boolean autoFlowControl,
    int flowControlWindow,
    int maxHeaderListSize,
    Supplier<Stopwatch> stopwatchFactory,
    Runnable tooManyPingsRunnable,
    TransportTracer transportTracer,
    Attributes eagAttributes,
    String authority) {
  Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive");
  Http2HeadersDecoder headersDecoder = new GrpcHttp2ClientHeadersDecoder(maxHeaderListSize);
  Http2FrameReader frameReader = new DefaultHttp2FrameReader(headersDecoder);
  Http2FrameWriter frameWriter = new DefaultHttp2FrameWriter();
  Http2Connection connection = new DefaultHttp2Connection(false);
  WeightedFairQueueByteDistributor dist = new WeightedFairQueueByteDistributor(connection);
  dist.allocationQuantum(16 * 1024); // Make benchmarks fast again.
  DefaultHttp2RemoteFlowController controller =
      new DefaultHttp2RemoteFlowController(connection, dist);
  connection.remote().flowController(controller);

  return newHandler(
      connection,
      frameReader,
      frameWriter,
      lifecycleManager,
      keepAliveManager,
      autoFlowControl,
      flowControlWindow,
      maxHeaderListSize,
      stopwatchFactory,
      tooManyPingsRunnable,
      transportTracer,
      eagAttributes,
      authority);
}
 

/**
 * Must be called by subclasses to initialize the handler and channel.
 */
protected final void initChannel(Http2HeadersDecoder headersDecoder) throws Exception {
  content = Unpooled.copiedBuffer("hello world", UTF_8);
  frameWriter = mock(Http2FrameWriter.class, delegatesTo(new DefaultHttp2FrameWriter()));
  frameReader = new DefaultHttp2FrameReader(headersDecoder);

  channel = new FakeClockSupportedChanel();
  handler = newHandler();
  channel.pipeline().addLast(handler);
  ctx = channel.pipeline().context(handler);

  writeQueue = initWriteQueue();
}
 

private static FakeGrpcHttp2ConnectionHandler newHandler(boolean noop) {
  DefaultHttp2Connection conn = new DefaultHttp2Connection(/*server=*/ false);
  DefaultHttp2ConnectionEncoder encoder =
      new DefaultHttp2ConnectionEncoder(conn, new DefaultHttp2FrameWriter());
  DefaultHttp2ConnectionDecoder decoder =
      new DefaultHttp2ConnectionDecoder(conn, encoder, new DefaultHttp2FrameReader());
  Http2Settings settings = new Http2Settings();
  return new FakeGrpcHttp2ConnectionHandler(
      /*channelUnused=*/ null, decoder, encoder, settings, noop);
}
 

private CapturingGrpcHttp2ConnectionHandler capturingGrpcHandler() {
  // Netty Boilerplate.  We don't really need any of this, but there is a tight coupling
  // between an Http2ConnectionHandler and its dependencies.
  Http2Connection connection = new DefaultHttp2Connection(true);
  Http2FrameWriter frameWriter = new DefaultHttp2FrameWriter();
  Http2FrameReader frameReader = new DefaultHttp2FrameReader(false);
  DefaultHttp2ConnectionEncoder encoder =
      new DefaultHttp2ConnectionEncoder(connection, frameWriter);
  DefaultHttp2ConnectionDecoder decoder =
      new DefaultHttp2ConnectionDecoder(connection, encoder, frameReader);

  return new CapturingGrpcHttp2ConnectionHandler(decoder, encoder, new Http2Settings());
}
 

static NettyServerHandler newHandler(
    ServerTransportListener transportListener,
    ChannelPromise channelUnused,
    List<ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer transportTracer,
    int maxStreams,
    int flowControlWindow,
    int maxHeaderListSize,
    int maxMessageSize,
    long keepAliveTimeInNanos,
    long keepAliveTimeoutInNanos,
    long maxConnectionIdleInNanos,
    long maxConnectionAgeInNanos,
    long maxConnectionAgeGraceInNanos,
    boolean permitKeepAliveWithoutCalls,
    long permitKeepAliveTimeInNanos) {
  Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive");
  Http2FrameLogger frameLogger = new Http2FrameLogger(LogLevel.DEBUG, NettyServerHandler.class);
  Http2HeadersDecoder headersDecoder = new GrpcHttp2ServerHeadersDecoder(maxHeaderListSize);
  Http2FrameReader frameReader = new Http2InboundFrameLogger(
      new DefaultHttp2FrameReader(headersDecoder), frameLogger);
  Http2FrameWriter frameWriter =
      new Http2OutboundFrameLogger(new DefaultHttp2FrameWriter(), frameLogger);
  return newHandler(
      channelUnused,
      frameReader,
      frameWriter,
      transportListener,
      streamTracerFactories,
      transportTracer,
      maxStreams,
      flowControlWindow,
      maxHeaderListSize,
      maxMessageSize,
      keepAliveTimeInNanos,
      keepAliveTimeoutInNanos,
      maxConnectionIdleInNanos,
      maxConnectionAgeInNanos,
      maxConnectionAgeGraceInNanos,
      permitKeepAliveWithoutCalls,
      permitKeepAliveTimeInNanos);
}
 

private Http2FrameReader frameReader() {
    return new Http2InboundFrameLogger(new DefaultHttp2FrameReader(), logger);
}
 

@Test
public void maxFrameSize() throws Exception {
    try (ServerSocket ss = new ServerSocket(0);
         ClientFactory clientFactory =
                 ClientFactory.builder()
                              .useHttp2Preface(true)
                              // Set the window size to the HTTP/2 default values to simplify the traffic.
                              .http2InitialConnectionWindowSize(Http2CodecUtil.DEFAULT_WINDOW_SIZE)
                              .http2InitialStreamWindowSize(Http2CodecUtil.DEFAULT_WINDOW_SIZE)
                              .http2MaxFrameSize(DEFAULT_MAX_FRAME_SIZE * 2) // == 16384 * 2
                              .build()) {

        final int port = ss.getLocalPort();
        final WebClient client = WebClient.builder("http://127.0.0.1:" + port)
                                          .factory(clientFactory)
                                          .build();
        client.get("/").aggregate();

        try (Socket s = ss.accept()) {
            final InputStream in = s.getInputStream();
            final BufferedOutputStream bos = new BufferedOutputStream(s.getOutputStream());

            readBytes(in, connectionPrefaceBuf().capacity()); // Read the connection preface and discard it.

            // Read a SETTINGS frame and validate it.
            assertSettingsFrameOfMaxFrameSize(in);

            sendEmptySettingsAndAckFrame(bos);

            readBytes(in, 9); // Read a SETTINGS_ACK frame and discard it.
            readHeadersFrame(in); // Read a HEADERS frame and discard it.

            sendHeaderFrame(bos);

            ////////////////////////////////////////
            // Transmission of data gets started. //
            ////////////////////////////////////////

            // Send a DATA frame that indicates sending data as much as 0x8000 for stream id 03.
            bos.write(new byte[] { 0x00, (byte) 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03 });
            bos.write(EMPTY_DATA);
            bos.write(EMPTY_DATA);
            bos.flush();

            readBytes(in, 13); // Read a WINDOW_UPDATE frame for connection and discard it.
            readBytes(in, 13); // Read a WINDOW_UPDATE frame for stream id 03 and discard it.

            // Send a DATA frame that exceed MAX_FRAME_SIZE by 1.
            bos.write(new byte[] { 0x00, (byte) 0x80, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03 });
            bos.flush(); // Triggers the client to send a GOAWAY frame for the connection.

            // The client send a GOAWAY frame and the server read it.
            final ByteBuf buffer = readGoAwayFrame(in);
            final DefaultHttp2FrameReader frameReader = new DefaultHttp2FrameReader();

            final CountDownLatch latch = new CountDownLatch(1);
            frameReader.readFrame(null, buffer, new Http2EventAdapter() {
                @Override
                public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode,
                                         ByteBuf debugData)
                        throws Http2Exception {
                    assertThat(lastStreamId).isZero(); // 0: connection error
                    assertThat(errorCode).isEqualTo(Http2Error.FRAME_SIZE_ERROR.code());
                    latch.countDown();
                }
            });
            latch.await();
            buffer.release();

            // Client should disconnect after receiving a GOAWAY frame.
            assertThat(in.read()).isEqualTo(-1);
        }
    }
}
 

private static Http2FrameReader frameReader() {
    return new Http2InboundFrameLogger(new DefaultHttp2FrameReader(), logger);
}
 

public HelloWorldHttp2Handler() {
    this(new DefaultHttp2Connection(true), new Http2InboundFrameLogger(
            new DefaultHttp2FrameReader(), logger), 
            new Http2OutboundFrameLogger(new DefaultHttp2FrameWriter(), logger), 
            new SimpleHttp2FrameListener());
}
 

static NettyServerHandler newHandler(
    ServerTransportListener transportListener,
    ChannelPromise channelUnused,
    List<? extends ServerStreamTracer.Factory> streamTracerFactories,
    TransportTracer transportTracer,
    int maxStreams,
    boolean autoFlowControl,
    int flowControlWindow,
    int maxHeaderListSize,
    int maxMessageSize,
    long keepAliveTimeInNanos,
    long keepAliveTimeoutInNanos,
    long maxConnectionIdleInNanos,
    long maxConnectionAgeInNanos,
    long maxConnectionAgeGraceInNanos,
    boolean permitKeepAliveWithoutCalls,
    long permitKeepAliveTimeInNanos) {
  Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive: %s",
      maxHeaderListSize);
  Http2FrameLogger frameLogger = new Http2FrameLogger(LogLevel.DEBUG, NettyServerHandler.class);
  Http2HeadersDecoder headersDecoder = new GrpcHttp2ServerHeadersDecoder(maxHeaderListSize);
  Http2FrameReader frameReader = new Http2InboundFrameLogger(
      new DefaultHttp2FrameReader(headersDecoder), frameLogger);
  Http2FrameWriter frameWriter =
      new Http2OutboundFrameLogger(new DefaultHttp2FrameWriter(), frameLogger);
  return newHandler(
      channelUnused,
      frameReader,
      frameWriter,
      transportListener,
      streamTracerFactories,
      transportTracer,
      maxStreams,
      autoFlowControl,
      flowControlWindow,
      maxHeaderListSize,
      maxMessageSize,
      keepAliveTimeInNanos,
      keepAliveTimeoutInNanos,
      maxConnectionIdleInNanos,
      maxConnectionAgeInNanos,
      maxConnectionAgeGraceInNanos,
      permitKeepAliveWithoutCalls,
      permitKeepAliveTimeInNanos);
}