From fc74901b18bb2849f2ed701e641f1447809d02b4 Mon Sep 17 00:00:00 2001 From: Alec Henninger Date: Thu, 11 Feb 2016 15:30:15 -0500 Subject: [PATCH 1/4] Introduce DocumentRepository for processing Document messages --- .../AsyncDocumentProcessorRoute.java | 39 ++++++++++++++++++ .../eventhandler/DocumentRepository.java | 25 +++++++++++ .../eventhandler/lightblue/Document.java | 23 +++++++++++ .../LightblueDocumentRepository.java | 41 +++++++++++++++++++ 4 files changed, 128 insertions(+) create mode 100644 lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java create mode 100644 lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.java create mode 100644 lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java create mode 100644 lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java diff --git a/lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java b/lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java new file mode 100644 index 0000000..219c16c --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java @@ -0,0 +1,39 @@ +/* + * 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; + +public class AsyncDocumentProcessorRoute extends RouteBuilder { + private final DocumentRepository documentRepository; + private final String fromUri; + + public AsyncDocumentProcessorRoute(DocumentRepository documentRepository, String fromUri) { + this.documentRepository = documentRepository; + this.fromUri = fromUri; + } + + @Override + public void configure() throws Exception { + from(fromUri) + .process(exchange -> { + documentRepository.insertOrUpdate(exchange.getIn().getBody()); + }); + } +} diff --git a/lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.java b/lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.java new file mode 100644 index 0000000..84cb65e --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.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.Collection; + +public interface DocumentRepository { + void insertOrUpdate(Collection document) throws Exception; +} diff --git a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java new file mode 100644 index 0000000..1e4caa2 --- /dev/null +++ b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.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.lightblue; + +public interface Document { + +} diff --git a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java new file mode 100644 index 0000000..56e344a --- /dev/null +++ b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java @@ -0,0 +1,41 @@ +/* + * 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.lightblue; + +import org.esbtools.eventhandler.DocumentRepository; +import org.esbtools.eventhandler.lightblue.locking.LockStrategy; +import org.esbtools.eventhandler.lightblue.locking.LockedResource; +import org.esbtools.eventhandler.lightblue.locking.LockedResources; + +import java.util.Collection; + +public class LightblueDocumentRepository implements DocumentRepository { + private final LockStrategy lockStrategy; + + public LightblueDocumentRepository(LockStrategy lockStrategy) { + this.lockStrategy = lockStrategy; + } + + @Override + public void insertOrUpdate(Collection documents) throws Exception { + try (LockedResources lockedDocuments = parseAndWaitForLocks(documents)) { + + } + } +} From e13c7231a0140853dfd02c3d663edeca90cfa667 Mon Sep 17 00:00:00 2001 From: Alec Henninger Date: Mon, 15 Feb 2016 13:10:19 -0500 Subject: [PATCH 2/4] Generalize Document message handling They are just "Messages" and messages can be "processed." It follows that it is an appropriate abstraction for messages which, in processing, do not need to know about any other message, and do not produce a result other an indication of "success" or "failure" where failure is indicated by an Exception, and success is indicated by the absence of an Exception. --- .../AsyncMessageProcessorRoute.java | 113 ++++++++++++++++++ .../eventhandler/EventHandlerException.java | 2 +- .../esbtools/eventhandler/FailedMessage.java | 65 ++++++++++ .../{DocumentRepository.java => Message.java} | 6 +- .../esbtools/eventhandler/MessageFactory.java | 6 +- ...orRoute.java => RecoverableException.java} | 20 +--- .../LightblueDocumentRepository.java | 41 ------- 7 files changed, 190 insertions(+), 63 deletions(-) create mode 100644 lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java create mode 100644 lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java rename lib/src/main/java/org/esbtools/eventhandler/{DocumentRepository.java => Message.java} (85%) rename lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java => lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java (87%) rename lib/src/main/java/org/esbtools/eventhandler/{AsyncDocumentProcessorRoute.java => RecoverableException.java} (57%) delete mode 100644 lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java 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..5663c01 --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java @@ -0,0 +1,113 @@ +/* + * 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.get(); + 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: " + 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(); + processingMessages.add(new MessageAndProcessingFuture(parsedMessage, processingFuture)); + } catch (Exception e) { + log.error("Failure parsing message. Body was: " + message, e); + // TODO: How to treat this failure? + // We don't have a Message but FailedMessage requires one. + // Type-unsafe Failure type? + } + } + + // Wait for processing to complete. + for (MessageAndProcessingFuture processingMsg : processingMessages) { + try { + processingMsg.future.get(processTimeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (InterruptedException | TimeoutException e) { + RecoverableException recoverableException = new RecoverableException(e); + failures.add(new FailedMessage(processingMsg.message, recoverableException)); + } catch (ExecutionException e) { + failures.add(new FailedMessage(processingMsg.message, e.getCause())); + } + } + + // Deal with failures... + exchange.getIn().setBody(failures); + }) + .to(failureUri); + } + + /** Simple struct for storing a message and its future processing result. */ + private static class MessageAndProcessingFuture { + final Message message; + final Future future; + + MessageAndProcessingFuture(Message message, Future future) { + this.message = message; + 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..e4f9114 --- /dev/null +++ b/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java @@ -0,0 +1,65 @@ +/* + * 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; + +public final class FailedMessage { + private final Message message; + private final Throwable exception; + + public FailedMessage(Message message, Throwable exception) { + this.message = message; + this.exception = exception; + } + + public Message message() { + return message; + } + + public Throwable exception() { + return exception; + } + + public boolean isRecoverable() { + return exception instanceof RecoverableException; + } + + @Override + public String toString() { + return "FailedMessage{" + + "exception=" + exception + + ", message=" + message + + '}'; + } + + @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(message, that.message) && + Objects.equals(exception, that.exception); + } + + @Override + public int hashCode() { + return Objects.hash(message, exception); + } +} diff --git a/lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.java b/lib/src/main/java/org/esbtools/eventhandler/Message.java similarity index 85% rename from lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.java rename to lib/src/main/java/org/esbtools/eventhandler/Message.java index 84cb65e..565e0c1 100644 --- a/lib/src/main/java/org/esbtools/eventhandler/DocumentRepository.java +++ b/lib/src/main/java/org/esbtools/eventhandler/Message.java @@ -18,8 +18,8 @@ package org.esbtools.eventhandler; -import java.util.Collection; +import java.util.concurrent.Future; -public interface DocumentRepository { - void insertOrUpdate(Collection document) throws Exception; +public interface Message { + Future process(); } diff --git a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java b/lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java similarity index 87% rename from lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java rename to lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java index 1e4caa2..479531a 100644 --- a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/Document.java +++ b/lib/src/main/java/org/esbtools/eventhandler/MessageFactory.java @@ -16,8 +16,8 @@ * along with this program. If not, see . */ -package org.esbtools.eventhandler.lightblue; - -public interface Document { +package org.esbtools.eventhandler; +public interface MessageFactory { + Message getMessageForBody(Object body); } diff --git a/lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java b/lib/src/main/java/org/esbtools/eventhandler/RecoverableException.java similarity index 57% rename from lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java rename to lib/src/main/java/org/esbtools/eventhandler/RecoverableException.java index 219c16c..5af8448 100644 --- a/lib/src/main/java/org/esbtools/eventhandler/AsyncDocumentProcessorRoute.java +++ b/lib/src/main/java/org/esbtools/eventhandler/RecoverableException.java @@ -18,22 +18,12 @@ package org.esbtools.eventhandler; -import org.apache.camel.builder.RouteBuilder; - -public class AsyncDocumentProcessorRoute extends RouteBuilder { - private final DocumentRepository documentRepository; - private final String fromUri; - - public AsyncDocumentProcessorRoute(DocumentRepository documentRepository, String fromUri) { - this.documentRepository = documentRepository; - this.fromUri = fromUri; +public class RecoverableException extends EventHandlerException { + public RecoverableException(String message, Throwable cause) { + super(message, cause); } - @Override - public void configure() throws Exception { - from(fromUri) - .process(exchange -> { - documentRepository.insertOrUpdate(exchange.getIn().getBody()); - }); + public RecoverableException(Throwable cause) { + super(cause); } } diff --git a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java b/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java deleted file mode 100644 index 56e344a..0000000 --- a/lightblue/src/main/java/org/esbtools/eventhandler/lightblue/LightblueDocumentRepository.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * 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.lightblue; - -import org.esbtools.eventhandler.DocumentRepository; -import org.esbtools.eventhandler.lightblue.locking.LockStrategy; -import org.esbtools.eventhandler.lightblue.locking.LockedResource; -import org.esbtools.eventhandler.lightblue.locking.LockedResources; - -import java.util.Collection; - -public class LightblueDocumentRepository implements DocumentRepository { - private final LockStrategy lockStrategy; - - public LightblueDocumentRepository(LockStrategy lockStrategy) { - this.lockStrategy = lockStrategy; - } - - @Override - public void insertOrUpdate(Collection documents) throws Exception { - try (LockedResources lockedDocuments = parseAndWaitForLocks(documents)) { - - } - } -} From 5f28892165e1b818b26b4a771190d723c0cb6101 Mon Sep 17 00:00:00 2001 From: Alec Henninger Date: Mon, 15 Feb 2016 13:57:09 -0500 Subject: [PATCH 3/4] Add reference to original unparsed msg in FailedMessage This gives more diagnostic information, and allows us to include the subset of failures which happen before we have a parsed message. --- .../AsyncMessageProcessorRoute.java | 30 +++++++++++------- .../esbtools/eventhandler/FailedMessage.java | 31 ++++++++++++++----- 2 files changed, 41 insertions(+), 20 deletions(-) diff --git a/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java index 5663c01..2f715e1 100644 --- a/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java +++ b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java @@ -65,7 +65,7 @@ public void configure() throws Exception { Collection messages = (Collection) exchangeBody; - List processingMessages = new ArrayList<>(messages.size()); + List processingMessages = new ArrayList<>(messages.size()); List failures = new ArrayList<>(); // Start processing all of the messages in the batch in parallel. @@ -73,24 +73,28 @@ public void configure() throws Exception { try { Message parsedMessage = messageFactory.getMessageForBody(message); Future processingFuture = parsedMessage.process(); - processingMessages.add(new MessageAndProcessingFuture(parsedMessage, processingFuture)); + ProcessingMessage processing = new ProcessingMessage( + message, parsedMessage, processingFuture); + processingMessages.add(processing); } catch (Exception e) { log.error("Failure parsing message. Body was: " + message, e); - // TODO: How to treat this failure? - // We don't have a Message but FailedMessage requires one. - // Type-unsafe Failure type? + failures.add(new FailedMessage(message, e)); } } // Wait for processing to complete. - for (MessageAndProcessingFuture processingMsg : processingMessages) { + for (ProcessingMessage processingMsg : processingMessages) { try { processingMsg.future.get(processTimeout.toMillis(), TimeUnit.MILLISECONDS); } catch (InterruptedException | TimeoutException e) { RecoverableException recoverableException = new RecoverableException(e); - failures.add(new FailedMessage(processingMsg.message, recoverableException)); + FailedMessage failure = new FailedMessage(processingMsg.originalMessage, + processingMsg.parsedMessage, recoverableException); + failures.add(failure); } catch (ExecutionException e) { - failures.add(new FailedMessage(processingMsg.message, e.getCause())); + FailedMessage failure = new FailedMessage(processingMsg.originalMessage, + processingMsg.parsedMessage, e.getCause()); + failures.add(failure); } } @@ -101,12 +105,14 @@ public void configure() throws Exception { } /** Simple struct for storing a message and its future processing result. */ - private static class MessageAndProcessingFuture { - final Message message; + private static class ProcessingMessage { + final Object originalMessage; + final Message parsedMessage; final Future future; - MessageAndProcessingFuture(Message message, Future future) { - this.message = message; + 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/FailedMessage.java b/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java index e4f9114..27bb3f0 100644 --- a/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java +++ b/lib/src/main/java/org/esbtools/eventhandler/FailedMessage.java @@ -19,18 +19,31 @@ package org.esbtools.eventhandler; import java.util.Objects; +import java.util.Optional; public final class FailedMessage { - private final Message message; + private final Object originalMessage; + private final Optional parsedMessage; private final Throwable exception; - public FailedMessage(Message message, Throwable exception) { - this.message = message; + public FailedMessage(Object originalMessage, Message parsedMessage, Throwable exception) { + this.originalMessage = originalMessage; + this.parsedMessage = Optional.of(parsedMessage); this.exception = exception; } - public Message message() { - return message; + 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() { @@ -45,7 +58,8 @@ public boolean isRecoverable() { public String toString() { return "FailedMessage{" + "exception=" + exception + - ", message=" + message + + ", parsedMessage=" + parsedMessage + + ", originalMessage=" + originalMessage + '}'; } @@ -54,12 +68,13 @@ 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(message, that.message) && + return Objects.equals(originalMessage, that.originalMessage) && + Objects.equals(parsedMessage, that.parsedMessage) && Objects.equals(exception, that.exception); } @Override public int hashCode() { - return Objects.hash(message, exception); + return Objects.hash(originalMessage, parsedMessage, exception); } } From 48844966c7a2ef0b82895ef73e4da38b53a1a8cb Mon Sep 17 00:00:00 2001 From: Alec Henninger Date: Mon, 15 Feb 2016 16:52:47 -0500 Subject: [PATCH 4/4] Test AsyncMessageProcessorRoute --- .../AsyncMessageProcessorRoute.java | 17 +- .../AsyncMessageProcessorRouteTest.java | 316 ++++++++++++++++++ .../LightblueDocumentEventRepository.java | 2 +- .../LightblueNotificationRepository.java | 4 +- 4 files changed, 330 insertions(+), 9 deletions(-) create mode 100644 lib/src/test/java/org/esbtools/eventhandler/AsyncMessageProcessorRouteTest.java diff --git a/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java index 2f715e1..9533bd1 100644 --- a/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java +++ b/lib/src/main/java/org/esbtools/eventhandler/AsyncMessageProcessorRoute.java @@ -37,7 +37,7 @@ public class AsyncMessageProcessorRoute extends RouteBuilder { private final Duration processTimeout; private final MessageFactory messageFactory; - private final int idCount = idCounter.get(); + private final int idCount = idCounter.getAndIncrement(); private final String routeId = "async-message-repository-" + idCount; private static final AtomicInteger idCounter = new AtomicInteger(0); @@ -60,7 +60,8 @@ public void configure() throws Exception { 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: " + exchangeBody); + "the uri '" + fromUri + "' returned the " + + exchangeBody.getClass().getName() + ": " + exchangeBody); } Collection messages = (Collection) exchangeBody; @@ -76,6 +77,7 @@ public void configure() throws Exception { 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)); @@ -86,21 +88,24 @@ public void configure() throws Exception { 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); - } catch (ExecutionException e) { - FailedMessage failure = new FailedMessage(processingMsg.originalMessage, - processingMsg.parsedMessage, e.getCause()); - failures.add(failure); } } // Deal with failures... exchange.getIn().setBody(failures); }) + .split(body()) .to(failureUri); } 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 {