protected void sendMergeView ( Collection < Address > coords , MergeData combined_merge_data , MergeId merge_id ) { if ( coords == null || coords . isEmpty ( ) || combined_merge_data == null ) return ; View view = combined_merge_data . view ; Digest digest = combined_merge_data . digest ; if ( view == null || digest == null ) { log . error ( Util . getMessage ( "ViewOrDigestIsNullCannotSendConsolidatedMergeView/Digest" ) ) ; return ; } int size = 0 ; if ( gms . flushProtocolInStack ) { gms . merge_ack_collector . reset ( coords ) ; size = gms . merge_ack_collector . size ( ) ; } Event install_merge_view_evt = new Event ( Event . INSTALL_MERGE_VIEW , view ) ; gms . getUpProtocol ( ) . up ( install_merge_view_evt ) ; gms . getDownProtocol ( ) . down ( install_merge_view_evt ) ; long start = System . currentTimeMillis ( ) ; for ( Address coord : coords ) { Message msg = new Message ( coord ) . setBuffer ( GMS . marshal ( view , digest ) ) . putHeader ( gms . getId ( ) , new GMS . GmsHeader ( GMS . GmsHeader . INSTALL_MERGE_VIEW ) . mergeId ( merge_id ) ) ; gms . getDownProtocol ( ) . down ( msg ) ; } //[JGRP-700] - FLUSH: flushing should span merge; if flush is in stack, wait for acks from subview coordinators if ( gms . flushProtocolInStack ) { try { gms . merge_ack_collector . waitForAllAcks ( gms . view_ack_collection_timeout ) ; log . trace ( "%s: received all ACKs (%d) for merge view %s in %d ms" , gms . local_addr , size , view , ( System . currentTimeMillis ( ) - start ) ) ; } catch ( TimeoutException e ) { log . warn ( "%s: failed to collect all ACKs (%d) for merge view %s after %d ms, missing ACKs from %s" , gms . local_addr , size , view , gms . view_ack_collection_timeout , gms . merge_ack_collector . printMissing ( ) ) ; } } }
Sends the new view and digest to all subgroup coordinators . Each coord will in turn broadcast the new view and digest to all the members of its subgroup