private CompletableFuture < Void > updateNode ( Member member ) { long updateTime = System . currentTimeMillis ( ) ; long lastUpdateTime = updateTimes . getOrDefault ( member . id ( ) , 0L ) ; Collection < InternalSubscriptionInfo > subscriptions = topics . values ( ) . stream ( ) . flatMap ( t -> t . remoteSubscriptions ( ) . stream ( ) . filter ( subscriber -> subscriber . timestamp ( ) . unixTimestamp ( ) >= lastUpdateTime ) ) . collect ( Collectors . toList ( ) ) ; CompletableFuture < Void > future = new CompletableFuture <> ( ) ; messagingService . sendAndReceive ( member . address ( ) , GOSSIP_MESSAGE_SUBJECT , SERIALIZER . encode ( subscriptions ) ) . whenComplete ( ( result , error ) -> { if ( error == null ) { updateTimes . put ( member . id ( ) , updateTime ) ; } future . complete ( null ) ; } ) ; return future ; }
Sends an update to the given node .