Java 类com.amazonaws.services.lambda.runtime.events.SNSEvent.SNSRecord 实例源码

@Override
public ApiGatewayResponse handleRequest(SNSEvent input, Context context) {

    if (CollectionUtils.isNotEmpty(input.getRecords())) {
        SNSRecord record = input.getRecords().get(0);
        if (StringUtils.containsIgnoreCase(record.getSNS().getTopicArn(), "handle-emoji")) {
            LOG.info("Got message to handle-emoji topic.");
            handleSlackEvent(record);
        } else if (StringUtils.containsIgnoreCase(record.getSNS().getTopicArn(), "s3-file-ready")) {
            LOG.info("Got message to s3-file-ready topic.");
            postImageToSlack(record);
        } else if (StringUtils.containsIgnoreCase(record.getSNS().getTopicArn(), "gif-generator-error")) {
            LOG.info("Got message to gif-generator-error topic.");
            postErrorMessageToSlack(record);
        }
    }

    Response responseBody = new Response("pprxmtr-file-fetcher called succesfully.", new HashMap<>());
    return ApiGatewayResponse.builder().setStatusCode(200).setObjectBody(responseBody).build();
}

public void testHandleRequest() {
    System.out.println("handleRequest");

    SNS sns = new SNS();
    sns.setMessage("test message");

    SNSRecord record = new SNSRecord();
    record.setSns(sns);

    SNSEvent request = new SNSEvent();
    List<SNSRecord> records = new ArrayList<>();
    records.add(record);
    request.setRecords(records);

    Context context = new TestContext();
    SnsEventHandler instance = new SnsEventHandler();

    instance.handleRequest(request, context);
}

@Override
public Object handleRequest(SNSEvent snsEvent, Context context) {
    logger = context.getLogger();
    awsHelper = new AWSHelper(logger);

    logger.log("Input: " + snsEvent);

    List<SNSRecord> records = snsEvent.getRecords();
    logger.log("Passed 001");
    SNSRecord record = records.get(0);
    logger.log("Passed 002");
    SNS sns = record.getSNS();
    logger.log("Passed 003");
    String boardToProcess = sns.getMessage();
    logger.log("Passed 004");


    processBoard(boardToProcess);
    logger.log("Passed 005");

    return null;
}

@Override
public Object handleRequest(SNSEvent snsEvent, Context context) {

    awsHelper = new AWSHelper(context.getLogger());

    logger = context.getLogger();
    logger.log("Input: " + snsEvent);
    // You gotta love Java's deep objects...
    List<SNSRecord> records = snsEvent.getRecords();
    logger.log("Passed 001");
    SNSRecord record = records.get(0);
    logger.log("Passed 002");
    SNS sns = record.getSNS();
    logger.log("Passed 003");
    String sessionId = sns.getMessage();
    logger.log("Passed 004");


    context.getLogger().log("Game of Life Session ID: " + sessionId);
    logger.log("Passed 005");

    calculateBoards(sessionId, NUMBER_OF_BOARDS);
    logger.log("Passed 006");
    return sessionId;
}

@Override
protected void extendActualJerseyContainerRequest(ContainerRequest actualContainerRequest,
        JRestlessContainerRequest containerRequest, SnsRecordAndLambdaContext snsRecordAndContext) {
    SNSRecord snsRecord = snsRecordAndContext.getSnsRecord();
    Context lambdaContext = snsRecordAndContext.getLambdaContext();
    actualContainerRequest.setRequestScopedInitializer(locator -> {
        Ref<SNSRecord> snsRecordRef = locator.<Ref<SNSRecord>>getInstance(SNS_RECORD_TYPE);
        if (snsRecordRef != null) {
            snsRecordRef.set(snsRecord);
        } else {
            LOG.error("SnsFeature has not been registered. SNSRecord injection won't work.");
        }
        Ref<Context> contextRef = locator
                .<Ref<Context>>getInstance(AbstractLambdaContextReferencingBinder.LAMBDA_CONTEXT_TYPE);
        if (contextRef != null) {
            contextRef.set(lambdaContext);
        } else {
            LOG.error("AwsFeature has not been registered. Context injection won't work.");
        }
    });
}

@SuppressWarnings("unchecked")
@Test
public void delegateRequest_ValidRequestAndReferencesGiven_ShouldSetReferencesOnRequestInitialization() {

    Context context = mock(Context.class);
    SNS sns = new SNS();
    sns.setTopicArn(":t");
    SNSRecord snsRecord = new SNSRecord();
    snsRecord.setSns(sns);

    RequestScopedInitializer requestScopedInitializer = getSetRequestScopedInitializer(context, snsRecord);

    Ref<SNSRecord> snsRef = mock(Ref.class);
    Ref<Context> contextRef = mock(Ref.class);

    InjectionManager injectionManager = mock(InjectionManager.class);
    when(injectionManager.getInstance(SNS_RECORD_TYPE)).thenReturn(snsRef);
    when(injectionManager.getInstance(AbstractLambdaContextReferencingBinder.LAMBDA_CONTEXT_TYPE)).thenReturn(contextRef);

    requestScopedInitializer.initialize(injectionManager);

    verify(snsRef).set(snsRecord);
    verify(contextRef).set(context);
}

