@Test public void testTimeout() { try { UserAgentImpl adam = MockAgentFactory.getAdam(); adam.unlock("adamspass"); UserAgentImpl eve = MockAgentFactory.getEve(); LocalNodeManager manager = new LocalNodeManager(); LocalNode testee1 = manager.launchAgent(adam); MessageResultListener l = new MessageResultListener(2000) { @Override public void notifyTimeout() { LocalNodeTest.testTimeoutVariable = true; } }; Message m = new Message(adam, eve, new PingPongContent(), 1000); manager.setPendingTimeOut(1000); testee1.sendMessage(m, l); Thread.sleep(30000); assertFalse(l.isSuccess()); assertTrue(l.isTimedOut()); assertEquals(0, l.getResults().length); assertTrue(testTimeoutVariable); } catch (Exception e) { e.printStackTrace(); Assert.fail(e.toString()); } }
@Override public void sendMessage(Message message, MessageResultListener listener, SendMode mode) { message.setSendingNodeId(this.getNodeId()); registerAnswerListener(message.getId(), listener); try { switch (mode) { case ANYCAST: if (!message.isTopic()) { localNodeManager.localSendMessage(localNodeManager.findFirstNodeWithAgent(message.getRecipientId()), message); } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { localNodeManager.localSendMessage(ids[0], message); } } break; case BROADCAST: if (!message.isTopic()) { Long[] ids = localNodeManager.findAllNodesWithAgent(message.getRecipientId()); if (ids.length == 0) { listener.collectException(new AgentNotRegisteredException(message.getRecipientId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } } } catch (AgentNotRegisteredException e) { localNodeManager.storeMessage(message, listener); } }
LocalNode extends Node { @Override public void sendMessage(Message message, MessageResultListener listener, SendMode mode) { message.setSendingNodeId(this.getNodeId()); registerAnswerListener(message.getId(), listener); try { switch (mode) { case ANYCAST: if (!message.isTopic()) { localNodeManager.localSendMessage(localNodeManager.findFirstNodeWithAgent(message.getRecipientId()), message); } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { localNodeManager.localSendMessage(ids[0], message); } } break; case BROADCAST: if (!message.isTopic()) { Long[] ids = localNodeManager.findAllNodesWithAgent(message.getRecipientId()); if (ids.length == 0) { listener.collectException(new AgentNotRegisteredException(message.getRecipientId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } } } catch (AgentNotRegisteredException e) { localNodeManager.storeMessage(message, listener); } } }
LocalNode extends Node { @Override public void sendMessage(Message message, MessageResultListener listener, SendMode mode) { message.setSendingNodeId(this.getNodeId()); registerAnswerListener(message.getId(), listener); try { switch (mode) { case ANYCAST: if (!message.isTopic()) { localNodeManager.localSendMessage(localNodeManager.findFirstNodeWithAgent(message.getRecipientId()), message); } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { localNodeManager.localSendMessage(ids[0], message); } } break; case BROADCAST: if (!message.isTopic()) { Long[] ids = localNodeManager.findAllNodesWithAgent(message.getRecipientId()); if (ids.length == 0) { listener.collectException(new AgentNotRegisteredException(message.getRecipientId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } } } catch (AgentNotRegisteredException e) { localNodeManager.storeMessage(message, listener); } } LocalNode(LocalNodeManager localNodeManager);  LocalNode(LocalNodeManager localNodeManager, ClassManager classManager); }
LocalNode extends Node { @Override public void sendMessage(Message message, MessageResultListener listener, SendMode mode) { message.setSendingNodeId(this.getNodeId()); registerAnswerListener(message.getId(), listener); try { switch (mode) { case ANYCAST: if (!message.isTopic()) { localNodeManager.localSendMessage(localNodeManager.findFirstNodeWithAgent(message.getRecipientId()), message); } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { localNodeManager.localSendMessage(ids[0], message); } } break; case BROADCAST: if (!message.isTopic()) { Long[] ids = localNodeManager.findAllNodesWithAgent(message.getRecipientId()); if (ids.length == 0) { listener.collectException(new AgentNotRegisteredException(message.getRecipientId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } } } catch (AgentNotRegisteredException e) { localNodeManager.storeMessage(message, listener); } } LocalNode(LocalNodeManager localNodeManager);  LocalNode(LocalNodeManager localNodeManager, ClassManager classManager); @Override Long getNodeId(); @Override void shutDown(); @Override void registerReceiver(MessageReceiver receiver); @Override void sendMessage(Message message, MessageResultListener listener, SendMode mode); @Override void sendMessage(Message message, Object atNodeId, MessageResultListener listener); @Deprecated @Override EnvelopeVersion fetchArtifact(long id); @Deprecated @Override void storeArtifact(EnvelopeVersion envelope); @Deprecated @Override void removeArtifact(long id, byte[] signature); @Override Object[] findRegisteredAgent(String agentId, int hintOfExpectedCount); @Override AgentImpl getAgent(String id); void storeAgent(Agent agent); @Override void storeAgent(AgentImpl agent); @Deprecated @Override void updateAgent(AgentImpl agent); @Override Object[] getOtherKnownNodes(); @Override NodeInformation getNodeInformation(Object nodeId); @Override EnvelopeVersion createEnvelope(String identifier, PublicKey authorPubKey, Serializable content,
			AgentImpl... reader); @Override EnvelopeVersion createEnvelope(EnvelopeVersion previousVersion, Serializable content); @Override EnvelopeVersion createEnvelope(EnvelopeVersion previousVersion, Serializable content, AgentImpl... reader); @Override void storeEnvelope(EnvelopeVersion envelope, AgentImpl author); @Override EnvelopeVersion fetchEnvelope(String identifier); @Override EnvelopeVersion createEnvelope(String identifier, PublicKey authorPubKey, Serializable content,
			Collection<?> readers); @Override EnvelopeVersion createEnvelope(EnvelopeVersion previousVersion, Serializable content, Collection<?> readers); @Override EnvelopeVersion createUnencryptedEnvelope(String identifier, PublicKey authorPubKey, Serializable content); @Override EnvelopeVersion createUnencryptedEnvelope(EnvelopeVersion previousVersion, Serializable content); @Override void storeEnvelope(EnvelopeVersion envelope, AgentImpl author, long timeoutMs); @Override void storeEnvelopeAsync(EnvelopeVersion envelope, AgentImpl author, StorageStoreResultHandler resultHandler,
			StorageCollisionHandler collisionHandler, StorageExceptionHandler exceptionHandler); @Override EnvelopeVersion fetchEnvelope(String identifier, long timeoutMs); @Override void fetchEnvelopeAsync(String identifier, StorageEnvelopeHandler envelopeHandler,
			StorageExceptionHandler exceptionHandler); @Override void removeEnvelope(String identifier); @Override EnvelopeVersion fetchArtifact(String identifier); }
LocalNode extends Node { @Override public void sendMessage(Message message, MessageResultListener listener, SendMode mode) { message.setSendingNodeId(this.getNodeId()); registerAnswerListener(message.getId(), listener); try { switch (mode) { case ANYCAST: if (!message.isTopic()) { localNodeManager.localSendMessage(localNodeManager.findFirstNodeWithAgent(message.getRecipientId()), message); } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { localNodeManager.localSendMessage(ids[0], message); } } break; case BROADCAST: if (!message.isTopic()) { Long[] ids = localNodeManager.findAllNodesWithAgent(message.getRecipientId()); if (ids.length == 0) { listener.collectException(new AgentNotRegisteredException(message.getRecipientId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } else { Long[] ids = localNodeManager.findAllNodesWithTopic(message.getTopicId()); if (ids.length == 0) { listener.collectException( new MessageException("No agent listening to topic " + message.getTopicId())); } else { listener.addRecipients(ids.length - 1); for (long id : ids) { localNodeManager.localSendMessage(id, message); } } } } } catch (AgentNotRegisteredException e) { localNodeManager.storeMessage(message, listener); } } LocalNode(LocalNodeManager localNodeManager);  LocalNode(LocalNodeManager localNodeManager, ClassManager classManager); @Override Long getNodeId(); @Override void shutDown(); @Override void registerReceiver(MessageReceiver receiver); @Override void sendMessage(Message message, MessageResultListener listener, SendMode mode); @Override void sendMessage(Message message, Object atNodeId, MessageResultListener listener); @Deprecated @Override EnvelopeVersion fetchArtifact(long id); @Deprecated @Override void storeArtifact(EnvelopeVersion envelope); @Deprecated @Override void removeArtifact(long id, byte[] signature); @Override Object[] findRegisteredAgent(String agentId, int hintOfExpectedCount); @Override AgentImpl getAgent(String id); void storeAgent(Agent agent); @Override void storeAgent(AgentImpl agent); @Deprecated @Override void updateAgent(AgentImpl agent); @Override Object[] getOtherKnownNodes(); @Override NodeInformation getNodeInformation(Object nodeId); @Override EnvelopeVersion createEnvelope(String identifier, PublicKey authorPubKey, Serializable content,
			AgentImpl... reader); @Override EnvelopeVersion createEnvelope(EnvelopeVersion previousVersion, Serializable content); @Override EnvelopeVersion createEnvelope(EnvelopeVersion previousVersion, Serializable content, AgentImpl... reader); @Override void storeEnvelope(EnvelopeVersion envelope, AgentImpl author); @Override EnvelopeVersion fetchEnvelope(String identifier); @Override EnvelopeVersion createEnvelope(String identifier, PublicKey authorPubKey, Serializable content,
			Collection<?> readers); @Override EnvelopeVersion createEnvelope(EnvelopeVersion previousVersion, Serializable content, Collection<?> readers); @Override EnvelopeVersion createUnencryptedEnvelope(String identifier, PublicKey authorPubKey, Serializable content); @Override EnvelopeVersion createUnencryptedEnvelope(EnvelopeVersion previousVersion, Serializable content); @Override void storeEnvelope(EnvelopeVersion envelope, AgentImpl author, long timeoutMs); @Override void storeEnvelopeAsync(EnvelopeVersion envelope, AgentImpl author, StorageStoreResultHandler resultHandler,
			StorageCollisionHandler collisionHandler, StorageExceptionHandler exceptionHandler); @Override EnvelopeVersion fetchEnvelope(String identifier, long timeoutMs); @Override void fetchEnvelopeAsync(String identifier, StorageEnvelopeHandler envelopeHandler,
			StorageExceptionHandler exceptionHandler); @Override void removeEnvelope(String identifier); @Override EnvelopeVersion fetchArtifact(String identifier);  }