@Override
public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
  if (keepAliveManager != null) {
    keepAliveManager.onDataReceived();
  }
  if (data == flowControlPing().payload()) {
    flowControlPing().updateWindow();
    if (logger.isLoggable(Level.FINE)) {
      logger.log(Level.FINE, String.format("Window: %d",
          decoder().flowController().initialWindowSize(connection().connectionStream())));
    }
  } else if (data == GRACEFUL_SHUTDOWN_PING) {
    if (gracefulShutdown == null) {
      // this should never happen
      logger.warning("Received GRACEFUL_SHUTDOWN_PING Ack but gracefulShutdown is null");
    } else {
      gracefulShutdown.secondGoAwayAndClose(ctx);
    }
  } else if (data != KEEPALIVE_PING) {
    logger.warning("Received unexpected ping ack. No ping outstanding");
  }
}
 

private void forcefulClose(final ChannelHandlerContext ctx, final ForcefulCloseCommand msg,
    ChannelPromise promise) throws Exception {
  // close() already called by NettyClientTransport, so just need to clean up streams
  connection().forEachActiveStream(new Http2StreamVisitor() {
    @Override
    public boolean visit(Http2Stream stream) throws Http2Exception {
      NettyClientStream.TransportState clientStream = clientStream(stream);
      if (clientStream != null) {
        clientStream.transportReportStatus(msg.getStatus(), true, new Metadata());
        resetStream(ctx, stream.id(), Http2Error.CANCEL.code(), ctx.newPromise());
      }
      stream.close();
      return true;
    }
  });
  promise.setSuccess();
}
 

/**
 * Handler for a GOAWAY being received. Fails any streams created after the
 * last known stream.
 */
private void goingAway(Status status) {
  lifecycleManager.notifyShutdown(status);
  final Status goAwayStatus = lifecycleManager.getShutdownStatus();
  final int lastKnownStream = connection().local().lastStreamKnownByPeer();
  try {
    connection().forEachActiveStream(new Http2StreamVisitor() {
      @Override
      public boolean visit(Http2Stream stream) throws Http2Exception {
        if (stream.id() > lastKnownStream) {
          NettyClientStream.TransportState clientStream = clientStream(stream);
          if (clientStream != null) {
            clientStream.transportReportStatus(
                goAwayStatus, RpcProgress.REFUSED, false, new Metadata());
          }
          stream.close();
        }
        return true;
      }
    });
  } catch (Http2Exception e) {
    throw new RuntimeException(e);
  }
}
 

@Override
public void onPingAckRead(ChannelHandlerContext ctx, long ackPayload) throws Http2Exception {
  Http2Ping p = ping;
  if (ackPayload == flowControlPing().payload()) {
    flowControlPing().updateWindow();
    if (logger.isLoggable(Level.FINE)) {
      logger.log(Level.FINE, String.format("Window: %d",
          decoder().flowController().initialWindowSize(connection().connectionStream())));
    }
  } else if (p != null) {
    if (p.payload() == ackPayload) {
      p.complete();
      ping = null;
    } else {
      logger.log(Level.WARNING, String.format(
          "Received unexpected ping ack. Expecting %d, got %d", p.payload(), ackPayload));
    }
  } else {
    logger.warning("Received unexpected ping ack. No ping outstanding");
  }
  if (keepAliveManager != null) {
    keepAliveManager.onDataReceived();
  }
}
 

private void forcefulClose(final ChannelHandlerContext ctx, final ForcefulCloseCommand msg,
    ChannelPromise promise) throws Exception {
  super.close(ctx, promise);
  connection().forEachActiveStream(new Http2StreamVisitor() {
    @Override
    public boolean visit(Http2Stream stream) throws Http2Exception {
      NettyServerStream.TransportState serverStream = serverStream(stream);
      if (serverStream != null) {
        PerfMark.startTask("NettyServerHandler.forcefulClose", serverStream.tag());
        PerfMark.linkIn(msg.getLink());
        try {
          serverStream.transportReportStatus(msg.getStatus());
          resetStream(ctx, stream.id(), Http2Error.CANCEL.code(), ctx.newPromise());
        } finally {
          PerfMark.stopTask("NettyServerHandler.forcefulClose", serverStream.tag());
        }
      }
      stream.close();
      return true;
    }
  });
}
 