@SuppressWarnings({ "unchecked", "rawtypes" })
private RequestScopedInitializer getSetRequestScopedInitializer(Context context, SNSRecord snsRecord) {
    SnsRecordAndLambdaContext reqAndContext = new SnsRecordAndLambdaContext(snsRecord, context);
    ArgumentCaptor<Consumer> containerEnhancerCaptor = ArgumentCaptor.forClass(Consumer.class);
    snsHandler.delegateRequest(reqAndContext);
    verify(container).handleRequest(any(), any(), any(), containerEnhancerCaptor.capture());

    ContainerRequest containerRequest = mock(ContainerRequest.class);
    containerEnhancerCaptor.getValue().accept(containerRequest);

    ArgumentCaptor<RequestScopedInitializer> requestScopedInitializerCaptor = ArgumentCaptor.forClass(RequestScopedInitializer.class);

    verify(containerRequest).setRequestScopedInitializer(requestScopedInitializerCaptor.capture());

    return requestScopedInitializerCaptor.getValue();
}

@SuppressWarnings("unchecked")
@Test
public void createResponseWriter_writeResponse_Always_ShouldDelegateResponseToHandler() throws IOException {
    SnsRecordAndLambdaContext reqAndContext = mock(SnsRecordAndLambdaContext.class);
    SNS sns = new SNS();
    sns.setTopicArn(":t");
    SNSRecord snsRecord = new SNSRecord();
    snsRecord.setSns(sns);
    when(reqAndContext.getSnsRecord()).thenReturn(snsRecord);

    StatusType statusType = mock(StatusType.class);
    Map<String, List<String>> headers = mock(Map.class);
    ByteArrayOutputStream entityOutputStream = mock(ByteArrayOutputStream.class);
    snsHandler.createResponseWriter(reqAndContext).writeResponse(statusType, headers, entityOutputStream);
    verify(snsHandler).handleReponse(reqAndContext, statusType, headers, entityOutputStream);
}

@Override
public void handler(SNSEvent event, Context context) throws HandlerException {
  if (!initialized) {
    init(context);
  }
  this.source = this.sources.get(0);
  this.inputFiles = new ArrayList<String>(0);

  for (SNSRecord record : event.getRecords()) {
    /*
     * Parse SNS as a S3 notification
     */
    String json = record.getSNS().getMessage();
    S3EventNotification s3Event = S3EventNotification.parseJson(json);

    /*
     * Validate the S3 file matches the regex
     */
    List<S3EventNotificationRecord> toProcess =
        new ArrayList<S3EventNotificationRecord>(s3Event.getRecords());
    for (S3EventNotificationRecord s3Record : s3Event.getRecords()) {
      String s3Path = String.format("s3://%s/%s", s3Record.getS3().getBucket().getName(),
          s3Record.getS3().getObject().getKey());
      try {
        this.source = SourceUtils.getSource(s3Path, this.sources);
      } catch (SourceNotFoundException e) {
        logger.warn("skipping processing " + s3Path);
        toProcess.remove(s3Record);
      }
    }

    if (toProcess.size() == 0) {
      logger.warn("Nothing to process");
      return;
    }

    this.inputFiles.addAll(toProcess.stream().map(m -> {
      return m.getS3().getObject().getKey();
    }).collect(Collectors.toList()));

    this.recordIterator = new S3EventIterator(context, toProcess, s3ClientFactory);

    super.process(context);
  }
}

@Override
public Void handleRequest(SNSEvent snsEvent, Context context) {
    for (SNSRecord snsRecord : snsEvent.getRecords()) {
        delegateRequest(new SnsRecordAndLambdaContext(snsRecord, context));
    }
    return null;
}

@Test
public void delegateRequest_ValidRequestAndNoReferencesGiven_ShouldNotFailOnRequestInitialization() {

    Context context = mock(Context.class);
    SNS sns = new SNS();
    sns.setTopicArn(":t");
    SNSRecord snsRecord = new SNSRecord();
    snsRecord.setSns(sns);

    RequestScopedInitializer requestScopedInitializer = getSetRequestScopedInitializer(context, snsRecord);

    InjectionManager injectionManager = mock(InjectionManager.class);
    requestScopedInitializer.initialize(injectionManager);
}

private SnsRecordAndLambdaContext createMinimalRequest() {
    SNS sns = new SNS();
    sns.setTopicArn(":t");
    SNSRecord snsRecord = new SNSRecord();
    snsRecord.setSns(sns);
    return new SnsRecordAndLambdaContext(snsRecord, null);
}

@Test
public void testMultipleRecordsCreateMultipleRequest() {
    SNSEvent snsEvent = new SNSEvent();
    SNSRecord snsRecord0 = createSnsRecord("a:b:mytopic", "inject-sns-record-member0");
    SNSRecord snsRecord1 = createSnsRecord("a:b:mytopic", "inject-sns-record-member1");
    snsEvent.setRecords(ImmutableList.of(snsRecord0, snsRecord1));
    handler.handleRequest(snsEvent, context);
    InOrder inOrder = Mockito.inOrder(testService);
    inOrder.verify(testService).injectedSns(snsRecord0.getSNS());
    inOrder.verify(testService).injectedSns(snsRecord1.getSNS());
}

