public void askSubscribers ( Actor sender , String topic , Object message , Callback cb ) { List < ReceiverActor > subscriber = topic2Subscriber . get ( topic ) ; if ( subscriber != null ) { List < IPromise > results = subscriber . stream ( ) . filter ( subs -> ! subs . equals ( sender ) ) . map ( subs -> subs . receiveAsk ( topic , message ) ) . collect ( Collectors . toList ( ) ) ; try { all ( ( List ) results ) . await ( 5000 ) ; // is non-blocking } catch ( Exception ex ) { // timeout goes here Log . Info ( this , "timeout in broadcast" ) ; } results . forEach ( promise -> cb . pipe ( promise . get ( ) ) ) ; cb . finish ( ) ; // important to release callback mapping in remoting ! } }
send a message to all and stream the result of each receiver back to sender