@Override
public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
  if (keepAliveManager != null) {
    keepAliveManager.onDataReceived();
  }
  if (data == flowControlPing().payload()) {
    flowControlPing().updateWindow();
    if (logger.isLoggable(Level.FINE)) {
      logger.log(Level.FINE, String.format("Window: %d",
          decoder().flowController().initialWindowSize(connection().connectionStream())));
    }
  } else if (data == GRACEFUL_SHUTDOWN_PING) {
    if (gracefulShutdown == null) {
      // this should never happen
      logger.warning("Received GRACEFUL_SHUTDOWN_PING Ack but gracefulShutdown is null");
    } else {
      gracefulShutdown.secondGoAwayAndClose(ctx);
    }
  } else if (data != KEEPALIVE_PING) {
    logger.warning("Received unexpected ping ack. No ping outstanding");
  }
}
 

private void onRstStreamRead(int streamId, long errorCode) throws Http2Exception {
  try {
    NettyServerStream.TransportState stream = serverStream(connection().stream(streamId));
    if (stream != null) {
      PerfMark.startTask("NettyServerHandler.onRstStreamRead", stream.tag());
      try {
        stream.transportReportStatus(
            Status.CANCELLED.withDescription("RST_STREAM received for code " + errorCode));
      } finally {
        PerfMark.stopTask("NettyServerHandler.onRstStreamRead", stream.tag());
      }
    }
  } catch (Throwable e) {
    logger.log(Level.WARNING, "Exception in onRstStreamRead()", e);
    // Throw an exception that will get handled by onStreamError.
    throw newStreamException(streamId, e);
  }
}
 

private ChannelPromise handleOutstandingControlFrames(ChannelHandlerContext ctx, ChannelPromise promise) {
    if (!limitReached) {
        if (outstandingControlFrames == maxOutstandingControlFrames) {
            // Let's try to flush once as we may be able to flush some of the control frames.
            ctx.flush();
        }
        if (outstandingControlFrames == maxOutstandingControlFrames) {
            limitReached = true;
            Http2Exception exception = Http2Exception.connectionError(Http2Error.ENHANCE_YOUR_CALM,
                    "Maximum number %d of outstanding control frames reached", maxOutstandingControlFrames);
            logger.info("Maximum number {} of outstanding control frames reached. Closing channel {}",
                    maxOutstandingControlFrames, ctx.channel(), exception);

            // First notify the Http2LifecycleManager and then close the connection.
            lifecycleManager.onError(ctx, true, exception);
            ctx.close();
        }
        outstandingControlFrames++;

        // We did not reach the limit yet, add the listener to decrement the number of outstanding control frames
        // once the promise was completed
        return promise.unvoid().addListener(outstandingControlFramesListener);
    }
    return promise;
}
 

private void forcefulClose(final ChannelHandlerContext ctx, final ForcefulCloseCommand msg,
    ChannelPromise promise) throws Exception {
  // close() already called by NettyClientTransport, so just need to clean up streams
  connection().forEachActiveStream(new Http2StreamVisitor() {
    @Override
    public boolean visit(Http2Stream stream) throws Http2Exception {
      NettyClientStream.TransportState clientStream = clientStream(stream);
      Tag tag = clientStream != null ? clientStream.tag() : PerfMark.createTag();
      PerfMark.startTask("NettyClientHandler.forcefulClose", tag);
      PerfMark.linkIn(msg.getLink());
      try {
        if (clientStream != null) {
          clientStream.transportReportStatus(msg.getStatus(), true, new Metadata());
          resetStream(ctx, stream.id(), Http2Error.CANCEL.code(), ctx.newPromise());
        }
        stream.close();
        return true;
      } finally {
        PerfMark.stopTask("NettyClientHandler.forcefulClose", tag);
      }
    }
  });
  promise.setSuccess();
}
 

private void writeErrorResponse(ChannelHandlerContext ctx, int streamId, HttpResponseStatus status,
                                @Nullable ByteBuf content) throws Http2Exception {
    final ByteBuf data =
            content != null ? content
                            : Unpooled.wrappedBuffer(status.toString().getBytes(StandardCharsets.UTF_8));

    writer.writeHeaders(
            ctx, streamId,
            new DefaultHttp2Headers(false)
                    .status(status.codeAsText())
                    .set(HttpHeaderNames.CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8.toString())
                    .setInt(HttpHeaderNames.CONTENT_LENGTH, data.readableBytes()),
            0, false, ctx.voidPromise());

    writer.writeData(ctx, streamId, data, 0, true, ctx.voidPromise());

    final Http2Stream stream = writer.connection().stream(streamId);
    if (stream != null && writer.flowController().hasFlowControlled(stream)) {
        // Ensure to flush the error response if it's flow-controlled so that it is sent
        // before an RST_STREAM frame.
        writer.flowController().writePendingBytes();
    }
}
 

