StompChatHandler Class — netty Architecture

Source Code

example/src/main/java/io/netty/example/stomp/websocket/StompChatHandler.java lines 39–195

@Sharable
public class StompChatHandler extends SimpleChannelInboundHandler<StompFrame> {

    private final ConcurrentMap<String, Set<StompSubscription>> chatDestinations =
            new ConcurrentHashMap<String, Set<StompSubscription>>();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, StompFrame inboundFrame) throws Exception {
        DecoderResult decoderResult = inboundFrame.decoderResult();
        if (decoderResult.isFailure()) {
            sendErrorFrame("rejected frame", decoderResult.toString(), ctx);
            return;
        }

        switch (inboundFrame.command()) {
        case STOMP:
        case CONNECT:
            onConnect(ctx, inboundFrame);
            break;
        case SUBSCRIBE:
            onSubscribe(ctx, inboundFrame);
            break;
        case SEND:
            onSend(ctx, inboundFrame);
            break;
        case UNSUBSCRIBE:
            onUnsubscribe(ctx, inboundFrame);
            break;
        case DISCONNECT:
            onDisconnect(ctx, inboundFrame);
            break;
        default:
            sendErrorFrame("unsupported command",
                           "Received unsupported command " + inboundFrame.command(), ctx);
        }
    }

    private void onSubscribe(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String destination = inboundFrame.headers().getAsString(DESTINATION);
        String subscriptionId = inboundFrame.headers().getAsString(ID);

        if (destination == null || subscriptionId == null) {
            sendErrorFrame("missed header", "Required 'destination' or 'id' header missed", ctx);
            return;
        }

        Set<StompSubscription> subscriptions = chatDestinations.get(destination);
        if (subscriptions == null) {
            subscriptions = new HashSet<StompSubscription>();
            Set<StompSubscription> previousSubscriptions = chatDestinations.putIfAbsent(destination, subscriptions);
            if (previousSubscriptions != null) {
                subscriptions = previousSubscriptions;
            }
        }

        final StompSubscription subscription = new StompSubscription(subscriptionId, destination, ctx.channel());
        if (subscriptions.contains(subscription)) {
            sendErrorFrame("duplicate subscription",
                           "Received duplicate subscription id=" + subscriptionId, ctx);
            return;
        }

        subscriptions.add(subscription);
        ctx.channel().closeFuture().addListener(f ->
                chatDestinations.get(subscription.destination()).remove(subscription));

        String receiptId = inboundFrame.headers().getAsString(RECEIPT);
        if (receiptId != null) {
            StompFrame receiptFrame = new DefaultStompFrame(StompCommand.RECEIPT);
            receiptFrame.headers().set(RECEIPT_ID, receiptId);
            ctx.writeAndFlush(receiptFrame);
        }
    }

    private void onSend(ChannelHandlerContext ctx, StompFrame inboundFrame) {
        String destination = inboundFrame.headers().getAsString(DESTINATION);
        if (destination == null) {
            sendErrorFrame("missed header", "required 'destination' header missed", ctx);
            return;
        }