Http3ServerPushStreamManager Class — netty Architecture

Source Code

codec-http3/src/main/java/io/netty/handler/codec/http3/Http3ServerPushStreamManager.java lines 49–290

public final class Http3ServerPushStreamManager {
    private static final AtomicLongFieldUpdater<Http3ServerPushStreamManager> nextIdUpdater =
            newUpdater(Http3ServerPushStreamManager.class, "nextId");
    private static final Object CANCELLED_STREAM = new Object();
    private static final Object PUSH_ID_GENERATED = new Object();
    private static final Object AWAITING_STREAM_ESTABLISHMENT = new Object();

    private final QuicChannel channel;
    private final ConcurrentMap<Long, Object> pushStreams;
    private final ChannelInboundHandler controlStreamListener;

    private volatile long nextId;

    /**
     * Creates a new instance.
     *
     * @param channel for which this manager is created.
     */
    public Http3ServerPushStreamManager(QuicChannel channel) {
        this(channel, 8);
    }

    /**
     * Creates a new instance.
     *
     * @param channel for which this manager is created.
     * @param initialPushStreamsCountHint a hint for the number of push streams that may be created.
     */
    public Http3ServerPushStreamManager(QuicChannel channel, int initialPushStreamsCountHint) {
        this.channel = requireNonNull(channel, "channel");
        pushStreams = newConcurrentHashMap(initialPushStreamsCountHint);
        controlStreamListener = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) {
                if (msg instanceof Http3CancelPushFrame) {
                    final long pushId = ((Http3CancelPushFrame) msg).id();
                    if (pushId >= nextId) {
                        connectionError(ctx, H3_ID_ERROR, "CANCEL_PUSH id greater than the last known id", true);
                        return;
                    }

                    pushStreams.computeIfPresent(pushId, (id, existing) -> {
                        if (existing == AWAITING_STREAM_ESTABLISHMENT) {
                            return CANCELLED_STREAM;
                        }
                        if (existing == PUSH_ID_GENERATED) {
                            throw new IllegalStateException("Unexpected push stream state " + existing +
                                    " for pushId: " + id);
                        }
                        assert existing instanceof QuicStreamChannel;
                        ((QuicStreamChannel) existing).close();
                        // remove the push stream from the map.
                        return null;
                    });
                }
                ReferenceCountUtil.release(msg);
            }
        };
    }

    /**
     * Returns {@code true} if server push is allowed at this point.
     *
     * @return {@code true} if server push is allowed at this point.
     */
    public boolean isPushAllowed() {
        return isPushAllowed(maxPushIdReceived(channel));
    }

    /**
     * Reserves a push ID to be used to create a new push stream subsequently. A push ID can only be used to create
     * exactly one push stream.
     *
     * @return Next push ID.
     * @throws IllegalStateException If it is not allowed to create any more push streams on the associated
     * {@link QuicChannel}. Use {@link #isPushAllowed()} to check if server push is allowed.
     */
    public long reserveNextPushId() {
        final long maxPushId = maxPushIdReceived(channel);
        if (isPushAllowed(maxPushId)) {
            return nextPushId();