@Test
public void testIsExpected() {
    final boolean expected = !Flags.verboseSocketExceptions();

    assertThat(Exceptions.isExpected(new Exception())).isFalse();
    assertThat(Exceptions.isExpected(new Exception("broken pipe"))).isFalse();
    assertThat(Exceptions.isExpected(new Exception("connection reset by peer"))).isFalse();
    assertThat(Exceptions.isExpected(new Exception("stream closed"))).isFalse();
    assertThat(Exceptions.isExpected(new Exception("SSLEngine closed already"))).isFalse();

    assertThat(Exceptions.isExpected(new ClosedChannelException())).isEqualTo(expected);
    assertThat(Exceptions.isExpected(ClosedSessionException.get())).isEqualTo(expected);

    assertThat(Exceptions.isExpected(new IOException("connection reset by peer"))).isEqualTo(expected);
    assertThat(Exceptions.isExpected(new IOException("invalid argument"))).isEqualTo(false);

    assertThat(Exceptions.isExpected(new ChannelException("broken pipe"))).isEqualTo(expected);
    assertThat(Exceptions.isExpected(new ChannelException("invalid argument"))).isEqualTo(false);

    assertThat(Exceptions.isExpected(new Http2Exception(Http2Error.INTERNAL_ERROR, "stream closed")))
            .isEqualTo(expected);
    assertThat(Exceptions.isExpected(new Http2Exception(Http2Error.INTERNAL_ERROR))).isEqualTo(false);

    assertThat(Exceptions.isExpected(new SSLException("SSLEngine closed already"))).isEqualTo(expected);
    assertThat(Exceptions.isExpected(new SSLException("Handshake failed"))).isEqualTo(false);
}
 

private void onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream)
    throws Http2Exception {
  flowControlPing().onDataRead(data.readableBytes(), padding);
  try {
    NettyServerStream.TransportState stream = serverStream(requireHttp2Stream(streamId));
    PerfMark.startTask("NettyServerHandler.onDataRead", stream.tag());
    try {
      stream.inboundDataReceived(data, endOfStream);
    } finally {
      PerfMark.stopTask("NettyServerHandler.onDataRead", stream.tag());
    }
  } catch (Throwable e) {
    logger.log(Level.WARNING, "Exception in onDataRead()", e);
    // Throw an exception that will get handled by onStreamError.
    throw newStreamException(streamId, e);
  }
}
 

private void onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream)
    throws Http2Exception {
  flowControlPing().onDataRead(data.readableBytes(), padding);
  try {
    NettyServerStream.TransportState stream = serverStream(requireHttp2Stream(streamId));
    stream.inboundDataReceived(data, endOfStream);
  } catch (Throwable e) {
    logger.log(Level.WARNING, "Exception in onDataRead()", e);
    // Throw an exception that will get handled by onStreamError.
    throw newStreamException(streamId, e);
  }
}
 

private void onRstStreamRead(int streamId, long errorCode) throws Http2Exception {
  try {
    NettyServerStream.TransportState stream = serverStream(connection().stream(streamId));
    if (stream != null) {
      stream.transportReportStatus(
          Status.CANCELLED.withDescription("RST_STREAM received for code " + errorCode));
    }
  } catch (Throwable e) {
    logger.log(Level.WARNING, "Exception in onRstStreamRead()", e);
    // Throw an exception that will get handled by onStreamError.
    throw newStreamException(streamId, e);
  }
}
 

@Override
protected void onConnectionError(ChannelHandlerContext ctx, boolean outbound, Throwable cause,
    Http2Exception http2Ex) {
  logger.log(Level.FINE, "Connection Error", cause);
  connectionError = cause;
  super.onConnectionError(ctx, outbound, cause, http2Ex);
}
 

