LocalNetwork implements Network { @Override public String getConnectionSpec() { return hostId; } LocalNetwork();  LocalNetwork(LocalWire wire); @Override boolean waitUntilReady(double seconds); @Override void attach(NetworkOwner owner); @Override void registerSession(String session); @Override void unregisterSession(String session); @Override boolean allocServiceAddress(RoutingNode recipient); @Override void freeServiceAddress(RoutingNode recipient); @Override void send(Message msg, List<RoutingNode> recipients); @Override void sync(); @Override void shutdown(); @Override String getConnectionSpec(); @Override IMirror getMirror();  }
@Test public void requireThatBlockingSendTimeOutInSendQ() throws InterruptedException { final LocalWire wire = new LocalWire(); final Server serverA = new Server(wire); final SourceSession source = serverA.newSourceSession(new StaticThrottlePolicy().setMaxPendingCount(1)); final Server serverB = new Server(wire); final IntermediateSession intermediate = serverB.newIntermediateSession(); final Server serverC = new Server(wire); final DestinationSession destination = serverC.newDestinationSession(); Message msg = new SimpleMessage("foo"); msg.setRoute(new Route().addHop(Hop.parse(intermediate.getConnectionSpec())) .addHop(Hop.parse(destination.getConnectionSpec()))); assertTrue(source.sendBlocking(msg).isAccepted()); long start = SystemTimer.INSTANCE.milliTime(); Message msg2 = new SimpleMessage("foo2"); msg2.setRoute(new Route().addHop(Hop.parse(intermediate.getConnectionSpec())) .addHop(Hop.parse(destination.getConnectionSpec()))); long TIMEOUT = 1000; msg2.setTimeRemaining(TIMEOUT); Result res = source.sendBlocking(msg2); assertFalse(res.isAccepted()); assertEquals(ErrorCode.TIMEOUT, res.getError().getCode()); assertTrue(res.getError().getMessage().endsWith("Timed out in sendQ")); long end = SystemTimer.INSTANCE.milliTime(); assertTrue(end >= start+TIMEOUT); assertTrue(end < start+5*TIMEOUT); msg = serverB.messages.poll(60, TimeUnit.SECONDS); assertTrue(msg instanceof SimpleMessage); assertEquals("foo", ((SimpleMessage)msg).getValue()); intermediate.forward(msg); msg = serverC.messages.poll(60, TimeUnit.SECONDS); assertTrue(msg instanceof SimpleMessage); assertEquals("foo", ((SimpleMessage)msg).getValue()); Reply reply = new SimpleReply("bar"); reply.swapState(msg); destination.reply(reply); reply = serverB.replies.poll(60, TimeUnit.SECONDS); assertTrue(reply instanceof SimpleReply); assertEquals("bar", ((SimpleReply)reply).getValue()); intermediate.forward(reply); reply = serverA.replies.poll(60, TimeUnit.SECONDS); assertEquals(ErrorCode.TIMEOUT, reply.getError(0).getCode()); assertTrue(reply.getError(0).getMessage().endsWith("Timed out in sendQ")); reply = serverA.replies.poll(60, TimeUnit.SECONDS); assertTrue(reply instanceof SimpleReply); assertEquals("bar", ((SimpleReply)reply).getValue()); serverA.mbus.destroy(); serverB.mbus.destroy(); serverC.mbus.destroy(); }