int capacity = 0;
ArrayList<ZFrame> workers = new ArrayList<ZFrame>();
while (true) {
// First, route any waiting replies from workers
PollItem backends[] = {
new PollItem(localbe, Poller.POLLIN),
new PollItem(cloudbe, Poller.POLLIN)
};
// If we have no workers anyhow, wait indefinitely
int rc = ZMQ.poll(backends,
capacity > 0 ? 1000 : -1);
if (rc == -1)
break; // Interrupted
// Handle reply from local worker
ZMsg msg = null;
if (backends[0].isReadable()) {
msg = ZMsg.recvMsg(localbe);
if (msg == null)
break; // Interrupted
ZFrame address = msg.unwrap();
workers.add(address);
capacity++;
// If it's READY, don't route the message any further
ZFrame frame = msg.getFirst();
if (new String(frame.getData()).equals(WORKER_READY)) {
msg.destroy();
msg = null;
}
}
// Or handle reply from peer broker
else if (backends[1].isReadable()) {
msg = ZMsg.recvMsg(cloudbe);
if (msg == null)
break; // Interrupted
// We don't use peer broker address for anything
ZFrame address = msg.unwrap();
address.destroy();
}
// Route reply to cloud if it's addressed to a broker
for (argn = 1; msg != null && argn < argv.length; argn++) {
byte[] data = msg.getFirst().getData();
if (argv[argn].equals(new String(data))) {
msg.send(cloudfe);
msg = null;
}
}
// Route reply to client if we still need to
if (msg != null)
msg.send(localfe);
// Now we route as many client requests as we have worker capacity
// for. We may reroute requests from our local frontend, but not from //
// the cloud frontend. We reroute randomly now, just to test things
// out. In the next version we'll do this properly by calculating
// cloud capacity://
while (capacity > 0) {
PollItem frontends[] = {
new PollItem(localfe, Poller.POLLIN),
new PollItem(cloudfe, Poller.POLLIN)
};
rc = ZMQ.poll(frontends, 0);
assert (rc >= 0);
int reroutable = 0;
// We'll do peer brokers first, to prevent starvation