/**
 * Handler for the Channel shutting down.
 */
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  try {
    if (keepAliveManager != null) {
      keepAliveManager.onTransportTermination();
    }
    if (maxConnectionIdleManager != null) {
      maxConnectionIdleManager.onTransportTermination();
    }
    if (maxConnectionAgeMonitor != null) {
      maxConnectionAgeMonitor.cancel(false);
    }
    final Status status =
        Status.UNAVAILABLE.withDescription("connection terminated for unknown reason");
    // Any streams that are still active must be closed
    connection().forEachActiveStream(new Http2StreamVisitor() {
      @Override
      public boolean visit(Http2Stream stream) throws Http2Exception {
        NettyServerStream.TransportState serverStream = serverStream(stream);
        if (serverStream != null) {
          serverStream.transportReportStatus(status);
        }
        return true;
      }
    });
  } finally {
    super.channelInactive(ctx);
  }
}
 

/**
 * Returns the given processed bytes back to inbound flow control.
 */
void returnProcessedBytes(Http2Stream http2Stream, int bytes) {
  try {
    decoder().flowController().consumeBytes(http2Stream, bytes);
  } catch (Http2Exception e) {
    throw new RuntimeException(e);
  }
}
 

/**
 * Sends the given gRPC frame to the client.
 */
private void sendGrpcFrame(ChannelHandlerContext ctx, SendGrpcFrameCommand cmd,
    ChannelPromise promise) throws Http2Exception {
  if (cmd.endStream()) {
    closeStreamWhenDone(promise, cmd.streamId());
  }
  // Call the base class to write the HTTP/2 DATA frame.
  encoder().writeData(ctx, cmd.streamId(), cmd.content(), 0, cmd.endStream(), promise);
}
 

@Override
public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
    boolean endOfStream) throws Http2Exception {
  if (keepAliveManager != null) {
    keepAliveManager.onDataReceived();
  }
  NettyServerHandler.this.onDataRead(streamId, data, padding, endOfStream);
  return padding;
}
 

@Test
public void decode_emptyHeaders() throws Http2Exception {
  Http2HeadersDecoder decoder = new GrpcHttp2ClientHeadersDecoder(8192);
  Http2HeadersEncoder encoder =
      new DefaultHttp2HeadersEncoder(NEVER_SENSITIVE);

  ByteBuf encodedHeaders = Unpooled.buffer();
  encoder.encodeHeaders(1 /* randomly chosen */, new DefaultHttp2Headers(false), encodedHeaders);

  Http2Headers decodedHeaders = decoder.decodeHeaders(3 /* randomly chosen */, encodedHeaders);
  assertEquals(0, decodedHeaders.size());
  assertThat(decodedHeaders.toString(), containsString("[]"));
}
 

@Override
public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode)
    throws Http2Exception {
  if (keepAliveManager != null) {
    keepAliveManager.onDataReceived();
  }
  NettyServerHandler.this.onRstStreamRead(streamId, errorCode);
}
 

@Override
public void onPushPromiseRead(
    ChannelHandlerContext ctx,
    int streamId,
    int promisedStreamId,
    Http2Headers headers,
    int padding)
    throws Http2Exception {
  // TODO(CK): We don't currently have a use case for these frames
}
 

/**
 * Handler for the Channel shutting down.
 */
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  try {
    logger.fine("Network channel is closed");
    Status status = Status.UNAVAILABLE.withDescription("Network closed for unknown reason");
    lifecycleManager.notifyShutdown(status);
    try {
      cancelPing(lifecycleManager.getShutdownThrowable());
      // Report status to the application layer for any open streams
      connection().forEachActiveStream(new Http2StreamVisitor() {
        @Override
        public boolean visit(Http2Stream stream) throws Http2Exception {
          NettyClientStream.TransportState clientStream = clientStream(stream);
          if (clientStream != null) {
            clientStream.transportReportStatus(
                lifecycleManager.getShutdownStatus(), false, new Metadata());
          }
          return true;
        }
      });
    } finally {
      lifecycleManager.notifyTerminated(status);
    }
  } finally {
    // Close any open streams
    super.channelInactive(ctx);
    if (keepAliveManager != null) {
      keepAliveManager.onTransportTermination();
    }
  }
}
 

@Override
public final void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  Http2Exception embedded = getEmbeddedHttp2Exception(cause);
  if (embedded == null) {
    // There was no embedded Http2Exception, assume it's a connection error. Subclasses are
    // responsible for storing the appropriate status and shutting down the connection.
    onError(ctx, /* outbound= */ false, cause);
  } else {
    super.exceptionCaught(ctx, cause);
  }
}
 

