KeepAliveManager(final Channel channel, @Nullable final KeepAlivePolicy keepAlivePolicy,
                 final Scheduler scheduler, final IdlenessDetector idlenessDetector) {
    this.channel = channel;
    this.scheduler = scheduler;
    if (keepAlivePolicy != null) {
        disallowKeepAliveWithoutActiveStreams = !keepAlivePolicy.withoutActiveStreams();
        pingAckTimeoutNanos = keepAlivePolicy.ackTimeout().toNanos();
        pingWriteCompletionListener = future -> {
            if (future.isSuccess() && keepAliveState == KEEP_ALIVE_ACK_PENDING) {
                // Schedule a task to verify ping ack within the pingAckTimeoutMillis
                keepAliveState = scheduler.afterDuration(() -> {
                    if (keepAliveState != null) {
                        keepAliveState = KEEP_ALIVE_ACK_TIMEDOUT;
                        LOGGER.debug(
                                "channel={}, timeout {}ns waiting for keep-alive PING(ACK), writing go_away.",
                                this.channel, pingAckTimeoutNanos);
                        channel.writeAndFlush(new DefaultHttp2GoAwayFrame(NO_ERROR))
                                .addListener(f -> {
                                    if (f.isSuccess()) {
                                        LOGGER.debug("Closing channel={}, after keep-alive timeout.", this.channel);
                                        KeepAliveManager.this.close0();
                                    }
                                });
                    }
                }, pingAckTimeoutNanos, NANOSECONDS);
            }
        };
        int idleInSeconds = (int) min(keepAlivePolicy.idleDuration().getSeconds(), Integer.MAX_VALUE);
        idlenessDetector.configure(channel, idleInSeconds, this::channelIdle);
    } else {
        disallowKeepAliveWithoutActiveStreams = false;
        pingAckTimeoutNanos = DEFAULT_ACK_TIMEOUT.toNanos();
        pingWriteCompletionListener = null;
    }
}
 

@Override
public void channelRegistered(final ChannelHandlerContext ctx) {
    // See SETTINGS_ENABLE_PUSH in https://tools.ietf.org/html/rfc7540#section-6.5.2
    ctx.writeAndFlush(new DefaultHttp2GoAwayFrame(PROTOCOL_ERROR));
    // Http2ConnectionHandler.processGoAwayWriteResult will close the connection after GO_AWAY is flushed
}
 

/**
 * WARNING: Found the OkHttp client gets confused by this behaviour (it ends up putting itself in a bad shutdown state
 * after receiving the first goaway frame, and then dropping any inflight responses but also timing out waiting for them).
 *
 * And worried that other http/2 stacks may be similar, so for now we should NOT use this.
 *
 * This is unfortunate, as FTL wanted this, and it is correct according to the spec.
 *
 * See this code in okhttp where it drops response header frame if state is already shutdown:
 * https://github.com/square/okhttp/blob/master/okhttp/src/main/java/okhttp3/internal/http2/Http2Connection.java#L609
 */
private void gracefullyWithDelay(EventExecutor executor, Channel parent, ChannelPromise promise)
{
    // See javadoc for explanation of why this may be disabled.
    boolean allowGracefulDelayed = ConnectionCloseChannelAttributes.allowGracefulDelayed(parent);
    if (! allowGracefulDelayed) {
        immediate(parent, promise);
        return;
    }

    if (! parent.isActive()) {
        promise.setSuccess();
        return;
    }

    // First send a 'graceful shutdown' GOAWAY frame.
    /*
    "A server that is attempting to gracefully shut down a connection SHOULD send an initial GOAWAY frame with
    the last stream identifier set to 231-1 and a NO_ERROR code. This signals to the client that a shutdown is
    imminent and that initiating further requests is prohibited."
      -- https://http2.github.io/http2-spec/#GOAWAY
     */
    DefaultHttp2GoAwayFrame goaway = new DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR);
    goaway.setExtraStreamIds(Integer.MAX_VALUE);
    parent.writeAndFlush(goaway);
    LOG.debug("gracefullyWithDelay: flushed initial go_away frame. channel=" + parent.id().asShortText());

    // In N secs time, throw an error that causes the http2 codec to send another GOAWAY frame
    // (this time with accurate lastStreamId) and then close the connection.
    int gracefulCloseDelay = ConnectionCloseChannelAttributes.gracefulCloseDelay(parent);
    executor.schedule(() -> {

        // Check that the client hasn't already closed the connection (due to the earlier goaway we sent).
        if (parent.isActive()) {
            // NOTE - the netty Http2ConnectionHandler specifically does not send another goaway when we call
            // channel.close() if one has already been sent .... so when we want more than one sent, we need to do it
            // explicitly ourselves like this.
            LOG.debug("gracefullyWithDelay: firing graceful_shutdown event to make netty send a final go_away frame and then close connection. channel="
                    + parent.id().asShortText());
            Http2Exception h2e = new Http2Exception(Http2Error.NO_ERROR, Http2Exception.ShutdownHint.GRACEFUL_SHUTDOWN);
            parent.pipeline().fireExceptionCaught(h2e);

            parent.close().addListener(future -> {
                promise.setSuccess();
            });
        } else {
            promise.setSuccess();
        }

    }, gracefulCloseDelay, TimeUnit.SECONDS);
}