private SNSRecord createSnsRecord(String topicArn, String subject) {
    SNS sns = new SNS();
    sns.setTopicArn(topicArn);
    sns.setSubject(subject);
    SNSRecord snsRecord = new SNSRecord();
    snsRecord.setSns(sns);
    return snsRecord;
}

@Override
public SNSEvent getTestEvent() throws Exception {
  /*
   * Upload a test resoruce to the mock S3
   */
  String payload = IOUtils.toString(
      new InputStreamReader(this.getClass().getResourceAsStream("basic_input.log"), "UTF-8"));
  this.client.putObject(S3_BUCKET, "basic_input.log", payload);

  /*
   * Create a S3EventNotification event
   */
  S3ObjectEntity objEntity = new S3ObjectEntity("basic_input.log", 1L, null, null);
  S3BucketEntity bucketEntity = new S3BucketEntity(S3_BUCKET, null, null);
  S3Entity entity = new S3Entity(null, bucketEntity, objEntity, null);

  S3EventNotificationRecord rec = new S3EventNotificationRecord(null, null, null,
      "1970-01-01T00:00:00.000Z", null, null, null, entity, null);

  List<S3EventNotificationRecord> notifications = new ArrayList<S3EventNotificationRecord>(2);
  notifications.add(rec);

  /*
   * Wrap as an SNS Event
   */
  S3EventNotification event = new S3EventNotification(notifications);
  SNSEvent.SNS sns = new SNSEvent.SNS();
  sns.setMessage(event.toJson());

  SNSEvent snsEvent = new SNSEvent();

  ArrayList<SNSRecord> snsRecords = new ArrayList<SNSRecord>(1);
  SNSRecord snsRecord = new SNSRecord();
  snsRecord.setEventSource("aws:sns");
  snsRecord.setEventVersion("1.0");
  snsRecord.setEventSubscriptionArn("arn");

  snsRecord.setSns(sns);
  snsRecords.add(snsRecord);
  snsEvent.setRecords(snsRecords);

  return snsEvent;
}

private SNSEvent getTestEvent(String bucket, boolean doPut) throws Exception {
  /*
   * Upload a test resoruce to the mock S3
   */
  if (doPut) {
    String payload = IOUtils.toString(
        new InputStreamReader(this.getClass().getResourceAsStream("basic_input.log"), "UTF-8"));
    this.client.putObject(bucket, "basic_input.log", payload);
  }
  /*
   * Create a S3EventNotification event
   */
  S3ObjectEntity objEntity = new S3ObjectEntity("basic_input.log", 1L, null, null);
  S3BucketEntity bucketEntity = new S3BucketEntity(bucket, null, null);
  S3Entity entity = new S3Entity(null, bucketEntity, objEntity, null);

  S3EventNotificationRecord rec = new S3EventNotificationRecord(null, null, null,
      "1970-01-01T00:00:00.000Z", null, null, null, entity, null);

  List<S3EventNotificationRecord> notifications = new ArrayList<S3EventNotificationRecord>(2);
  notifications.add(rec);

  /*
   * Wrap as an SNS Event
   */
  S3EventNotification event = new S3EventNotification(notifications);
  SNSEvent.SNS sns = new SNSEvent.SNS();
  sns.setMessage(event.toJson());

  SNSEvent snsEvent = new SNSEvent();

  ArrayList<SNSRecord> snsRecords = new ArrayList<SNSRecord>(1);
  SNSRecord snsRecord = new SNSRecord();
  snsRecord.setEventSource("aws:sns");
  snsRecord.setEventVersion("1.0");
  snsRecord.setEventSubscriptionArn("arn");

  snsRecord.setSns(sns);
  snsRecords.add(snsRecord);
  snsEvent.setRecords(snsRecords);

  return snsEvent;
}

@Override
protected void configure() {
    bindReferencingLambdaContextFactory();
    bindReferencingFactory(SNSRecord.class, ReferencingSnsRecordFactory.class,
            new GenericType<Ref<SNSRecord>>() { });
}

@Inject
ReferencingSnsRecordFactory(final Provider<Ref<SNSRecord>> referenceFactory) {
    super(referenceFactory);
}

public SnsRecordAndLambdaContext(SNSRecord snsRecord, Context lambdaContext) {
    this.snsRecord = snsRecord;
    this.lambdaContext = lambdaContext;
}

public SNSRecord getSnsRecord() {
    return snsRecord;
}

@Path("/inject-sns-record")
@POST
public void injectSnsRecord(@javax.ws.rs.core.Context SNSRecord snsRecord) {
    service.injectedSns(snsRecord.getSNS());
}

@Path("/inject-sns-record-mock")
@POST
public void injectSnsRecordMock(@javax.ws.rs.core.Context SNSRecord snsRecord) {
    service.injectedSnsRecord(snsRecord);
}

void injectedSnsRecord(SNSRecord snsRecord);