Http3PushStreamServerInitializer() — netty Function Reference

Source Code

codec-http3/src/main/java/io/netty/handler/codec/http3/Http3ServerPushStreamManager.java lines 212–270

    private Http3PushStreamServerInitializer pushStreamInitializer(long pushId, @Nullable ChannelHandler handler) {
        final Http3PushStreamServerInitializer initializer;
        if (handler instanceof Http3PushStreamServerInitializer) {
            initializer = (Http3PushStreamServerInitializer) handler;
        } else {
            initializer = null;
        }
        return new Http3PushStreamServerInitializer(pushId) {
            @Override
            protected void initPushStream(QuicStreamChannel ch) {
                ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                    private boolean stateUpdated;

                    @Override
                    public void channelActive(ChannelHandlerContext ctx) {
                        if (!stateUpdated) {
                            updatePushStreamsMap();
                        }
                    }

                    @Override
                    public void handlerAdded(ChannelHandlerContext ctx) {
                        if (!stateUpdated && ctx.channel().isActive()) {
                            updatePushStreamsMap();
                        }
                    }

                    private void updatePushStreamsMap() {
                        assert !stateUpdated;
                        stateUpdated = true;
                        pushStreams.compute(pushId, (id, existing) -> {
                            if (existing == AWAITING_STREAM_ESTABLISHMENT) {
                                return ch;
                            }
                            if (existing == CANCELLED_STREAM) {
                                ch.close();
                                return null; // remove push stream.
                            }
                            throw new IllegalStateException("Unexpected push stream state " +
                                    existing + " for pushId: " + id);
                        });
                    }

                    @Override
                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                        if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
                            pushStreams.remove(pushId);
                        }
                        ctx.fireUserEventTriggered(evt);
                    }
                });
                if (initializer != null) {
                    initializer.initPushStream(ch);
                } else if (handler != null) {
                    ch.pipeline().addLast(handler);
                }
            }
        };
    }