public Map<StreamCompositeKey, List<StreamPosition>> snapshot()
    {
        final Map<StreamCompositeKey, List<StreamPosition>> streams = new TreeMap<>(LINES_COMPARATOR);
        final Object2ObjectHashMap<StreamCompositeKey, String> fullChannelByStream = new Object2ObjectHashMap<>();

        counters.forEach(
            (counterId, typeId, keyBuffer, label) ->
            {
                if ((typeId >= PUBLISHER_LIMIT_TYPE_ID && typeId <= RECEIVER_POS_TYPE_ID) ||
                    typeId == SENDER_LIMIT_TYPE_ID || typeId == PUBLISHER_POS_TYPE_ID)
                {
                    final int channelLength = keyBuffer.getInt(CHANNEL_OFFSET, LITTLE_ENDIAN);
                    final String channel =
                        keyBuffer.getStringWithoutLengthAscii(CHANNEL_OFFSET + SIZE_OF_INT, channelLength);

                    final int uriIndex = label.indexOf("aeron:");
                    final String fullChannel;
                    if (uriIndex >= 0)
                    {
                        int joinPositionIndex = -1;
                        if (SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID == typeId &&
                            (joinPositionIndex = label.lastIndexOf(" @")) > uriIndex)
                        {
                            fullChannel = label.substring(uriIndex, joinPositionIndex);
                        }
                        else
                        {
                            fullChannel = label.substring(uriIndex);
                        }
                    }
                    else
                    {
                        fullChannel = channel;
                    }

                    final StreamCompositeKey key = new StreamCompositeKey(
                        keyBuffer.getInt(SESSION_ID_OFFSET), keyBuffer.getInt(STREAM_ID_OFFSET), channel);

                    final StreamPosition position = new StreamPosition(
                        keyBuffer.getLong(REGISTRATION_ID_OFFSET), counters.getCounterValue(counterId), typeId);

                    List<StreamPosition> positions = streams.get(key);
                    if (null == positions)
                    {
                        positions = new ArrayList<>();
                        streams.put(key, positions);
                    }
                    positions.add(position);

                    final String existingFullChannel = fullChannelByStream.get(key);
                    if (null == existingFullChannel || fullChannel.length() > existingFullChannel.length())
                    {
                        fullChannelByStream.put(key, fullChannel);
                    }
                }
            });

        streams.keySet().forEach(k -> k.fullChannel = fullChannelByStream.get(k));

        return streams;
    }
