Some more examples

Author: Scott Gerring

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

@@ -7,9 +7,12 @@
 import java.util.List;
 import java.util.function.Consumer;
 
+
 /**
- * Let's make this builder-style so that we can extend the interface
- * if we need to with extra fields without breaking clients
+ * A builder-style interface we can use within an existing Lambda RequestHandler to
+ * deal with our batch responses.
+ *
+ * @see software.amazon.lambda.powertools.batch.examples.ExampleMessageHandlerBuilder
  */
 public class BatchMessageHandlerBuilder {
 

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

@@ -7,8 +7,26 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+/**
+ * An abstract base class that will be extended per-event-source to supply event-source
+ * specific functionality. As much as possible is pulled into the base class, and
+ * event-source specific logic is delegated downwards (e.g., mapping errors to response,
+ * extracting messages from batch).
+ *
+ * IRL, this would be implemented leaning on the {@link BatchMessageHandlerBuilder}, here
+ * I have provided implementation inline for illustrative purposes.
+ *
+ * @param <T> The batch event type
+ * @param <U> The individual message type for each message within teh batch
+ * @param <V> The batch result type
+ */
 public abstract class BatchRequestHandler<T, U, V> implements RequestHandler<T, V> {
 
+    /**
+     * Used to wrap the result of processing a single message. We wrap the message itself,
+     * and optionally an exception that was raised during the processing. A lack of
+     * exception indicates success.
+     */
     protected class MessageProcessingResult<U> {
         private final U message;
         private final Exception exception;
@@ -19,12 +37,19 @@ public MessageProcessingResult(U message, Exception exception) {
         }
     }
 
+    /**
+     * The batch processing logic goes here. This can be generic across all message types.
+     *
+     * @param input The batch message to process
+     * @param context The Lambda execution environment context object.
+     * @return
+     */
     @Override
     public V handleRequest(T input, Context context) {
-        // Extract messages
+        // Extract messages from batch
         List<U> messages = extractMessages(input);
 
-        // Try process them
+        // For each message, map it to either 1/ a successful result, or 2/ an exception
         List<MessageProcessingResult<U>> results = messages.stream().map(m -> {
             try {
                 processItem(m, context);
@@ -34,20 +59,32 @@ public V handleRequest(T input, Context context) {
             }
         }).collect(Collectors.toList());
 
-        // Generate the response
+        // Generate the response from the list of results
         return writeResponse(results);
     }
 
     /**
      * Provided by the event-specific child to extract the individual records
      * from the batch request
      *
-     * @param input
-     * @return
+     * @param input The batch
+     * @return the messages within the batch
      */
     protected abstract List<U> extractMessages(T input);
 
+    /**
+     * Given the set of message processing results, generates the appropriate batch
+     * processing result to return to Lambda.
+     *
+     * @param results the result of processing each message, and the messages themselves
+     * @return the batch processing result to return to lambda
+     */
     protected abstract V writeResponse(Iterable<MessageProcessingResult<U>> results);
 
+    /**
+     * To be provided by the user. Processes an individual message within the batch
+     * @param message
+     * @param context
+     */
     public abstract void processItem(U message, Context context);
 }

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

@@ -4,8 +4,10 @@
 import com.amazonaws.services.lambda.runtime.events.SQSEvent;
 import software.amazon.lambda.powertools.batch.SQSBatchRequestHandler;
 
+/**
+ * An example handler that is implemented by extending our {@link 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/ExampleMessageHandlerBuilder.java

@@ -6,8 +6,8 @@
 import software.amazon.lambda.powertools.batch.BatchMessageHandlerBuilder;
 
 /**
- * This is just here for illustrative purposes, and won't
- * be released with this code!
+ * An example handler that is implemented by using the {@link BatchMessageHandlerBuilder} inline
+ * in an existing RequestHandler.
  */
 public class ExampleMessageHandlerBuilder implements RequestHandler<SQSEvent, Object> {
 

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch2/BatchProcessor.java

@@ -0,0 +1,47 @@
+package software.amazon.lambda.powertools.batch2;
+
+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 sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+public interface BatchProcessor<I, O> {
+
+    default O processBatch(I input) {
+        // depending on the input type (SQS/Kinesis/Dynamo/...), create the appropriate response
+
+        // TODO - need a way of extracting items from input. Defer to interface?
+
+        // browse the list of items
+        // for each item
+        try {
+            // Need to typematch to get the processItem calls to bind
+            if (input instanceof SQSEvent.SQSMessage) {
+                processItem((SQSEvent.SQSMessage) input);
+            } else if (input instanceof KinesisEvent.KinesisEventRecord) {
+                processItem((KinesisEvent.KinesisEventRecord) input);
+            } else if (input instanceof DynamodbEvent.DynamodbStreamRecord) {
+                processItem((DynamodbEvent.DynamodbStreamRecord) input);
+            }
+        }
+        catch(Throwable t) {
+                // put item in item failure list
+        }
+
+        // TODO - need a way of converting resultset back to return type. Defer to interface?
+
+        throw new NotImplementedException();
+    }
+
+    default void processItem(SQSEvent.SQSMessage message) {
+        System.out.println(message.getMessageId());
+    }
+
+    default void processItem(KinesisEvent.KinesisEventRecord record) {
+        System.out.println(record.getEventID());
+    }
+
+    default void processItem(DynamodbEvent.DynamodbStreamRecord record) {
+        System.out.println(record.getEventID());
+    }
+}

powertools-batch/src/main/java/software/amazon/lambda/powertools/batch2/ProcessSQSMessageBatch.java

@@ -0,0 +1,29 @@
+package software.amazon.lambda.powertools.batch2;
+
+import com.amazonaws.services.lambda.runtime.Context;
+import com.amazonaws.services.lambda.runtime.RequestHandler;
+import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+
+public class ProcessSQSMessageBatch implements RequestHandler<SQSEvent, SQSBatchResponse>, BatchProcessor<SQSEvent, SQSBatchResponse> {
+
+    @Override
+    public SQSBatchResponse handleRequest(SQSEvent sqsEvent, Context context) {
+        // processBatch is implemented as default in the interface and handle exceptions in the processElement function to add the item to the BatchItemFailure list
+
+        // TODO - as we're getting this on the outside, we could pass in `sqsEvent.getMessages()` which would
+        // fix the "how do I extract event-specific message" problem in the interface default.
+        return this.processBatch(sqsEvent); // we may need to pass context too... ?
+
+        // TODO - as we're still on the outside, we could take the list of results from `this.processBatch()` and
+        // map it back to the SQS-specific response, again solving the "how do I do this event-specific thing"
+        // problem in the base class.
+    }
+
+    // this method comes from the BatchProcessor interface, developers need to override the appropriate one
+    @Override
+    public void processItem(SQSEvent.SQSMessage message) {
+        // do some stuff with this item
+    }
+
+}

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

@@ -0,0 +1,13 @@
+package software.amazon.lambda.powertools.batch2.examples;
+
+import com.amazonaws.services.lambda.runtime.events.SQSEvent;
+import software.amazon.lambda.powertools.batch2.ProcessSQSMessageBatch;
+
+public class ExampleBatchRequestHandler extends ProcessSQSMessageBatch {
+
+    // this method comes from the BatchProcessor interface, developers need to override the appropriate one
+    @Override
+    public void processItem(SQSEvent.SQSMessage message) {
+        // do some stuff with this item
+    }
+}