aeron-samples/src/main/java/io/aeron/samples/BacklogStat.java
public Map<StreamCompositeKey, StreamBacklog> snapshot()
    {
        final Map<StreamCompositeKey, StreamBacklog> streams = new HashMap<>();

        counters.forEach(
            (counterId, typeId, keyBuffer, label) ->
            {
                if ((typeId >= PUBLISHER_LIMIT_TYPE_ID && typeId <= RECEIVER_POS_TYPE_ID) ||
                    typeId == SENDER_LIMIT_TYPE_ID || typeId == PER_IMAGE_TYPE_ID || typeId == PUBLISHER_POS_TYPE_ID)
                {
                    final StreamCompositeKey key = new StreamCompositeKey(
                        keyBuffer.getInt(SESSION_ID_OFFSET),
                        keyBuffer.getInt(STREAM_ID_OFFSET),
                        keyBuffer.getStringAscii(CHANNEL_OFFSET));

                    final StreamBacklog streamBacklog = streams.computeIfAbsent(key, (ignore) -> new StreamBacklog());
                    final long registrationId = keyBuffer.getLong(REGISTRATION_ID_OFFSET);
                    final long value = counters.getCounterValue(counterId);
                    switch (typeId)
                    {
                        case PublisherLimit.PUBLISHER_LIMIT_TYPE_ID:
                            streamBacklog.createPublisherIfAbsent().registrationId(registrationId);
                            streamBacklog.createPublisherIfAbsent().limit(value);
                            break;

                        case PublisherPos.PUBLISHER_POS_TYPE_ID:
                            streamBacklog.createPublisherIfAbsent().registrationId(registrationId);
                            streamBacklog.createPublisherIfAbsent().position(value);
                            break;

                        case SenderPos.SENDER_POSITION_TYPE_ID:
                            streamBacklog.createSenderIfAbsent().registrationId(registrationId);
                            streamBacklog.createSenderIfAbsent().position(value);
                            break;

                        case SenderLimit.SENDER_LIMIT_TYPE_ID:
                            streamBacklog.createSenderIfAbsent().registrationId(registrationId);
                            streamBacklog.createSenderIfAbsent().limit(value);
                            break;

                        case ReceiverHwm.RECEIVER_HWM_TYPE_ID:
                            streamBacklog.createReceiverIfAbsent().registrationId(registrationId);
                            streamBacklog.createReceiverIfAbsent().highWaterMark(value);
                            break;

                        case ReceiverPos.RECEIVER_POS_TYPE_ID:
                            streamBacklog.createReceiverIfAbsent().registrationId(registrationId);
                            streamBacklog.createReceiverIfAbsent().position(value);
                            break;

                        case SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID:
                            streamBacklog.subscriberBacklogs().put(registrationId, new Subscriber(value));
                            break;
                    }
                }
            });

        return streams;
    }
