class RadioSession extends SessionBase
    {
        enum State
        {
            GROUP,
            BODY
        }

        private State state;
        private Msg pending;

        public RadioSession(IOThread ioThread, boolean connect, SocketBase socket, final Options options,
                            final Address addr)
        {
            super(ioThread, connect, socket, options, addr);

            state = State.GROUP;
        }

        @Override
        public boolean pushMsg(Msg msg)
        {
            if (msg.isCommand()) {
                byte commandNameSize = msg.get(0);

                if (msg.size() < commandNameSize + 1) {
                    return super.pushMsg(msg);
                }

                byte[] data = msg.data();

                String commandName = new String(data, 1, commandNameSize, StandardCharsets.US_ASCII);

                int groupLength;
                String group;
                Msg joinLeaveMsg = new Msg();

                // Set the msg type to either JOIN or LEAVE
                if (commandName.equals("JOIN")) {
                    groupLength = msg.size() - 5;
                    group = new String(data, 5, groupLength, StandardCharsets.US_ASCII);
                    joinLeaveMsg.initJoin();
                }
                else if (commandName.equals("LEAVE")) {
                    groupLength = msg.size() - 6;
                    group = new String(data, 6, groupLength, StandardCharsets.US_ASCII);
                    joinLeaveMsg.initLeave();
                }
                // If it is not a JOIN or LEAVE just push the message
                else {
                    return super.pushMsg(msg);
                }

                //  Set the group
                joinLeaveMsg.setGroup(group);

                //  Push the join or leave command
                msg = joinLeaveMsg;
                return super.pushMsg(msg);
            }

            return super.pushMsg(msg);
        }

        @Override
        protected Msg pullMsg()
        {<FILL_FUNCTION_BODY>}

        @Override
        protected void reset()
        {
            super.reset();
            state = State.GROUP;
        }
    }

            Msg msg;

            switch (state) {
                case GROUP:
                    pending = super.pullMsg();
                    if (pending == null) {
                        return null;
                    }

                    //  First frame is the group
                    msg = new Msg(pending.getGroup().getBytes(StandardCharsets.US_ASCII));
                    msg.setFlags(Msg.MORE);

                    //  Next status is the body
                    state = State.BODY;
                    break;
                case BODY:
                    msg = pending;
                    state = State.GROUP;
                    break;
                default:
                    throw new IllegalStateException();
            }

            return msg;
        
<methods>public final boolean bind(java.lang.String) ,public final void cancel(java.util.concurrent.atomic.AtomicBoolean) ,public final void close() ,public final boolean connect(java.lang.String) ,public final int connectPeer(java.lang.String) ,public boolean disconnectPeer(int) ,public final int errno() ,public final void eventAcceptFailed(java.lang.String, int) ,public final void eventAccepted(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventBindFailed(java.lang.String, int) ,public final void eventCloseFailed(java.lang.String, int) ,public final void eventClosed(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventConnectDelayed(java.lang.String, int) ,public final void eventConnectRetried(java.lang.String, int) ,public final void eventConnected(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventDisconnected(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventHandshakeFailedAuth(java.lang.String, int) ,public final void eventHandshakeFailedNoDetail(java.lang.String, int) ,public final void eventHandshakeFailedProtocol(java.lang.String, int) ,public final void eventHandshakeSucceeded(java.lang.String, int) ,public final void eventHandshaken(java.lang.String, int) ,public final void eventListening(java.lang.String, java.nio.channels.SelectableChannel) ,public final java.nio.channels.SelectableChannel getFD() ,public final int getSocketOpt(int) ,public final java.lang.Object getSocketOptx(int) ,public final void hiccuped(zmq.pipe.Pipe) ,public final void inEvent() ,public final boolean join(java.lang.String) ,public final boolean leave(java.lang.String) ,public final boolean monitor(java.lang.String, int) ,public final void pipeTerminated(zmq.pipe.Pipe) ,public final int poll(int, int, java.util.concurrent.atomic.AtomicBoolean) ,public final void readActivated(zmq.pipe.Pipe) ,public final zmq.Msg recv(int) ,public final zmq.Msg recv(int, java.util.concurrent.atomic.AtomicBoolean) ,public final boolean send(zmq.Msg, int) ,public final boolean send(zmq.Msg, int, java.util.concurrent.atomic.AtomicBoolean) ,public final boolean setEventHook(zmq.ZMQ.EventConsummer, int) ,public final boolean setSocketOpt(int, java.lang.Object) ,public final boolean termEndpoint(java.lang.String) ,public java.lang.String toString() ,public java.lang.String typeString() ,public final void writeActivated(zmq.pipe.Pipe) <variables>private boolean active,protected java.lang.String connectRid,private final non-sealed java.util.concurrent.atomic.AtomicBoolean ctxTerminated,private final non-sealed java.util.concurrent.atomic.AtomicBoolean destroyed,private final non-sealed MultiMap<java.lang.String,zmq.SocketBase.EndpointPipe> endpoints,private java.nio.channels.SocketChannel fileDesc,private zmq.poll.Poller.Handle handle,private final non-sealed MultiMap<java.lang.String,zmq.pipe.Pipe> inprocs,private final non-sealed ThreadLocal<java.lang.Boolean> isInEventThreadLocal,private long lastTsc,private final non-sealed zmq.IMailbox mailbox,private final non-sealed AtomicReference<zmq.ZMQ.EventConsummer> monitor,private int monitorEvents,private final non-sealed Set<zmq.pipe.Pipe> pipes,private zmq.poll.Poller poller,private boolean rcvmore,private final non-sealed boolean threadSafe,private final non-sealed java.util.concurrent.locks.ReentrantLock threadSafeSync,private int ticks