@Test public void processMessage() throws Exception { WriteOrder message = new WriteOrder() .setArchiveId("tarId") .setBucket("bucket") .setFilePath("file_bucket") .setSize(1000L) .setDigest("digest"); CountDownLatch countDownLatch = new CountDownLatch(1); doAnswer((args) -> { countDownLatch.countDown(); return null; }).when(readWriteQueue).addIfAbsent(any(), any()); writeOrderCreator.startListener(); writeOrderCreator.addToQueue(message); assertThat(countDownLatch.await(1, TimeUnit.MINUTES)).isTrue(); verify(archiveReferentialRepository).updateLocationToReadyOnDisk("tarId", 1000L, "digest"); verifyNoMoreInteractions(archiveReferentialRepository); verify(readWriteQueue).addIfAbsent(anyList(), eq(message)); verifyNoMoreInteractions(readWriteQueue); }
@Override protected void processMessage(WriteOrder message) throws QueueProcessingException { try { sendMessageToQueue(message); } catch (Exception ex) { throw new QueueProcessingException(QueueProcessingException.RetryPolicy.RETRY, "Could not process message " + JsonHandler.unprettyPrint(message), ex); } }
WriteOrderCreator extends QueueProcessor<WriteOrder> { @Override protected void processMessage(WriteOrder message) throws QueueProcessingException { try { sendMessageToQueue(message); } catch (Exception ex) { throw new QueueProcessingException(QueueProcessingException.RetryPolicy.RETRY, "Could not process message " + JsonHandler.unprettyPrint(message), ex); } } }
WriteOrderCreator extends QueueProcessor<WriteOrder> { @Override protected void processMessage(WriteOrder message) throws QueueProcessingException { try { sendMessageToQueue(message); } catch (Exception ex) { throw new QueueProcessingException(QueueProcessingException.RetryPolicy.RETRY, "Could not process message " + JsonHandler.unprettyPrint(message), ex); } } WriteOrderCreator(ArchiveReferentialRepository archiveReferentialRepository,
        QueueRepository readWriteQueue); }
WriteOrderCreator extends QueueProcessor<WriteOrder> { @Override protected void processMessage(WriteOrder message) throws QueueProcessingException { try { sendMessageToQueue(message); } catch (Exception ex) { throw new QueueProcessingException(QueueProcessingException.RetryPolicy.RETRY, "Could not process message " + JsonHandler.unprettyPrint(message), ex); } } WriteOrderCreator(ArchiveReferentialRepository archiveReferentialRepository,
        QueueRepository readWriteQueue); void sendMessageToQueue(WriteOrder message); }
WriteOrderCreator extends QueueProcessor<WriteOrder> { @Override protected void processMessage(WriteOrder message) throws QueueProcessingException { try { sendMessageToQueue(message); } catch (Exception ex) { throw new QueueProcessingException(QueueProcessingException.RetryPolicy.RETRY, "Could not process message " + JsonHandler.unprettyPrint(message), ex); } } WriteOrderCreator(ArchiveReferentialRepository archiveReferentialRepository,
        QueueRepository readWriteQueue); void sendMessageToQueue(WriteOrder message);  }