public Map<StreamCompositeKey, StreamBacklog> snapshot()
    {
        final Map<StreamCompositeKey, StreamBacklog> streams = new HashMap<>();

        counters.forEach(
            (counterId, typeId, keyBuffer, label) ->
            {
                if ((typeId >= PUBLISHER_LIMIT_TYPE_ID && typeId <= RECEIVER_POS_TYPE_ID) ||
                    typeId == SENDER_LIMIT_TYPE_ID || typeId == PER_IMAGE_TYPE_ID || typeId == PUBLISHER_POS_TYPE_ID)
                {
                    final StreamCompositeKey key = new StreamCompositeKey(
                        keyBuffer.getInt(SESSION_ID_OFFSET),
                        keyBuffer.getInt(STREAM_ID_OFFSET),
                        keyBuffer.getStringAscii(CHANNEL_OFFSET));

                    final StreamBacklog streamBacklog = streams.computeIfAbsent(key, (ignore) -> new StreamBacklog());
                    final long registrationId = keyBuffer.getLong(REGISTRATION_ID_OFFSET);
                    final long value = counters.getCounterValue(counterId);
                    switch (typeId)
                    {
                        case PublisherLimit.PUBLISHER_LIMIT_TYPE_ID:
                            streamBacklog.createPublisherIfAbsent().registrationId(registrationId);
                            streamBacklog.createPublisherIfAbsent().limit(value);
                            break;

                        case PublisherPos.PUBLISHER_POS_TYPE_ID:
                            streamBacklog.createPublisherIfAbsent().registrationId(registrationId);
                            streamBacklog.createPublisherIfAbsent().position(value);
                            break;

                        case SenderPos.SENDER_POSITION_TYPE_ID:
                            streamBacklog.createSenderIfAbsent().registrationId(registrationId);
                            streamBacklog.createSenderIfAbsent().position(value);
                            break;

                        case SenderLimit.SENDER_LIMIT_TYPE_ID:
                            streamBacklog.createSenderIfAbsent().registrationId(registrationId);
                            streamBacklog.createSenderIfAbsent().limit(value);
                            break;

                        case ReceiverHwm.RECEIVER_HWM_TYPE_ID:
                            streamBacklog.createReceiverIfAbsent().registrationId(registrationId);
                            streamBacklog.createReceiverIfAbsent().highWaterMark(value);
                            break;

                        case ReceiverPos.RECEIVER_POS_TYPE_ID:
                            streamBacklog.createReceiverIfAbsent().registrationId(registrationId);
                            streamBacklog.createReceiverIfAbsent().position(value);
                            break;

                        case SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID:
                            streamBacklog.subscriberBacklogs().put(registrationId, new Subscriber(value));
                            break;
                    }
                }
            });

        return streams;
    }
public Map < StreamCompositeKey , StreamBacklog > snapshot ( ) { final Map < StreamCompositeKey , StreamBacklog > streams = new HashMap <> ( ) ; counters . forEach ( ( counterId , typeId , keyBuffer , label ) -> { if ( ( typeId >= PUBLISHER_LIMIT_TYPE_ID && typeId <= RECEIVER_POS_TYPE_ID ) || typeId == SENDER_LIMIT_TYPE_ID || typeId == PER_IMAGE_TYPE_ID || typeId == PUBLISHER_POS_TYPE_ID ) { final StreamCompositeKey key = new StreamCompositeKey ( keyBuffer . getInt ( SESSION_ID_OFFSET ) , keyBuffer . getInt ( STREAM_ID_OFFSET ) , keyBuffer . getStringAscii ( CHANNEL_OFFSET ) ) ; final StreamBacklog streamBacklog = streams . computeIfAbsent ( key , ( ignore ) - > new StreamBacklog ( ) ) ; final long registrationId = keyBuffer . getLong ( REGISTRATION_ID_OFFSET ) ; final long value = counters . getCounterValue ( counterId ) ; switch ( typeId ) { case PublisherLimit . PUBLISHER_LIMIT_TYPE_ID : streamBacklog . createPublisherIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createPublisherIfAbsent ( ) . limit ( value ) ; break ; case PublisherPos . PUBLISHER_POS_TYPE_ID : streamBacklog . createPublisherIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createPublisherIfAbsent ( ) . position ( value ) ; break ; case SenderPos . SENDER_POSITION_TYPE_ID : streamBacklog . createSenderIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createSenderIfAbsent ( ) . position ( value ) ; break ; case SenderLimit . SENDER_LIMIT_TYPE_ID : streamBacklog . createSenderIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createSenderIfAbsent ( ) . limit ( value ) ; break ; case ReceiverHwm . RECEIVER_HWM_TYPE_ID : streamBacklog . createReceiverIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createReceiverIfAbsent ( ) . highWaterMark ( value ) ; break ; case ReceiverPos . RECEIVER_POS_TYPE_ID : streamBacklog . createReceiverIfAbsent ( ) . registrationId ( registrationId ) ; streamBacklog . createReceiverIfAbsent ( ) . position ( value ) ; break ; case SubscriberPos . SUBSCRIBER_POSITION_TYPE_ID : streamBacklog . subscriberBacklogs ( ) . put ( registrationId , new Subscriber ( value ) ) ; break ; } } } ) ; return streams ; }
Take a snapshot of all the backlog information and group by stream.

@return a snapshot of all the backlog information and group by stream.
Take a snapshot of all the backlog information and group by stream .