public < T > void invokeNodeRequest ( final NodeRequest request , final NodeRequestsListener < T > listener ) { // if we're not on the dobjmgr thread, get there if ( ! _omgr . isDispatchThread ( ) ) { _omgr . postRunnable ( new Runnable ( ) { public void run ( ) { invokeNodeRequest ( request , listener ) ; } } ) ; return ; } // serialize the action to make sure we can byte [ ] requestBytes = flattenRequest ( request ) ; // build a set of node names (including the local node) to which to send the request final Set < String > nodes = findApplicableNodes ( request ) ; if ( nodes . isEmpty ( ) ) { listener . requestsProcessed ( new NodeRequestsResultImpl < T > ( ) ) ; return ; } final Map < String , T > results = Maps . newHashMap ( ) ; final Map < String , String > failures = Maps . newHashMap ( ) ; final AtomicInteger completedNodes = new AtomicInteger ( ) ; for ( final String node : nodes ) { invokeNodeRequest ( node , requestBytes , new InvocationService . ResultListener ( ) { public void requestProcessed ( Object result ) { // check off this node's successful response @ SuppressWarnings ( "unchecked" ) T castResult = ( T ) result ; results . put ( node , castResult ) ; nodeDone ( ) ; } public void requestFailed ( String cause ) { failures . put ( node , cause ) ; nodeDone ( ) ; } protected void nodeDone ( ) { if ( completedNodes . incrementAndGet ( ) == nodes . size ( ) ) { // if all nodes have responded, let caller know listener . requestsProcessed ( new NodeRequestsResultImpl < T > ( results , failures ) ) ; } } } ) ; } }
Invokes the supplied request on all servers in parallel . The request will execute on the distributed object thread but this method does not need to be called from there .