private void update ( Collection < InternalSubscriptionInfo > subscriptions ) { for ( InternalSubscriptionInfo subscription : subscriptions ) { InternalTopic topic = topics . computeIfAbsent ( subscription . topic , InternalTopic :: new ) ; InternalSubscriptionInfo matchingSubscription = topic . remoteSubscriptions ( ) . stream ( ) . filter ( s -> s . memberId ( ) . equals ( subscription . memberId ( ) ) && s . logicalTimestamp ( ) . equals ( subscription . logicalTimestamp ( ) ) ) . findFirst ( ) . orElse ( null ) ; if ( matchingSubscription == null ) { topic . addRemoteSubscription ( subscription ) ; } else if ( subscription . isTombstone ( ) ) { topic . removeRemoteSubscription ( subscription ) ; } } }
Handles a collection of subscription updates received via the gossip protocol .