diff --git a/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java new file mode 100644 index 0000000..9533bd1 --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java @@ -0,0 +1,124 @@ +/* + * Copyright 2016 esbtools Contributors and/or its affiliates. + * + * This file is part of esbtools. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.esbtools.eventhandler; + +import org.apache.camel.builder.RouteBuilder; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +public class AsyncMessageProcessorRoute extends RouteBuilder { + private final String fromUri; + private final String failureUri; + private final Duration processTimeout; + private final MessageFactory messageFactory; + + private final int idCount = idCounter.getAndIncrement(); + private final String routeId = "async-message-repository-" + idCount; + + private static final AtomicInteger idCounter = new AtomicInteger(0); + + public AsyncMessageProcessorRoute(String fromUri, String failureUri, Duration processTimeout, + MessageFactory messageFactory) { + this.fromUri = Objects.requireNonNull(fromUri, "fromUri"); + this.failureUri = Objects.requireNonNull(failureUri, "failureUri"); + this.processTimeout = Objects.requireNonNull(processTimeout, "processTimeout"); + this.messageFactory = Objects.requireNonNull(messageFactory, "messageFactory"); + } + + @Override + public void configure() throws Exception { + from(fromUri) + .routeId(routeId) + .process(exchange -> { + Object exchangeBody = exchange.getIn().getBody(); + + if (!(exchangeBody instanceof Collection)) { + throw new IllegalArgumentException("Expected `fromUri` to deliver exchanges with " + + "Collection bodies so that we may batch process for efficiency. However, " + + "the uri '" + fromUri + "' returned the " + + exchangeBody.getClass().getName() + ": " + exchangeBody); + } + + Collection messages = (Collection) exchangeBody; + + List processingMessages = new ArrayList<>(messages.size()); + List failures = new ArrayList<>(); + + // Start processing all of the messages in the batch in parallel. + for (Object message : messages) { + try { + Message parsedMessage = messageFactory.getMessageForBody(message); + Future processingFuture = parsedMessage.process(); + ProcessingMessage processing = new ProcessingMessage( + message, parsedMessage, processingFuture); + processingMessages.add(processing); + log.debug("Processing on route {}: {}", routeId, parsedMessage); + } catch (Exception e) { + log.error("Failure parsing message. Body was: " + message, e); + failures.add(new FailedMessage(message, e)); + } + } + + // Wait for processing to complete. + for (ProcessingMessage processingMsg : processingMessages) { + try { + processingMsg.future.get(processTimeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + log.error("Failed to process message: " + processingMsg.parsedMessage, e); + FailedMessage failure = new FailedMessage(processingMsg.originalMessage, + processingMsg.parsedMessage, e.getCause()); + failures.add(failure); + } catch (InterruptedException | TimeoutException e) { + log.warn("Timed out processing message: " + processingMsg.parsedMessage, e); + RecoverableException recoverableException = new RecoverableException(e); + FailedMessage failure = new FailedMessage(processingMsg.originalMessage, + processingMsg.parsedMessage, recoverableException); + failures.add(failure); + } + } + + // Deal with failures... + exchange.getIn().setBody(failures); + }) + .split(body()) + .to(failureUri); + } + + /** Simple struct for storing a message and its future processing result. */ + private static class ProcessingMessage { + final Object originalMessage; + final Message parsedMessage; + final Future future; + + ProcessingMessage(Object originalMessage, Message parsedMessage, Future future) { + this.originalMessage = originalMessage; + this.parsedMessage = parsedMessage; + this.future = future; + } + } +} diff --git a/lib/src/main/java/org/esbtools/eventhandler/EventHandlerException.java b/lib/src/main/java/org/esbtools/eventhandler/EventHandlerException.java index a194464..bd5da86 100644 --- a/lib/src/main/java/org/esbtools/eventhandler/EventHandlerException.java +++ b/lib/src/main/java/org/esbtools/eventhandler/EventHandlerException.java @@ -18,7 +18,7 @@ package org.esbtools.eventhandler; -public class EventHandlerException extends RuntimeException { +public class EventHandlerException extends Exception { public EventHandlerException(String message) { super(message); } diff --git a/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java b/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java new file mode 100644 index 0000000..27bb3f0 --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java @@ -0,0 +1,80 @@ +/* + * Copyright 2016 esbtools Contributors and/or its affiliates. + * + * This file is part of esbtools. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.esbtools.eventhandler; + +import java.util.Objects; +import java.util.Optional; + +public final class FailedMessage { + private final Object originalMessage; + private final Optional parsedMessage; + private final Throwable exception; + + public FailedMessage(Object originalMessage, Message parsedMessage, Throwable exception) { + this.originalMessage = originalMessage; + this.parsedMessage = Optional.of(parsedMessage); + this.exception = exception; + } + + public FailedMessage(Object originalMessage, Throwable exception) { + this.originalMessage = originalMessage; + this.parsedMessage = Optional.empty(); + this.exception = exception; + } + + public Object originalMessage() { + return originalMessage; + } + + public Optional parsedMessage() { + return parsedMessage; + } + + public Throwable exception() { + return exception; + } + + public boolean isRecoverable() { + return exception instanceof RecoverableException; + } + + @Override + public String toString() { + return "FailedMessage{" + + "exception=" + exception + + ", parsedMessage=" + parsedMessage + + ", originalMessage=" + originalMessage + + '}'; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + FailedMessage that = (FailedMessage) o; + return Objects.equals(originalMessage, that.originalMessage) && + Objects.equals(parsedMessage, that.parsedMessage) && + Objects.equals(exception, that.exception); + } + + @Override + public int hashCode() { + return Objects.hash(originalMessage, parsedMessage, exception); + } +} diff --git a/lib/src/main/java/org/esbtools/eventhandler/Message.java b/lib/src/main/java/org/esbtools/eventhandler/Message.java new file mode 100644 index 0000000..565e0c1 --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/Message.java @@ -0,0 +1,25 @@ +/* + * Copyright 2016 esbtools Contributors and/or its affiliates. + * + * This file is part of esbtools. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.esbtools.eventhandler; + +import java.util.concurrent.Future; + +public interface Message { + Future process(); +} diff --git a/lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java b/lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java new file mode 100644 index 0000000..479531a --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java @@ -0,0 +1,23 @@ +/* + * Copyright 2016 esbtools Contributors and/or its affiliates. + * + * This file is part of esbtools. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.esbtools.eventhandler; + +public interface MessageFactory { + Message getMessageForBody(Object body); +} diff --git a/lib/src/main/java/org/esbtools/eventhandler/RecoverableException.java b/lib/src/main/java/org/esbtools/eventhandler/RecoverableException.java new file mode 100644 index 0000000..5af8448 --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/RecoverableException.java @@ -0,0 +1,29 @@ +/* + * Copyright 2016 esbtools Contributors and/or its affiliates. + * + * This file is part of esbtools. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.esbtools.eventhandler; + +public class RecoverableException extends EventHandlerException { + public RecoverableException(String message, Throwable cause) { + super(message, cause); + } + + public RecoverableException(Throwable cause) { + super(cause); + } +} diff --git a/lib/src/test/java/org/esbtools/eventhandler/AsyncMessageProcessorRouteTest.java b/lib/src/test/java/org/esbtools/eventhandler/AsyncMessageProcessorRouteTest.java new file mode 100644 index 0000000..36f079f --- /dev/null +++ b/lib/src/test/java/org/esbtools/eventhandler/AsyncMessageProcessorRouteTest.java @@ -0,0 +1,316 @@ +/* + * Copyright 2016 esbtools Contributors and/or its affiliates. + * + * This file is part of esbtools. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +package org.esbtools.eventhandler; + +import com.google.common.truth.Truth; +import com.google.common.util.concurrent.Futures; +import com.jayway.awaitility.Awaitility; +import org.apache.camel.EndpointInject; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit4.CamelTestSupport; +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +@RunWith(JUnit4.class) +public class AsyncMessageProcessorRouteTest extends CamelTestSupport { + List persistence = new ArrayList<>(); + + MessageFactory messageFactory = new ByTypeMessageFactory(persistence); + + @EndpointInject(uri = "direct:incoming") + ProducerTemplate toIncoming; + + @EndpointInject(uri = "direct:short_timeout") + ProducerTemplate toShortTimeout; + + @EndpointInject(uri = "mock:failures") + MockEndpoint toFailures; + + /** + * Demonstrates the kind of message parsing behavior a real factory might do by creating one of + * three types of messages based on the provided object: + * + *
    + *
  • {@link FailingMessage} when the object is an Exception
    • + *
  • {@link TimeConsumingMessage} when the object is a {@link Duration}
    • + *
  • In all other cases, creates a {@link TracingPersistingMessage} which persists when + * message processing starts and ends with the provided object as the final result.
    • + *
+ * + *

Additionally, if the object is a {@link SimulatedMessageFactoryFailure} then this, as you + * might guess, causes the message factory to throw an exception instead of returning a message. + */ + static class ByTypeMessageFactory implements MessageFactory { + private final List persistence; + + ByTypeMessageFactory(List persistence) { + this.persistence = persistence; + } + + @Override + public Message getMessageForBody(Object body) { + if (body instanceof Exception) { + return new FailingMessage((Exception) body); + } + + if (body instanceof Duration) { + return new TimeConsumingMessage((Duration) body); + } + + if (body instanceof SimulatedMessageFactoryFailure) { + throw ((SimulatedMessageFactoryFailure) body).exception; + } + + return new TracingPersistingMessage(body, persistence); + } + } + + /** + * Creates two message processor routes: one which reads from "direct:incoming" with a long + * timeout which should not be hit in normal tests, and another with a very short timeout, + * reading from "direct:short_timeout", in order to test timeout handling. + */ + @Override + protected RouteBuilder[] createRouteBuilders() throws Exception { + return new RouteBuilder[]{ + new AsyncMessageProcessorRoute("direct:incoming", "mock:failures", + Duration.ofMinutes(1), messageFactory), + new AsyncMessageProcessorRoute("direct:short_timeout", "mock:failures", + Duration.ofMillis(1), messageFactory) + }; + } + + @Test(timeout = 1000L) + public void shouldProcessReceivedMessagesInCollectionInParallel() { + List messages = Arrays.asList("fun", "with", "messages"); + + toIncoming.sendBody(messages); + + Awaitility.await().until( + () -> persistence, + Matchers.hasItems("fun", "with", "messages")); + + Truth.assertThat(persistence.subList(0, 3)) + .named("first three persisted results") + .containsExactly("processing fun", "processing with", "processing messages"); + } + + @Test(expected = Exception.class) + public void shouldFailIfEndpointDoesNotReceiveACollectionType() { + toIncoming.sendBody("not a collection"); + } + + @Test + public void shouldWrapFailuresAndSendToFailureUri() { + toFailures.expectedMessageCount(1); + + Exception exception = new Exception("Simulated failure"); + List messages = Collections.singletonList(exception); + + // Our message factory treats incoming exceptions as failing messages. + toIncoming.sendBody(messages); + + toFailures.expectedMessageCount(1); + + Object failure = toFailures.getExchanges().get(0).getIn().getBody(); + + Truth.assertThat(failure).named("message sent to failure endpoint") + .isInstanceOf(FailedMessage.class); + + FailedMessage failedMessage = (FailedMessage) failure; + + Truth.assertThat(failedMessage.parsedMessage().get()) + .isEqualTo(new FailingMessage(exception)); + Truth.assertThat(failedMessage.originalMessage()).isEqualTo(exception); + Truth.assertThat(failedMessage.exception()).isEqualTo(exception); + } + + @Test + public void shouldCatchMessageFactoryFailures() throws InterruptedException { + toFailures.expectedMessageCount(1); + + RuntimeException exception = new RuntimeException("Simulated runtime exception while parsing message"); + SimulatedMessageFactoryFailure originalMsg = new SimulatedMessageFactoryFailure(exception); + + toIncoming.sendBody(Collections.singleton(originalMsg)); + + toFailures.assertIsSatisfied(); + + Object failure = toFailures.getExchanges().get(0).getIn().getBody(); + + Truth.assertThat(failure).named("message sent to failure endpoint") + .isInstanceOf(FailedMessage.class); + + FailedMessage failedMessage = (FailedMessage) failure; + + Truth.assertThat(failedMessage.parsedMessage().isPresent()) + .named("presence of a parsed message").isFalse(); + Truth.assertThat(failedMessage.originalMessage()).isEqualTo(originalMsg); + Truth.assertThat(failedMessage.exception()).isEqualTo(exception); + } + + @Test(timeout = 1000L) + public void shouldKeepProcessingRemainingMessagesDespiteFailures() { + List messages = Arrays.asList( + new SimulatedMessageFactoryFailure(new RuntimeException("Simulated parse failure")), + new Exception("Simulated processing failure"), + "success!"); + + toIncoming.sendBody(messages); + + Awaitility.await().until( + () -> persistence, + Matchers.hasItem("success!")); + } + + @Test(timeout = 1000L) + public void shouldWrapTimeoutsInRecoverableExceptionsAndSendToFailuresUri() throws InterruptedException { + toFailures.expectedMessageCount(1); + + toShortTimeout.sendBody(Arrays.asList(Duration.ofSeconds(5))); + + toFailures.assertIsSatisfied(); + + Object failure = toFailures.getExchanges().get(0).getIn().getBody(); + + Truth.assertThat(failure).named("message sent to failure endpoint") + .isInstanceOf(FailedMessage.class); + + FailedMessage failedMessage = (FailedMessage) failure; + + Truth.assertThat(failedMessage.parsedMessage().get()).isInstanceOf(TimeConsumingMessage.class); + Truth.assertThat(failedMessage.originalMessage()).isEqualTo(Duration.ofSeconds(5)); + Truth.assertThat(failedMessage.exception()).isInstanceOf(RecoverableException.class); + } + + static class FailingMessage implements Message { + private final Exception exception; + + FailingMessage(Exception exception) { + this.exception = exception; + } + + @Override + public Future process() { + return Futures.immediateFailedFuture(exception); + } + + @Override + public String toString() { + return "FailingMessage{" + + "exception=" + exception + + '}'; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + FailingMessage that = (FailingMessage) o; + return Objects.equals(exception, that.exception); + } + + @Override + public int hashCode() { + return Objects.hash(exception); + } + } + + static class TimeConsumingMessage implements Message { + private final Duration duration; + + TimeConsumingMessage(Duration duration) { + this.duration = duration; + } + + @Override + public Future process() { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + return executor.submit(() -> { + try { + Thread.sleep(duration.toMillis()); + } catch (InterruptedException e) { + e.printStackTrace(); + } + }); + } finally { + executor.shutdown(); + } + } + + @Override + public String toString() { + return "TimeConsumingMessage{" + + "duration=" + duration + + '}'; + } + } + + /** + * Tracks when processing starts and when processing is done. + */ + static class TracingPersistingMessage implements Message { + private final Object body; + private final List persistence; + + TracingPersistingMessage(Object body, List persistence) { + this.body = body; + this.persistence = persistence; + } + + @Override + public Future process() { + persistence.add("processing " + body); + + return Futures.lazyTransform( + Futures.immediateFuture(body), + (e) -> persistence.add(e)); + } + + @Override + public String toString() { + return "LazyPersistingMessage{" + + "body=" + body + + '}'; + } + } + + static class SimulatedMessageFactoryFailure { + private final RuntimeException exception; + + public SimulatedMessageFactoryFailure(RuntimeException exception) { + this.exception = exception; + } + } +} diff --git a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentEventRepository.java b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentEventRepository.java index b857b78..fac6b08 100644 --- a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentEventRepository.java +++ b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentEventRepository.java @@ -341,7 +341,7 @@ private static DocumentEventEntity asEntity(DocumentEvent event) { return ((LightblueDocumentEvent) event).wrappedDocumentEventEntity(); } - throw new EventHandlerException("Unknown event type. Only LightblueDocumentEvent is " + + throw new IllegalArgumentException("Unknown event type. Only LightblueDocumentEvent is " + "supported. Event type was: " + event.getClass()); } diff --git a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueNotificationRepository.java b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueNotificationRepository.java index 3414145..0736c6f 100644 --- a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueNotificationRepository.java +++ b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueNotificationRepository.java @@ -266,8 +266,8 @@ private static NotificationEntity asEntity(Notification notification) { return ((LightblueNotification) notification).wrappedNotificationEntity(); } - throw new EventHandlerException("Unknown notification type. Only LightblueNotification " + - "are supported. Event type was: " + notification.getClass()); + throw new IllegalArgumentException("Unknown notification type. Only " + + "LightblueNotification is supported. Event type was: " + notification.getClass()); } static class ProcessingNotification implements Lockable {