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 .