Variant 1

Author: Scott Gerring

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/BatchMessageHandlerBuilder.java

@@ -1,10 +1,6 @@
 package software.amazon.lambda.powertools.batch;
 
-import com.amazonaws.services.lambda.runtime.events.DynamodbEvent;
-import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
-import com.amazonaws.services.lambda.runtime.events.SQSEvent;
-import software.amazon.lambda.powertools.batch.BatchMessageProcessor;
-import software.amazon.lambda.powertools.batch.message.*;
+import com.amazonaws.services.lambda.runtime.events.*;
 import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
 import java.time.Duration;
@@ -32,15 +28,15 @@ public BatchMessageHandlerBuilder withTimeout(Duration timeout) {
     }
 
     // TODO - can we put a meaningful type on the message, or do we need to do it, per-message-type?
-    public List<String> process(SQSEvent message, Consumer<SqsMessage> messageHandler) {
+    public SQSBatchResponse process(SQSEvent message, Consumer<SQSEvent.SQSMessage> messageHandler) {
         throw new NotImplementedException();
     };
 
-    public List<String> process(KinesisEvent message, Consumer<KinesisDataStreamsMessage> messageHandler) {
+    public StreamsEventResponse process(KinesisEvent message, Consumer<KinesisEvent.KinesisEventRecord> messageHandler) {
         throw new NotImplementedException();
     }
 
-    public List<String> process(DynamodbEvent message, Consumer<DynamoDbStreamMessage> messageHandler) {
+    public StreamsEventResponse process(DynamodbEvent message, Consumer<DynamodbEvent.DynamodbStreamRecord> messageHandler) {
         throw new NotImplementedException();
     }
 

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/BatchMessageProcessor.java

@@ -0,0 +1,53 @@
+package software.amazon.lambda.powertools.batch;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.sun.xml.internal.ws.api.message.Message;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public abstract class BatchRequestHandler<T, U, V> implements RequestHandler<T, V> {
+
+    protected class MessageProcessingResult<U> {
+        private final U message;
+        private final Exception exception;
+
+        public MessageProcessingResult(U message, Exception exception) {
+            this.message = message;
+            this.exception = exception;
+        }
+    }
+
+    @Override
+    public V handleRequest(T input, Context context) {
+        // Extract messages
+        List<U> messages = extractMessages(input);
+
+        // Try process them
+        List<MessageProcessingResult<U>> results = messages.stream().map(m -> {
+            try {
+                processItem(m, context);
+                return new MessageProcessingResult<>(m, null);
+            } catch (Exception e) {
+                return new MessageProcessingResult<>(m, e);
+            }
+        }).collect(Collectors.toList());
+
+        // Generate the response
+        return writeResponse(results);
+    }
+
+    /**
+     * Provided by the event-specific child to extract the individual records
+     * from the batch request
+     *
+     * @param input
+     * @return
+     */
+    protected abstract List<U> extractMessages(T input);
+
+    protected abstract V writeResponse(Iterable<MessageProcessingResult<U>> results);
+
+    public abstract void processItem(U message, Context context);
+}

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/BatchRequestHandler.java

@@ -0,0 +1,23 @@
+package software.amazon.lambda.powertools.batch;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
+import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import java.util.List;
+
+public abstract class KinesisBatchRequestHandler extends BatchRequestHandler<KinesisEvent, KinesisEvent.KinesisEventRecord, StreamsEventResponse> {
+
+    @Override
+    protected List<KinesisEvent.KinesisEventRecord> extractMessages(KinesisEvent input) {
+        return input.getRecords();
+    }
+
+    @Override
+    protected StreamsEventResponse writeResponse(Iterable<MessageProcessingResult<KinesisEvent.KinesisEventRecord>> results) {
+        // Here we map up the kinesis-specific response for the batch based on the success of the individual messages
+        throw new NotImplementedException();
+    }
+
+}

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/KinesisBatchRequestHandler.java

@@ -0,0 +1,24 @@
+package software.amazon.lambda.powertools.batch;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
+import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import java.util.List;
+
+public abstract class SQSBatchRequestHandler extends BatchRequestHandler<SQSEvent, SQSEvent.SQSMessage, SQSBatchResponse> {
+
+    @Override
+    protected List<SQSEvent.SQSMessage> extractMessages(SQSEvent input) {
+        return input.getRecords();
+    }
+
+    @Override
+    protected SQSBatchResponse writeResponse(Iterable<MessageProcessingResult<SQSEvent.SQSMessage>> results) {
+        // Here we map up the SQS-specific response for the batch based on the success of the individual messages
+        throw new NotImplementedException();
+    }
+}

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/SQSBatchRequestHandler.java

@@ -0,0 +1,13 @@
+package software.amazon.lambda.powertools.batch.examples;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import software.amazon.lambda.powertools.batch.SQSBatchRequestHandler;
+
+public class ExampleBatchRequestHandler extends SQSBatchRequestHandler {
+
+    @Override
+    public void processItem(SQSEvent.SQSMessage message, Context context) {
+        // Process an SQS message without throwing
+    }
+}

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch/examples/ExampleBatchRequestHandler.java

@@ -4,22 +4,17 @@
 import com.amazonaws.services.lambda.runtime.RequestHandler;
 import com.amazonaws.services.lambda.runtime.events.SQSEvent;
 import software.amazon.lambda.powertools.batch.BatchMessageHandlerBuilder;
-import software.amazon.lambda.powertools.batch.BatchMessageProcessor;
-import software.amazon.lambda.powertools.batch.message.BatchProcessorMessageType;
-import software.amazon.lambda.powertools.batch.message.SqsMessage;
-
-import java.util.List;
 
 /**
  * This is just here for illustrative purposes, and won't
  * be released with this code!
  */
-public class ExampleMessageHandler implements RequestHandler<SQSEvent, List<String>> {
+public class ExampleMessageHandlerBuilder implements RequestHandler<SQSEvent, Object> {
 
     @Override
-    public List<String> handleRequest(SQSEvent sqsEvent, Context context) {
+    public Object handleRequest(SQSEvent sqsEvent, Context context) {
         return new BatchMessageHandlerBuilder()
-                .process(sqsEvent, (SqsMessage message) ->  {
+                .process(sqsEvent, (SQSEvent.SQSMessage message) ->  {
                     // Process the message without throwing an exception
                 });
     }