AsyncMessageProcessorRoute

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 <http://www.gnu.org/licenses/>.
+ */
+
+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<ProcessingMessage> processingMessages = new ArrayList<>(messages.size());
+            List<FailedMessage> 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;
+        }
+    }
+}

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);
     }

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 <http://www.gnu.org/licenses/>.
+ */
+
+package org.esbtools.eventhandler;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public final class FailedMessage {
+    private final Object originalMessage;
+    private final Optional<Message> 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<Message> 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);
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ */
+
+package org.esbtools.eventhandler;
+
+import java.util.concurrent.Future;
+
+public interface Message {
+    Future<?> process();
+}

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 <http://www.gnu.org/licenses/>.
+ */
+
+package org.esbtools.eventhandler;
+
+public interface MessageFactory {
+    Message getMessageForBody(Object body);
+}

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 <http://www.gnu.org/licenses/>.
+ */
+
+package org.esbtools.eventhandler;
+
+public class RecoverableException extends EventHandlerException {
+    public RecoverableException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public RecoverableException(Throwable cause) {
+        super(cause);
+    }
+}