import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
@Override
public boolean cancelTask(String hostname) {
    try {
        log.debug("Calling cancelTaskRpc for {}...", hostname);
        MethodCall call = new MethodCall(getClass().getMethod("cancelTaskRpc", String.class));
        RequestOptions ops = new RequestOptions(ResponseMode.GET_ALL, 5000);
        call.setArgs(hostname);
        RspList<CancelTaskRpcResponse> responses = rpcDispatcher.callRemoteMethods(
            rpcDispatcher.getChannel()
                         .getView()
                         .getMembers()
                         .stream()
                         .filter(h -> h.toString().equalsIgnoreCase(hostname))
                         .collect(Collectors.toList()),
            call, ops);
        log.debug("Got answer: {}", responses);
        return responses.getResults()
                        .stream()
                        .anyMatch(CancelTaskRpcResponse::isCancelled);
    } catch (Exception e) {
        log.catching(e);
        return false;
    }
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public void mouseReleased(MouseEvent e) {
    Point p=e.getPoint();
    if(pick == null)
        return;

    pick.x=p.x;
    pick.y=p.y;
    pick.fixed=pickfixed;


    try {
        MethodCall call=new MethodCall("moveNode", new Object[]{pick}, new Class[]{Node.class});
        wb.disp.callRemoteMethods(null, call, new RequestOptions(ResponseMode.GET_ALL, 0));
    }
    catch(Exception ex) {
        log.error(ex.toString());
    }

    pick=null;
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public void start(String name) {
    myname=name;
    int xloc=(int)(10 + 250 * Math.random());
    int yloc=(int)(10 + 250 * Math.random());

    try {
        MethodCall call=new MethodCall("addNode",
                                       new Object[]{name,my_addr,new Integer(xloc),new Integer(yloc)},
                                       new Class[]{String.class,Address.class,int.class,int.class});
        wb.disp.callRemoteMethods(null, call, new RequestOptions(ResponseMode.GET_ALL, 0));
    }
    catch(Exception e) {
        log.error(e.toString());
    }
    repaint();
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
/**
 * Tests the deadlock resolution using self-calls on a single channel. The deadlock detection
 * is turned on so the method call should go straight through. If there is a problem, JUnit will
 * timeout.
 *
 * @throws Exception
 */
public void testOneChannel() throws Exception {
    c1 = createChannel(true);
    ServerObject serverObject = new ServerObject("obj1");
    RpcDispatcher disp=new RpcDispatcher(c1, serverObject);
    serverObject.setRpcDispatcher(disp);
    c1.connect(name);
    Address localAddress = c1.getAddress();

    // call the nested group method on itself
    MethodCall call = new MethodCall("outerMethod", new Object[0], new Class[0]);
    log("calling outerMethod() on all members");
    RspList rspList = disp.callRemoteMethods(null, call, new RequestOptions(ResponseMode.GET_ALL, 0));
    log("results of outerMethod(): " + rspList);

    Assert.assertEquals(1, rspList.size());
    assertEquals("outerMethod[innerMethod]", rspList.getValue(localAddress));
    assertTrue(rspList.isReceived(localAddress));
    assertFalse(rspList.isSuspected(localAddress));
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
/**
 * Tests the deadlock resolution using two different channels. The deadlock detection
 * is turned on. It implements the following scenario:
 *
 * Channel1                              Channel2
 *    |                                     |
 *    + -------------------------------> outerMethod()
 *    |                                    RPC
 *    |                                     |
 *    |                                     |
 *    |                                     |
 *    | <-- innerMethod() <-----------------+ ---------+
 *    |                                     |          |
 *    |                                     | <-- innerMethod()
 *
 * If there is a deadlock, JUnit will timeout and fail the test.
 *
 */
public void testTwoChannels() throws Throwable {
    ServerObject obj1, obj2 = null;

    c1 = createChannel(true);
    obj1 = new ServerObject("obj1");
    RpcDispatcher disp1=new RpcDispatcher(c1, obj1);
    obj1.setRpcDispatcher(disp1);
    c1.connect(name);

    c2 = createChannel(c1);
    obj2 = new ServerObject("obj2");
    RpcDispatcher disp2=new RpcDispatcher(c2, obj2);
    obj2.setRpcDispatcher(disp2);
    c2.connect(name);
    Address localAddress2 = c2.getAddress();

    // call a point-to-point method on Member 2 that triggers a nested distributed RPC
    MethodCall call = new MethodCall("outerMethod", new Object[0], new Class[0]);
    log("calling outerMethod() on " + localAddress2);
    Object retval = disp1.callRemoteMethod(localAddress2, call, new RequestOptions(ResponseMode.GET_ALL, 0));
    log("results of outerMethod(): " + retval);
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public void testTwoChannelsWithInitialMulticast() throws Exception {
    ServerObject obj1, obj2 = null;

    c1 = createChannel(true);
    obj1 = new ServerObject("obj1");
    RpcDispatcher disp1=new RpcDispatcher(c1, obj1);
    obj1.setRpcDispatcher(disp1);
    c1.connect(name);

    c2 = createChannel(c1);
    obj2 = new ServerObject("obj2");
    RpcDispatcher disp2=new RpcDispatcher(c2, obj2);
    obj2.setRpcDispatcher(disp2);
    c2.connect(name);

    Vector<Address> dests=new Vector<>();
    dests.add(c1.getAddress());
    dests.add(c2.getAddress());

    // call a point-to-point method on Member 2 that triggers a nested distributed RPC
    MethodCall call = new MethodCall("outerMethod", new Object[0], new Class[0]);
    log("calling outerMethod() on all members");
    RspList rsps = disp1.callRemoteMethods(dests, call, new RequestOptions(ResponseMode.GET_ALL, 0));
    log("results of outerMethod():\n" + rsps);
    Assert.assertEquals(2, rsps.size());
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public String outerMethod() throws Exception {
    log("**** outerMethod() received, calling innerMethod() on all members");
    MethodCall call = new MethodCall("innerMethod", new Object[0], new Class[0]);
    // RspList rspList = disp.callRemoteMethods(null, call, GroupResponseMode.GET_ALL, 5000);
    RequestOptions opts=new RequestOptions(ResponseMode.GET_ALL, 0, false, null, (Message.Flag[])null);
    opts.setFlags(Message.Flag.OOB);
    RspList<String> rspList = disp.callRemoteMethods(null, call, opts);
    List<String> results = rspList.getResults();
    log("results of calling innerMethod():\n" + rspList);
    StringBuilder sb=new StringBuilder("outerMethod[");
    boolean first=true;
    for(String s: results) {
        if(first)
            first=false;
        else
            sb.append(";");
        sb.append(s);
    }
    sb.append("]");
    return sb.toString();
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
@Override
public <T> T remoteExecute(MethodCall action, long timeout) {
  logTrace(() -> String.format("RemoteExecute sync action %s with timeout %s", action, timeout));
  RequestOptions options = new RequestOptions()
      .setFlags(JGROUPS_FLAGS)
      .setMode(ResponseMode.GET_ALL)
      .setTimeout(timeout);

  try {
    NotifyingFuture<RspList<T>> notifyingFuture = this.<T>execute(action, options);
    RspList<T> rspList = notifyingFuture.get(timeout, TimeUnit.MILLISECONDS);
    return futureDone(rspList);
  } catch (Exception e) {
    throw new VertxException(e);
  }
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
protected Address whereIsState(Serializable stateId, long timeout) throws Exception {
	if (remoteMembers.isEmpty()) {
		return null;
	}
	RspList<Boolean> resp = this.disp.callRemoteMethods(getRemoteMembersCopy(), new MethodCall((short)(methodMap.size() - 5), new Object[]{stateId}), new RequestOptions(ResponseMode.GET_ALL, timeout));
	Collection<Rsp<Boolean>> values = resp.values();
	Rsp<Boolean> rsp = null;
	for (Rsp<Boolean> response : values) {
		if (Boolean.TRUE.equals(response.getValue())) {
			rsp = response;
			break;
		}
	}
	if (rsp == null) {
		return null;
	}
	return rsp.getSender();
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
private void sendMessage(int size) throws Exception {
    long start, stop;
    MyHandler handler=new MyHandler(new byte[size]);
    d1.setRequestHandler(handler);
    start=System.currentTimeMillis();
    RspList rsps=d1.castMessage(null, new Message(), new RequestOptions(ResponseMode.GET_ALL, 0));
    stop=System.currentTimeMillis();
    System.out.println("rsps:\n" + rsps);
    System.out.println("call took " + (stop - start) + " ms");
    assertNotNull(rsps);
    Assert.assertEquals(1, rsps.size());
    byte[] buf=(byte[])rsps.getFirst();
    assertNotNull(buf);
    Assert.assertEquals(size, buf.length);
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
private void sendMessageToBothChannels(int size) throws Exception {
    long start, stop;
    d1.setRequestHandler(new MyHandler(new byte[size]));

    b=createChannel(a);
    b.setName("B");
    d2=new MessageDispatcher(b, null, null, new MyHandler(new byte[size]));
    b.connect("MessageDispatcherUnitTest");
    Assert.assertEquals(2,b.getView().size());

    System.out.println("casting message");
    start=System.currentTimeMillis();
    RspList rsps=d1.castMessage(null, new Message(), new RequestOptions(ResponseMode.GET_ALL, 0));
    stop=System.currentTimeMillis();
    System.out.println("rsps:\n" + rsps);
    System.out.println("call took " + (stop - start) + " ms");
    assertNotNull(rsps);
    Assert.assertEquals(2,rsps.size());
    Rsp rsp=rsps.get(a.getAddress());
    assertNotNull(rsp);
    byte[] ret=(byte[])rsp.getValue();
    Assert.assertEquals(size, ret.length);

    rsp=rsps.get(b.getAddress());
    assertNotNull(rsp);
    ret=(byte[])rsp.getValue();
    Assert.assertEquals(size, ret.length);

    Util.close(b);
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public void testLeaveDuringSend() throws Exception {
    final JChannel[]          channels = new JChannel[NUM];
    final MessageDispatcher[] dispatchers = new MessageDispatcher[NUM];

    for(int i=0; i < NUM; i++) {
        channels[i]= new JChannel(new SHARED_LOOPBACK(),
                                  new SHARED_LOOPBACK_PING(),
                                  new MERGE3(),
                                  new FD().setValue("timeout", 1000).setValue("max_tries", 1),
                                  new NAKACK2(),
                                  new UNICAST3(),
                                  new STABLE(),
                                  new GMS(),
                                  new RSVP().setValue("ack_on_delivery", false)
                                    .setValue("throw_exception_on_timeout", false));
        channels[i].setName(Character.toString((char) ('A' + i)));
        channels[i].setDiscardOwnMessages(true);
        dispatchers[i]=new MessageDispatcher(channels[i], null, null, new MyRequestHandler());
        channels[i].connect("DynamicDiscardTest");
        System.out.print(i + 1 + " ");
    }
    Util.waitUntilAllChannelsHaveSameSize(10000, 1000, channels);

    // discard all messages (except those to self)
    DISCARD discard = new DISCARD();
    channels[0].getProtocolStack().insertProtocol(discard, ProtocolStack.ABOVE, TP.class);
    discard.setDiscardAll(true);

    // send a RSVP message
    Message msg = new Message(null, "message2");
    msg.setFlag(Message.Flag.RSVP, Message.Flag.OOB);
    RspList<Object> rsps = dispatchers[0].castMessage(null, msg, RequestOptions.SYNC().setTimeout(5000));

    Rsp<Object> objectRsp = rsps.get(channels[1].getAddress());
    assertFalse(objectRsp.wasReceived());
    assertTrue(objectRsp.wasSuspected());
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
private <T> NotifyingFuture<RspList<T>> internalExecute(MethodCall action, RequestOptions options) throws Exception {
  if (active) {
    return dispatcher.<T>callRemoteMethodsWithFuture(null, action, options);
  } else {
    throw new VertxException("Executor service is closed");
  }
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public JGroupsOutputStream(RpcDispatcher disp, List<Address> dests, Serializable stateId, short methodOffset, boolean sendCreate) throws IOException {
    this.disp=disp;
    this.dests=dests;
    this.stateId=stateId;
    this.methodOffset = methodOffset;
    if (sendCreate) {
     try {
     	disp.callRemoteMethods(this.dests, new MethodCall(methodOffset, new Object[] {stateId}), new RequestOptions(ResponseMode.GET_NONE, 0).setAnycasting(dests != null));
     } catch(Exception e) {
     	throw new IOException(e);
     }
    }
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public void close() throws IOException {
    if(closed) {
        return;
    }
    flush();
    try {
    	disp.callRemoteMethods(dests, new MethodCall((short)(methodOffset + 2), new Object[] {stateId}), new RequestOptions(ResponseMode.GET_NONE, 0).setAnycasting(dests != null));
    } catch(Exception e) {
    }
    closed=true;
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
public void flush() throws IOException {
    checkClosed();
    try {
        if(index == 0) {
            return;
        }
    	disp.callRemoteMethods(dests, new MethodCall((short)(methodOffset + 1), new Object[] {stateId, Arrays.copyOf(buffer, index)}), new RequestOptions(ResponseMode.GET_NONE, 0).setAnycasting(dests != null));
        index=0;
    } catch(Exception e) {
    	throw new IOException(e);
    }
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
@Override
public void sendRequest(long requestId, Collection<Address> dest_mbrs, Message msg, RspCollector coll, RequestOptions options) throws Exception {
    msg.putHeader(MUX_ID, header);
    super.sendRequest(requestId, dest_mbrs, msg, coll, options);
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
RspList sendGroupRpc() throws Exception {
    return disp.callRemoteMethods(null, "print", new Object[]{new Integer(i++)}, new Class[] {int.class},
                                  new RequestOptions(ResponseMode.GET_ALL, 0));
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
protected void invoke(RpcDispatcher disp, Address target, int num_bytes) throws Exception {
    // B invokes (blocking) A.forward
    disp.callRemoteMethod(a.getAddress(), new MethodCall(FORWARD, target, num_bytes), RequestOptions.SYNC().setTimeout(5000));
}
 

import org.jgroups.blocks.RequestOptions; //导入依赖的package包/类
private <T> NotifyingFuture<RspList<T>> execute(MethodCall action, RequestOptions options) throws Exception {
  return this.<T>internalExecute(action, options);
}