TransactionProcessor extends AbstractProcessor<ExecutionContext> { @Override public void handle(final ExecutionContext context) { final String requestId = context.request().id(); String transactionId = context.request().headers().getFirst(transactionHeader); if (transactionId == null) { transactionId = requestId; context.request().headers().set(transactionHeader, transactionId); } context.request().metrics().setTransactionId(transactionId); context.response().headers().set(transactionHeader, transactionId); context.request().headers().set(requestHeader, requestId); context.response().headers().set(requestHeader, requestId); ((MutableExecutionContext)context).request(new TransactionRequest(transactionId, context.request())); next.handle(context); } TransactionProcessor();   TransactionProcessor(String transactionHeader, String requestHeader); @Override void handle(final ExecutionContext context);  }
@Test public void shouldHaveTransactionId() throws InterruptedException { final CountDownLatch lock = new CountDownLatch(1); Mockito.when(request.id()).thenReturn(UUID.toString(UUID.random())); new TransactionProcessor() .handler(context -> { Assert.assertNotNull(context.request().transactionId()); Assert.assertEquals(context.request().transactionId(), context.request().headers().getFirst(TransactionProcessor.DEFAULT_TRANSACTION_ID_HEADER)); Assert.assertEquals(context.request().transactionId(), context.request().metrics().getTransactionId()); Assert.assertEquals(context.request().transactionId(), response.headers().getFirst(TransactionProcessor.DEFAULT_TRANSACTION_ID_HEADER)); Assert.assertEquals(context.request().id(), context.request().headers().getFirst(TransactionProcessor.DEFAULT_REQUEST_ID_HEADER)); Assert.assertEquals(context.request().id(), response.headers().getFirst(TransactionProcessor.DEFAULT_REQUEST_ID_HEADER)); lock.countDown(); }) .handle(context); Assert.assertTrue(lock.await(10000, TimeUnit.MILLISECONDS)); }
@Test public void shouldPropagateSameTransactionId() throws InterruptedException { final CountDownLatch lock = new CountDownLatch(1); String transactionId = UUID.toString(UUID.random()); request.headers().set(TransactionProcessor.DEFAULT_TRANSACTION_ID_HEADER, transactionId); new TransactionProcessor() .handler(context -> { Assert.assertNotNull(context.request().transactionId()); Assert.assertEquals(transactionId, context.request().transactionId()); Assert.assertEquals(transactionId, context.request().headers().getFirst(TransactionProcessor.DEFAULT_TRANSACTION_ID_HEADER)); Assert.assertEquals(transactionId, context.request().metrics().getTransactionId()); Assert.assertEquals(context.request().transactionId(), response.headers().getFirst(TransactionProcessor.DEFAULT_TRANSACTION_ID_HEADER)); Assert.assertNotEquals(transactionId, context.request().id()); Assert.assertNotEquals(transactionId, context.request().headers().getFirst(TransactionProcessor.DEFAULT_REQUEST_ID_HEADER)); Assert.assertNotEquals(transactionId, context.request().metrics().getRequestId()); Assert.assertEquals(context.request().id(), response.headers().getFirst(TransactionProcessor.DEFAULT_REQUEST_ID_HEADER)); lock.countDown(); }).handle(context); Assert.assertTrue(lock.await(10000, TimeUnit.MILLISECONDS)); }
@Test public void shouldHaveTransactionIdWithCustomHeader() throws InterruptedException { final CountDownLatch lock = new CountDownLatch(1); Mockito.when(request.id()).thenReturn(UUID.toString(UUID.random())); new TransactionProcessor(CUSTOM_TRANSACTION_ID_HEADER, CUSTOM_REQUEST_ID_HEADER) .handler(context -> { Assert.assertNotNull(context.request().transactionId()); Assert.assertEquals(context.request().transactionId(), context.request().headers().getFirst(CUSTOM_TRANSACTION_ID_HEADER)); Assert.assertEquals(context.request().transactionId(), context.request().metrics().getTransactionId()); Assert.assertEquals(context.request().transactionId(), response.headers().getFirst(CUSTOM_TRANSACTION_ID_HEADER)); lock.countDown(); }).handle(context); Assert.assertTrue(lock.await(10000, TimeUnit.MILLISECONDS)); }
@Test public void shouldPropagateSameTransactionIdWithCustomHeader() throws InterruptedException { final CountDownLatch lock = new CountDownLatch(1); String transactionId = UUID.toString(UUID.random()); request.headers().set(CUSTOM_TRANSACTION_ID_HEADER, transactionId); new TransactionProcessor(CUSTOM_TRANSACTION_ID_HEADER, CUSTOM_REQUEST_ID_HEADER) .handler(context -> { Assert.assertNotNull(context.request().transactionId()); Assert.assertEquals(transactionId, context.request().transactionId()); Assert.assertEquals(transactionId, context.request().headers().getFirst(CUSTOM_TRANSACTION_ID_HEADER)); Assert.assertEquals(transactionId, context.request().metrics().getTransactionId()); Assert.assertEquals(context.request().transactionId(), response.headers().getFirst(CUSTOM_TRANSACTION_ID_HEADER)); lock.countDown(); }).handle(context); Assert.assertTrue(lock.await(10000, TimeUnit.MILLISECONDS)); }