@SuppressWarnings("MethodLength")
    public void onFragment(final DirectBuffer buffer, final int offset, final int length, final Header header)
    {
        messageHeaderDecoder.wrap(buffer, offset);

        final int templateId = messageHeaderDecoder.templateId();
        final int schemaId = messageHeaderDecoder.schemaId();
        if (schemaId != MessageHeaderDecoder.SCHEMA_ID)
        {
            if (listenerExtension != null)
            {
                listenerExtension.onExtensionMessage(
                    messageHeaderDecoder.blockLength(),
                    templateId,
                    schemaId,
                    messageHeaderDecoder.version(),
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    length - MessageHeaderDecoder.ENCODED_LENGTH);
                return;
            }
            throw new ClusterException("expected schemaId=" +
                MessageHeaderDecoder.SCHEMA_ID + ", actual=" + schemaId);
        }

        switch (templateId)
        {
            case SessionMessageHeaderDecoder.TEMPLATE_ID:
            {
                sessionMessageHeaderDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                final long sessionId = sessionMessageHeaderDecoder.clusterSessionId();
                if (sessionId == clusterSessionId)
                {
                    listener.onMessage(
                        sessionId,
                        sessionMessageHeaderDecoder.timestamp(),
                        buffer,
                        offset + SESSION_HEADER_LENGTH,
                        length - SESSION_HEADER_LENGTH,
                        header);
                }
                break;
            }

            case SessionEventDecoder.TEMPLATE_ID:
            {
                sessionEventDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                final long sessionId = sessionEventDecoder.clusterSessionId();
                if (sessionId == clusterSessionId)
                {
                    listener.onSessionEvent(
                        sessionEventDecoder.correlationId(),
                        sessionId,
                        sessionEventDecoder.leadershipTermId(),
                        sessionEventDecoder.leaderMemberId(),
                        sessionEventDecoder.code(),
                        sessionEventDecoder.detail());
                }
                break;
            }

            case NewLeaderEventDecoder.TEMPLATE_ID:
            {
                newLeaderEventDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                final long sessionId = newLeaderEventDecoder.clusterSessionId();
                if (sessionId == clusterSessionId)
                {
                    listener.onNewLeader(
                        sessionId,
                        newLeaderEventDecoder.leadershipTermId(),
                        newLeaderEventDecoder.leaderMemberId(),
                        newLeaderEventDecoder.ingressEndpoints());
                }
                break;
            }

            case AdminResponseDecoder.TEMPLATE_ID:
            {
                adminResponseDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                final long sessionId = adminResponseDecoder.clusterSessionId();
                if (sessionId == clusterSessionId)
                {
                    final long correlationId = adminResponseDecoder.correlationId();
                    final AdminRequestType requestType = adminResponseDecoder.requestType();
                    final AdminResponseCode responseCode = adminResponseDecoder.responseCode();
                    final String message = adminResponseDecoder.message();
                    final int payloadOffset = adminResponseDecoder.offset() +
                        AdminResponseDecoder.BLOCK_LENGTH +
                        AdminResponseDecoder.messageHeaderLength() +
                        message.length() +
                        AdminResponseDecoder.payloadHeaderLength();
                    final int payloadLength = adminResponseDecoder.payloadLength();
                    listener.onAdminResponse(
                        sessionId,
                        correlationId,
                        requestType,
                        responseCode,
                        message,
                        buffer,
                        payloadOffset,
                        payloadLength);
                }
                break;
            }

            default:
                break;
        }
    }
@Test
    void onFragmentIsANoOpIfSessionIdDoesNotMatchOnSessionEvent()
    {
        final int offset = 8;
        final long clusterSessionId = 42;
        final long correlationId = 777;
        final long leadershipTermId = 6;
        final int leaderMemberId = 3;
        final EventCode eventCode = EventCode.REDIRECT;
        final int version = 18;
        final String eventDetail = "Event details";
        sessionEventEncoder
            .wrapAndApplyHeader(buffer, offset, messageHeaderEncoder)
            .clusterSessionId(clusterSessionId)
            .correlationId(correlationId)
            .leadershipTermId(leadershipTermId)
            .leaderMemberId(leaderMemberId)
            .code(eventCode)
            .version(version)
            .detail(eventDetail);

        final EgressListener egressListener = mock(EgressListener.class);
        final Header header = new Header(0, 0);
        final EgressAdapter adapter = new EgressAdapter(
            egressListener, clusterSessionId + 1, mock(Subscription.class), 3);

        adapter.onFragment(buffer, offset, sessionEventEncoder.encodedLength(), header);

        verifyNoInteractions(egressListener);
    }