@Test
    @SuppressWarnings("MethodLength")
    @InterruptAfter(5)
    void shouldCollectStreamStats()
    {
        final ExclusivePublication pub1 = aeron.addExclusivePublication(
            "aeron:ipc?alias=test-should-collect-stats|term-length=65536|tether=true|reliable=true|" +
            "ssc=true|sparse=false|linger=60s|mtu=4096",
            2000);
        final ConcurrentPublication pub2 =
            aeron.addPublication("aeron:udp?session-id=" + (pub1.sessionId() - 333) +
            "|endpoint=localhost:5555|mtu=2k|term-length=128K",
            909);
        final ConcurrentPublication pub3 = aeron.addPublication(pub2.channel(), pub2.streamId());
        final ExclusivePublication pub4 = aeron.addExclusivePublication(
            "aeron:udp?endpoint=localhost:7777|interface=127.0.0.1|init-term-id=-6548658|term-id=42|" +
            "term-offset=1024|mtu=8192|ttl=5|group=true|term-length=256K|so-rcvbuf=128K|so-sndbuf=128K|rcv-wnd=128K|" +
            "sparse=true|linger=0|nak-delay=100ms|tether=false|reliable=false|ssc=false|" +
            Tests.generateStringWithSuffix("alias=", "x", 300),
            500);
        final ExclusivePublication pub5 = aeron.addExclusivePublication(
            "aeron:udp?session-id=" + pub2.sessionId() +
            "|alias= @test @not @split @channel @uri @by @at @even @if @no @subscription @exists|" +
            "endpoint=localhost:6666|interface=127.0.0.1|term-length=512K|mtu=1376", 1);
        assertEquals(pub1.sessionId() + 1, pub4.sessionId());
        assertEquals(pub1.sessionId() - 333, pub2.sessionId());
        assertEquals(pub2.sessionId(), pub3.sessionId());
        assertNotSame(pub2, pub3);
        assertEquals(pub2.sessionId(), pub5.sessionId());

        final Subscription sub1 = aeron.addSubscription(pub1.channel(), pub1.streamId());
        final Subscription sub2 = aeron.addSubscription(pub2.channel(), pub2.streamId());
        final Subscription sub4 = aeron.addSubscription(pub4.channel(), pub4.streamId());

        Tests.awaitConnected(pub1);
        Tests.awaitConnected(pub2);
        Tests.awaitConnected(pub3);
        Tests.awaitConnected(pub4);
        Tests.awaitConnected(sub1);
        Tests.awaitConnected(sub2);
        Tests.awaitConnected(sub4);

        final int sub2RcvPosCounterId = findCounterIdByStream(
            aeron.countersReader(), ReceiverPos.RECEIVER_POS_TYPE_ID, pub2.streamId());
        final int sub4RcvPosCounterId = findCounterIdByStream(
            aeron.countersReader(), ReceiverPos.RECEIVER_POS_TYPE_ID, pub4.streamId());

        final UnsafeBuffer msg = new UnsafeBuffer(new byte[128]);
        ThreadLocalRandom.current().nextBytes(msg.byteArray());
        for (int i = 0; i < 5; i++)
        {
            while ((pub1.offer(msg) < 0))
            {
                Tests.yield();
            }
            while ((pub2.offer(msg, 0, 64) < 0))
            {
                Tests.yield();
            }
            while ((pub3.offer(msg, 0, 80) < 0))
            {
                Tests.yield();
            }
            while ((pub4.offer(msg, 105, 13) < 0))
            {
                Tests.yield();
            }
        }

        final Subscription sub5 = aeron.addSubscription(pub4.channel(), pub4.streamId());
        assertNotSame(sub4, sub5);
        final CountingFragmentHandler sub1Handler = new CountingFragmentHandler(sub1);
        final CountingFragmentHandler sub2Handler = new CountingFragmentHandler(sub2);
        final CountingFragmentHandler sub4Handler = new CountingFragmentHandler(sub4);

        sub1Handler.pollUntil(3);
        sub2Handler.pollUntil(7);
        sub4Handler.pollUntil(5);

        while (1120 != aeron.countersReader().getCounterValue(sub2RcvPosCounterId))
        {
            Tests.yield();
        }

        while (1716702414144L != aeron.countersReader().getCounterValue(sub4RcvPosCounterId))
        {
            Tests.yield();
        }

        final StreamStat streamStat = new StreamStat(aeron.countersReader());
        final Map<StreamStat.StreamCompositeKey, List<StreamStat.StreamPosition>> snapshot = streamStat.snapshot();
        assertInstanceOf(SortedMap.class, snapshot);
        final List<Map.Entry<StreamStat.StreamCompositeKey, List<StreamStat.StreamPosition>>> entries =
            new ArrayList<>(snapshot.entrySet());

        Map.Entry<StreamStat.StreamCompositeKey, List<StreamStat.StreamPosition>> entry = entries.get(0);
        assertStreamInfo(
            entry.getKey(),
            pub5.sessionId(),
            pub5.streamId(),
            pub5.channel().substring(0, StreamCounter.MAX_CHANNEL_LENGTH),
            pub5.channel());
        assertEquals(4, entry.getValue().size());
        assertStreamPosition(entry.getValue().get(0), PublisherPos.PUBLISHER_POS_TYPE_ID, 0);
        assertStreamPosition(entry.getValue().get(1), PublisherLimit.PUBLISHER_LIMIT_TYPE_ID, 0);
        assertStreamPosition(entry.getValue().get(2), SenderPos.SENDER_POSITION_TYPE_ID, 0);
        assertStreamPosition(entry.getValue().get(3), SenderLimit.SENDER_LIMIT_TYPE_ID, 0);

        entry = entries.get(1);
        assertStreamInfo(entry.getKey(), pub2.sessionId(), pub2.streamId(), pub2.channel(), pub2.channel());
        assertEquals(7, entry.getValue().size());
        assertStreamPosition(entry.getValue().get(0), PublisherPos.PUBLISHER_POS_TYPE_ID, 1120);
        assertStreamPosition(entry.getValue().get(1), PublisherLimit.PUBLISHER_LIMIT_TYPE_ID, 66656);
        assertStreamPosition(entry.getValue().get(2), SenderPos.SENDER_POSITION_TYPE_ID, 1120);
        assertStreamPosition(entry.getValue().get(3), SenderLimit.SENDER_LIMIT_TYPE_ID, 65536);
        assertStreamPosition(entry.getValue().get(4), SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID, 448);
        assertStreamPosition(entry.getValue().get(5), ReceiverHwm.RECEIVER_HWM_TYPE_ID, 1120);
        assertStreamPosition(entry.getValue().get(6), ReceiverPos.RECEIVER_POS_TYPE_ID, 1120);

        entry = entries.get(2);
        assertStreamInfo(
            entry.getKey(),
            pub1.sessionId(),
            pub1.streamId(),
            pub1.channel().substring(0, StreamCounter.MAX_CHANNEL_LENGTH),
            pub1.channel());
        assertEquals(3, entry.getValue().size());
        assertStreamPosition(entry.getValue().get(0), PublisherPos.PUBLISHER_POS_TYPE_ID, 800);
        assertStreamPosition(entry.getValue().get(1), PublisherLimit.PUBLISHER_LIMIT_TYPE_ID, 32768);
        assertStreamPosition(entry.getValue().get(2), SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID, 320);

        entry = entries.get(3);
        final int pub4LimitCounterId = aeron.countersReader()
            .findByTypeIdAndRegistrationId(PublisherLimit.PUBLISHER_LIMIT_TYPE_ID, pub4.registrationId());
        assertNotEquals(CountersReader.NULL_COUNTER_ID, pub4LimitCounterId);
        final String pub4LimitLabel = aeron.countersReader().getCounterLabel(pub4LimitCounterId);
        assertStreamInfo(
            entry.getKey(),
            pub4.sessionId(),
            pub4.streamId(),
            pub4.channel().substring(0, StreamCounter.MAX_CHANNEL_LENGTH),
            pub4LimitLabel.substring(pub4LimitLabel.indexOf("aeron:")));
        assertEquals(8, entry.getValue().size());
        assertStreamPosition(entry.getValue().get(0), PublisherPos.PUBLISHER_POS_TYPE_ID, 1716702414144L);
        assertStreamPosition(entry.getValue().get(1), PublisherLimit.PUBLISHER_LIMIT_TYPE_ID, 1716702545216L);
        assertStreamPosition(entry.getValue().get(2), SenderPos.SENDER_POSITION_TYPE_ID, 1716702414144L);
        assertStreamPosition(entry.getValue().get(3), SenderLimit.SENDER_LIMIT_TYPE_ID, 1716702544896L);
        assertStreamPosition(entry.getValue().get(4), SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID, 1716702414016L);
        assertStreamPosition(entry.getValue().get(5), ReceiverHwm.RECEIVER_HWM_TYPE_ID, 1716702414144L);
        assertStreamPosition(entry.getValue().get(6), ReceiverPos.RECEIVER_POS_TYPE_ID, 1716702414144L);
        assertStreamPosition(entry.getValue().get(7), SubscriberPos.SUBSCRIBER_POSITION_TYPE_ID, 1716702413824L);
    }