@Test public void directMessageProcessStepTestOnFailure() throws NoPeerConnectionException, InvalidProcessStateException, IOException { NetworkManager nodeA = network.get(0); final NetworkManager nodeB = network.get(1); try { String data = randomString(); String contentKey = randomString(); Parameters parameters = new Parameters().setLocationKey(nodeB.getNodeId()).setContentKey(contentKey); assertNull(nodeA.getDataManager().get(parameters)); nodeB.getConnection().getPeer().peer().rawDataReply(new DenyingMessageReplyHandler()); final TestDirectMessage message = new TestDirectMessage(nodeB.getNodeId(), nodeB.getConnection().getPeer() .peerAddress(), contentKey, new H2HTestData(data), false); BaseMessageProcessStep step = new BaseMessageProcessStep(nodeA.getMessageManager()) { @Override public void handleResponse(ResponseMessage responseMessage) { Assert.fail("Should be not used."); } @Override protected Void doExecute() throws InvalidProcessStateException, ProcessExecutionException { try { send(message, getPublicKey(nodeB)); Assert.fail(); } catch (SendFailedException e) { throw new ProcessExecutionException(this, "Expected behavior."); } return null; } }; TestExecutionUtil.executeProcessTillFailed(step); assertNull(nodeB.getDataManager().get(parameters)); } finally { nodeB.getConnection().getPeer().peer().rawDataReply(new MessageReplyHandler(nodeB, serializer)); } }
protected void send(BaseMessage message, PublicKey receiverPublicKey) throws SendFailedException { if (message instanceof IRequestMessage) { IRequestMessage requestMessage = (IRequestMessage) message; requestMessage.setCallBackHandler(this); responseLatch = new CountDownLatch(1); } boolean success; if (message instanceof BaseDirectMessage) { success = messageManager.sendDirect((BaseDirectMessage) message, receiverPublicKey); } else { success = messageManager.send(message, receiverPublicKey); } if (!success) { throw new SendFailedException("No success sending the message."); } else if (responseLatch != null) { try { if (!responseLatch.await(H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS, TimeUnit.MILLISECONDS)) { throw new SendFailedException("Response did not arrive within the configured wait time of " + H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS + "ms"); } } catch (InterruptedException e) { throw new SendFailedException("Cannot wait for the response because interrupted"); } } }
BaseMessageProcessStep extends ProcessStep<Void> implements IResponseCallBackHandler { protected void send(BaseMessage message, PublicKey receiverPublicKey) throws SendFailedException { if (message instanceof IRequestMessage) { IRequestMessage requestMessage = (IRequestMessage) message; requestMessage.setCallBackHandler(this); responseLatch = new CountDownLatch(1); } boolean success; if (message instanceof BaseDirectMessage) { success = messageManager.sendDirect((BaseDirectMessage) message, receiverPublicKey); } else { success = messageManager.send(message, receiverPublicKey); } if (!success) { throw new SendFailedException("No success sending the message."); } else if (responseLatch != null) { try { if (!responseLatch.await(H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS, TimeUnit.MILLISECONDS)) { throw new SendFailedException("Response did not arrive within the configured wait time of " + H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS + "ms"); } } catch (InterruptedException e) { throw new SendFailedException("Cannot wait for the response because interrupted"); } } } }
BaseMessageProcessStep extends ProcessStep<Void> implements IResponseCallBackHandler { protected void send(BaseMessage message, PublicKey receiverPublicKey) throws SendFailedException { if (message instanceof IRequestMessage) { IRequestMessage requestMessage = (IRequestMessage) message; requestMessage.setCallBackHandler(this); responseLatch = new CountDownLatch(1); } boolean success; if (message instanceof BaseDirectMessage) { success = messageManager.sendDirect((BaseDirectMessage) message, receiverPublicKey); } else { success = messageManager.send(message, receiverPublicKey); } if (!success) { throw new SendFailedException("No success sending the message."); } else if (responseLatch != null) { try { if (!responseLatch.await(H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS, TimeUnit.MILLISECONDS)) { throw new SendFailedException("Response did not arrive within the configured wait time of " + H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS + "ms"); } } catch (InterruptedException e) { throw new SendFailedException("Cannot wait for the response because interrupted"); } } } BaseMessageProcessStep(IMessageManager messageManager); }
BaseMessageProcessStep extends ProcessStep<Void> implements IResponseCallBackHandler { protected void send(BaseMessage message, PublicKey receiverPublicKey) throws SendFailedException { if (message instanceof IRequestMessage) { IRequestMessage requestMessage = (IRequestMessage) message; requestMessage.setCallBackHandler(this); responseLatch = new CountDownLatch(1); } boolean success; if (message instanceof BaseDirectMessage) { success = messageManager.sendDirect((BaseDirectMessage) message, receiverPublicKey); } else { success = messageManager.send(message, receiverPublicKey); } if (!success) { throw new SendFailedException("No success sending the message."); } else if (responseLatch != null) { try { if (!responseLatch.await(H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS, TimeUnit.MILLISECONDS)) { throw new SendFailedException("Response did not arrive within the configured wait time of " + H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS + "ms"); } } catch (InterruptedException e) { throw new SendFailedException("Cannot wait for the response because interrupted"); } } } BaseMessageProcessStep(IMessageManager messageManager); final void handleResponseMessage(ResponseMessage responseMessage); abstract void handleResponse(ResponseMessage responseMessage); }
BaseMessageProcessStep extends ProcessStep<Void> implements IResponseCallBackHandler { protected void send(BaseMessage message, PublicKey receiverPublicKey) throws SendFailedException { if (message instanceof IRequestMessage) { IRequestMessage requestMessage = (IRequestMessage) message; requestMessage.setCallBackHandler(this); responseLatch = new CountDownLatch(1); } boolean success; if (message instanceof BaseDirectMessage) { success = messageManager.sendDirect((BaseDirectMessage) message, receiverPublicKey); } else { success = messageManager.send(message, receiverPublicKey); } if (!success) { throw new SendFailedException("No success sending the message."); } else if (responseLatch != null) { try { if (!responseLatch.await(H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS, TimeUnit.MILLISECONDS)) { throw new SendFailedException("Response did not arrive within the configured wait time of " + H2HConstants.DIRECT_DOWNLOAD_AWAIT_MS + "ms"); } } catch (InterruptedException e) { throw new SendFailedException("Cannot wait for the response because interrupted"); } } } BaseMessageProcessStep(IMessageManager messageManager); final void handleResponseMessage(ResponseMessage responseMessage); abstract void handleResponse(ResponseMessage responseMessage);  }