AWS - Serverless: <3> Message event with S3 & SNS

Asynchronous event trigger is from S3 and SNS.

• S3 is Amazon Simple Storage Service to storage objects that offers industry-leading scalability, data availability, security, and performance.

• SNS is Amazon Simple Notification Service.

We will do a simple flow like that:

Prerequisite:

• Complete the previous lesson to understand SAM template.

Step 1: SAM template has the change:

BucketAnh:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-${AWS::Region}
  TopicAnh:
    Type: AWS::SNS::Topic
DeliverMsgFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: customer
      Handler: com.anhnt.customer.DeliverMsg::onMessage
      Policies:
        - S3ReadPolicy:
            BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-${AWS::Region}
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt TopicAnh.TopicName
      Environment:
        Variables:
          TOPIC_ANH: !Ref TopicAnh
      Events:
        S3Event:
          Type: S3
          Properties:
            Bucket: !Ref BucketAnh
            Events: s3:ObjectCreated:*
  ConsumeMsgFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: customer
      Handler: com.anhnt.customer.DeliverMsg::onSNSMessage
      Policies:
        - S3ReadPolicy:
            BucketName: !Sub ${AWS::StackName}-${AWS::AccountId}-${AWS::Region}
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt TopicAnh.TopicName
      Events:
        SNSEvent:
          Type: SNS
          Properties:
            Topic: !Ref TopicAnh

Overview:

• !Sub : Intrinsic function - substitution of ${AWS::StackName}-${AWS::AccountId}-${AWS::Region}

• !GetAtt: Intrinsic function - Get Attribute of Resource

• Type: AWS::SNS::Topic => Create a dynamic topic name like '${AWS::StackName}-TopicAnh-xxxxxxxxx'

• Policy to access get Object from S3 and push message to SNS

Step 2: Add more code with 2 functions

• onMessage(S3Event) get event from S3 and put into SNS

• onSNSMessage(SNSEvent) get event from SNS and print value to log.

public class DeliverMsg {

    private final ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules()
  .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);

    AmazonS3 s3 = AmazonS3ClientBuilder.defaultClient();
    AmazonSNS sns = AmazonSNSClientBuilder.defaultClient();
    public void onMessage(S3Event event){
        event.getRecords().forEach(record-> {
            S3ObjectInputStream stream = s3.getObject(record.getS3().getBucket().getName(), record.getS3().getObject().getKey()).getObjectContent();
            try {
                List<CreateCustomerRequest> list = Arrays.asList(objectMapper.readValue(stream, CreateCustomerRequest[].class));
                System.out.println(list);
                list.forEach(e->{
                    try {
                        System.out.println("Environment: "+System.getenv("TOPIC_ANH"));
                        sns.publish(System.getenv("TOPIC_ANH"), objectMapper.writeValueAsString(e));
                    } catch (JsonProcessingException e1) {                        
                        e1.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            } 
        });
    }

    public void onSNSMessage(SNSEvent event){
        event.getRecords().forEach(e -> {
            try {
                CreateCustomerRequest request = objectMapper.readValue(e.getSNS().getMessage(), CreateCustomerRequest.class);
                System.out.println(request);
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        });
    }
    
}

Step 3: Build and deploy with 'sam build' and 'sam deploy'

Step 4: Go to service 'S3', then update file events.messages in bucket 'customer-xxxxxx-us-east-1' . Its content is like:

[{
  "username": "seabird86@gmail.com",
  "name": "jack1",
  "mobile": "0978958387",
  "date_of_birth": "2017-11-15",
  "balance": 10.02
},
{
  "username": "1seabird86@gmail.com",
  "name": "jack2",
  "date_of_birth": "2017-11-16",
  "balance": 10.02
}]

Step 5: Go to 'cloud watch' => 'log groups' to track:

• Check with log group 'customer-DeliverMsg..' and 'customer-ConsumeMsg..'. These log will contains the printed line from the code 'System.out.println()'.