@Override
protected void onStreamError(ChannelHandlerContext ctx, boolean outbound, Throwable cause,
    Http2Exception.StreamException http2Ex) {
  // Close the stream with a status that contains the cause.
  NettyClientStream.TransportState stream = clientStream(connection().stream(http2Ex.streamId()));
  if (stream != null) {
    stream.transportReportStatus(Utils.statusFromThrowable(cause), false, new Metadata());
  } else {
    logger.log(Level.FINE, "Stream error for unknown stream " + http2Ex.streamId(), cause);
  }

  // Delegate to the base class to send a RST_STREAM.
  super.onStreamError(ctx, outbound, cause, http2Ex);
}
 

@Override
public void onHeadersRead(ChannelHandlerContext ctx,
    int streamId,
    Http2Headers headers,
    int streamDependency,
    short weight,
    boolean exclusive,
    int padding,
    boolean endStream) throws Http2Exception {
  NettyClientHandler.this.onHeadersRead(streamId, headers, endStream);
}
 

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
{
    try {
        if (cause instanceof Http2Exception) {
            incrementErrorCounter(registry, errorCounterName, metricId, (Http2Exception) cause);
        }
    }
    finally {
        super.exceptionCaught(ctx, cause);
    }
}
 

/**
 * Sends initial connection window to the remote endpoint if necessary.
 */
private void sendInitialConnectionWindow() throws Http2Exception {
  if (ctx.channel().isActive() && initialConnectionWindow > 0) {
    Http2Stream connectionStream = connection().connectionStream();
    int currentSize = connection().local().flowController().windowSize(connectionStream);
    int delta = initialConnectionWindow - currentSize;
    decoder().flowController().incrementWindowSize(connectionStream, delta);
    initialConnectionWindow = -1;
    ctx.flush();
  }
}
 

/**
 * Sends initial connection window to the remote endpoint if necessary.
 */
private void sendInitialConnectionWindow() throws Http2Exception {
  if (!initialWindowSent && ctx.channel().isActive()) {
    Http2Stream connectionStream = connection().connectionStream();
    int currentSize = connection().local().flowController().windowSize(connectionStream);
    int delta = initialConnectionWindow - currentSize;
    decoder().flowController().incrementWindowSize(connectionStream, delta);
    initialWindowSent = true;
    ctx.flush();
  }
}
 

@Test
public void decode_requestHeaders() throws Http2Exception {
  Http2HeadersDecoder decoder = new GrpcHttp2ServerHeadersDecoder(DEFAULT_MAX_HEADER_LIST_SIZE);
  Http2HeadersEncoder encoder =
      new DefaultHttp2HeadersEncoder(NEVER_SENSITIVE);

  Http2Headers headers = new DefaultHttp2Headers(false);
  headers.add(of(":scheme"), of("https")).add(of(":method"), of("GET"))
      .add(of(":path"), of("index.html")).add(of(":authority"), of("foo.grpc.io"))
      .add(of("custom"), of("header"));
  encodedHeaders = Unpooled.buffer();
  encoder.encodeHeaders(1 /* randomly chosen */, headers, encodedHeaders);

  Http2Headers decodedHeaders = decoder.decodeHeaders(3 /* randomly chosen */, encodedHeaders);
  assertEquals(headers.get(of(":scheme")), decodedHeaders.scheme());
  assertEquals(headers.get(of(":method")), decodedHeaders.method());
  assertEquals(headers.get(of(":path")), decodedHeaders.path());
  assertEquals(headers.get(of(":authority")), decodedHeaders.authority());
  assertEquals(headers.get(of("custom")), decodedHeaders.get(of("custom")));
  assertEquals(headers.size(), decodedHeaders.size());

  String toString = decodedHeaders.toString();
  assertContainsKeyAndValue(toString, ":scheme", decodedHeaders.scheme());
  assertContainsKeyAndValue(toString, ":method", decodedHeaders.method());
  assertContainsKeyAndValue(toString, ":path", decodedHeaders.path());
  assertContainsKeyAndValue(toString, ":authority", decodedHeaders.authority());
  assertContainsKeyAndValue(toString, "custom", decodedHeaders.get(of("custom")));
}