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;
    }
Take a snapshot of all the backlog information and group by stream.

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