CoapRetryFilter extends AbstractIoFilter { @Override public void messageWriting(final IoSession session, final WriteRequest message, WriteFilterChainController controller) { LOGGER.debug("Processing a MESSAGE_WRITING for session {}", session); final CoapMessage coapMsg = (CoapMessage) message.getMessage(); final String transmissionId = CoapTransmission.uniqueId(session, coapMsg); switch (coapMsg.getType()) { case NON_CONFIRMABLE: controller.callWriteNextFilter(message); break; case RESET: case ACK: processed.put(transmissionId, coapMsg); controller.callWriteNextFilter(message); break; case CONFIRMABLE: CoapTransmission t = inFlight.get(transmissionId); if (t == null) { t = new CoapTransmission(session, coapMsg); inFlight.put(t.getId(), t); } ScheduledFuture<?> future = retryExecutor.schedule(new Runnable() { @Override public void run() { CoapTransmission t = inFlight.get(transmissionId); if (t != null && t.timeout()) { LOGGER.debug("Retry for message with ID {}", coapMsg.requestId()); session.write(coapMsg); } else { LOGGER.debug("No more retry for message with ID {}", coapMsg.requestId()); } } }, t.getNextTimeout(), TimeUnit.MILLISECONDS); t.setRetryFuture(future); controller.callWriteNextFilter(message); break; } }  @Override void messageReceived(IoSession session, Object in, ReadFilterChainController controller); @Override void messageWriting(final IoSession session, final WriteRequest message, WriteFilterChainController controller);  }
@Test public void retry_confirmable_message() throws InterruptedException { CoapMessage msg = new CoapMessage(1, MessageType.CONFIRMABLE, 1, 1234, null, null, null); WriteRequest writeRequest = new DefaultWriteRequest(msg); filter.messageWriting(session, writeRequest, writeController); Thread.sleep(3500L); verify(writeController).callWriteNextFilter(writeRequest); session.write(msg); }
@Test public void non_confirmable_message_writing() { CoapMessage msg = new CoapMessage(1, MessageType.NON_CONFIRMABLE, 1, 1234, "token".getBytes(), null, "payload".getBytes()); WriteRequest writeRequest = new DefaultWriteRequest(msg); filter.messageWriting(session, writeRequest, writeController); verify(writeController).callWriteNextFilter(writeRequest); Mockito.verifyNoMoreInteractions(writeController); }