import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
public List<S3SNSNotification> parse(Message message) {
    List<S3SNSNotification> notifications = Lists.newArrayList();

    try {
        SQSMessage envelope = om.readValue(message.getBody(), SQSMessage.class);

        if (envelope.message == null) {
            return Collections.emptyList();
        }

        S3EventNotification s3EventNotification = S3EventNotification.parseJson(envelope.message);

        notifications.addAll(s3EventNotification.getRecords().stream().map(record -> new S3SNSNotification(
                message.getReceiptHandle(),
                record.getS3().getBucket().getName(),
                record.getS3().getObject().getUrlDecodedKey()
        )).collect(Collectors.toList()));
    } catch (Exception e) {
        LOG.error("Could not parse SNS notification: " + message.getBody(), e);
        throw new RuntimeException("Could not parse SNS notification: " + message.getBody(), e);
    }

    return notifications;
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
/**
 * Helper method that parses a JSON object from a resource on the classpath
 * as an instance of the provided type.
 *
 * @param resource
 *            the path to the resource (relative to this class)
 * @param clazz
 *            the type to parse the JSON into
 */
public static <T> T parse(String resource, Class<T> clazz)
		throws IOException {

	InputStream stream = TestUtils.class.getResourceAsStream(resource);
	try {
		if (clazz == S3Event.class) {
			String json = IOUtils.toString(stream);
			S3EventNotification event = S3EventNotification.parseJson(json);

			@SuppressWarnings("unchecked")
			T result = (T) new S3Event(event.getRecords());
			return result;

		} else if (clazz == SNSEvent.class) {
			return snsEventMapper.readValue(stream, clazz);
		} else if (clazz == DynamodbEvent.class) {
			return dynamodbEventMapper.readValue(stream, clazz);
		} else {
			return mapper.readValue(stream, clazz);
		}
	} finally {
		stream.close();
	}
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Test
public void testS3EventViaSNSEventHandlerSuccessfully() throws Exception {
    InputStream input = this.getClass().getClassLoader().getResourceAsStream(S3_VIA_SNS_REQUEST_JSON_TEMPLATE);

    new Expectations(S3EventAction.class) {
        {
            new S3EventAction().handle((EventActionRequest<S3EventNotification>)any, (Context)any);
            times = 1;
        }
    };

    Optional<S3EventResult> result = executor.apply(readJson(input), mockContext);
    assertEquals(1, result.get().getSuccessItems().size());
    assertEquals(0, result.get().getFailureItems().size());
    assertEquals(0, result.get().getSkippedItems().size());
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Test
public void testS3EventViaSNSEventHandlerSuccessfully() throws Exception {
    InputStream input = this.getClass().getClassLoader().getResourceAsStream(S3_REQUEST_JSON_TEMPLATE);

    new Expectations(S3EventAction.class) {
        {
            new S3EventAction().handle((EventActionRequest<S3EventNotification>)any, (Context)any);
            times = 1;
        }
    };

    S3EventResult result = handler.handleRequest(supplyEvent(input), mockContext);
    assertEquals(1, result.getSuccessItems().size());
    assertEquals(0, result.getFailureItems().size());
    assertEquals(0, result.getSkippedItems().size());
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Override
public S3EventNotification 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);

  return new S3EventNotification(notifications);
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
private S3EventNotification 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);

  return new S3EventNotification(notifications);
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Test
public void testS3MessageS3FileNoExists() throws Exception
{
    setLogLevel(UploadDownloadHelperServiceImpl.class, LogLevel.OFF);

    uploadDownloadServiceTestHelper.createDatabaseEntitiesForUploadDownloadTesting();

    UploadSingleInitiationResponse resultUploadSingleInitiationResponse = uploadDownloadService.initiateUploadSingle(uploadDownloadServiceTestHelper
        .createUploadSingleInitiationRequest(NAMESPACE, BDEF_NAME, FORMAT_USAGE_CODE, FORMAT_FILE_TYPE_CODE, FORMAT_VERSION, NAMESPACE, BDEF_NAME_2,
            FORMAT_USAGE_CODE_2, FORMAT_FILE_TYPE_CODE_2, FORMAT_VERSION_2, TARGET_S3_KEY));

    String filePath = resultUploadSingleInitiationResponse.getSourceBusinessObjectData().getStorageUnits().get(0).getStorageFiles().get(0).getFilePath();

    S3Entity s3Entity = new S3Entity(null, null, new S3ObjectEntity(filePath, 0L, null, null), null);

    List<S3EventNotificationRecord> records = new ArrayList<>();
    records.add(new S3EventNotificationRecord(null, null, null, null, null, null, null, s3Entity, null));

    S3EventNotification s3EventNotification = new S3EventNotification(records);

    setLogLevel(UploadDownloadServiceImpl.class, LogLevel.OFF);
    setLogLevel(HerdJmsMessageListener.class, LogLevel.OFF);

    // Try to process an S3 JMS message, when source S3 file does not exist.
    herdJmsMessageListener.processMessage(jsonHelper.objectToJson(s3EventNotification), null);
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
/**
 * Helper method that parses a JSON object from a resource on the classpath
 * as an instance of the provided type.
 *
 * @param resource the path to the resource (relative to this class)
 * @param clazz the type to parse the JSON into
 */
public static <T> T parse(String resource, Class<T> clazz)
        throws IOException {

    InputStream stream = TestUtils.class.getResourceAsStream(resource);
    try {
        if (clazz == S3Event.class) {
            String json = IOUtils.toString(stream);
            S3EventNotification event = S3EventNotification.parseJson(json);

            @SuppressWarnings("unchecked")
            T result = (T) new S3Event(event.getRecords());
            return result;

        } else {
            return mapper.readValue(stream, clazz);
        }
    } finally {
        stream.close();
    } 
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Override
public Void handleRequest(S3Event s3Event, Context context) {

    Collection<Partition> requiredPartitions = new HashSet<>();
    TableService tableService = new TableService();

    for (S3EventNotification.S3EventNotificationRecord record : s3Event.getRecords()) {

        String bucket = record.getS3().getBucket().getName();
        String key = record.getS3().getObject().getKey();

        System.out.printf("S3 event [Event: %s, Bucket: %s, Key: %s]%n", record.getEventName(), bucket, key);

        S3Object s3Object = new S3Object(bucket, key);

        if (s3Object.hasDateTimeKey()) {
            requiredPartitions.add(partitionConfig.createPartitionFor(s3Object));
        }
    }

    if (!requiredPartitions.isEmpty()) {
        Collection<Partition> missingPartitions = determineMissingPartitions(
                partitionConfig.tableName(),
                requiredPartitions,
                tableService);
        tableService.addPartitions(partitionConfig.tableName(), missingPartitions);
    }

    return null;
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Override
public Void handleRequest(S3Event s3Event, Context context) {

    Collection<Partition> partitionsToRemove = new HashSet<>();
    TableService tableService = new TableService();

    for (S3EventNotification.S3EventNotificationRecord record : s3Event.getRecords()) {

        String bucket = record.getS3().getBucket().getName();
        String key = record.getS3().getObject().getKey();

        System.out.printf("S3 event [Event: %s, Bucket: %s, Key: %s]%n", record.getEventName(), bucket, key);

        S3Object s3Object = new S3Object(bucket, key);

        if (s3Object.hasDateTimeKey()) {
            partitionsToRemove.add(partitionConfig.createPartitionFor(s3Object));
        }
    }

    if (!partitionsToRemove.isEmpty()) {
        tableService.removePartitions(
                partitionConfig.tableName(),
                partitionsToRemove.stream().map(Partition::spec).collect(Collectors.toList()));
    }

    return null;
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Override
public Void handleRequest(S3Event s3Event, Context context){

    Collection<Partition>requiredPartitions = new HashSet<>();
    TableService tableService = new TableService();
    DynamoDB dynamoDBClient=new DynamoDB(new AmazonDynamoDBClient(new EnvironmentVariableCredentialsProvider()));

    for(S3EventNotification.S3EventNotificationRecord record:s3Event.getRecords()){

        String bucket=record.getS3().getBucket().getName();
        String key=record.getS3().getObject().getKey();

        System.out.printf("S3event[Event:%s,Bucket:%s,Key:%s]%n",record.getEventName(),bucket,key);

        S3Object s3Object=new S3Object(bucket,key);

        if(s3Object.hasDateTimeKey()){
            Partition partition = partitionConfig.createPartitionFor(s3Object);

            //Check if the partition exists in DynamoDBtable, if not add the partition details to the table, skip otherwise
            if (tryAddMissingPartition(partitionConfig.dynamoDBTableName(), dynamoDBClient, partition)) {
                requiredPartitions.add(partition);
            }
        }
    }

    if(!requiredPartitions.isEmpty()){
        tableService.addPartitions(partitionConfig.tableName(),requiredPartitions, true);
    }

    return null;
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Override
protected S3EventResult handleEvent(S3EventNotification event, Context context) {
    AwsEventRequest request = readEvent(event);
    S3EventResult result = new S3EventResult();
    AwsEventResponse res = actionRouterHandle(request, context);
    if (res.isSuccessful()) {
        result.addSuccessItem(request);
    } else {
        logger.error("Failed processing S3Event", res.getCause());
        result.addFailureItem(request);
    }
    return result;
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Test
public void testResolveS3EventViaSNSEventSuccessfully() throws Exception {
    InputStream input = this.getClass().getClassLoader().getResourceAsStream(S3_VIA_SNS_REQUEST_JSON_TEMPLATE);

    Optional<S3EventNotification> event = executor.resolve(readJson(input));
    assertEquals(1, event.get().getRecords().size());
    assertEquals("aws:s3", event.get().getRecords().get(0).getEventSource());
    assertEquals("us-east-1", event.get().getRecords().get(0).getAwsRegion());
    assertEquals("dev-nsmg-logs-temp", event.get().getRecords().get(0).getS3().getBucket().getName());
    assertEquals("transformed_sample_logs.json", event.get().getRecords().get(0).getS3().getObject().getKey());
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
private S3EventNotification supplyEvent(InputStream input) {
    try {
        return om.readValue(input, S3EventNotification.class);
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Override
public Handler<S3EventNotification> getHandler() {
  S3Handler handler = new S3Handler();
  handler.s3ClientFactory = this.clientFactory;

  return handler;
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@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);
  }
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
public void handler(S3EventNotification event, Context context) throws HandlerException {
  if (!initialized) {
    init(context);
  }

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

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

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

  super.process(context);
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Test
public void testSourceRegexFail() throws Throwable {
  BaseHandler.CONFIG_FILE = "/com/nextdoor/bender/handler/config_s3_source.json";

  TestContext ctx = new TestContext();
  ctx.setFunctionName("unittest");
  ctx.setInvokedFunctionArn("arn:aws:lambda:us-east-1:123:function:test-function:staging");

  BaseHandler<S3EventNotification> handler = (BaseHandler) getHandler();
  handler.init(ctx);

  handler.handler(getTestEvent("foo", false), ctx);
  assertEquals(0, DummyTransportHelper.BufferedTransporter.output.size());
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
@Test
public void testSourceRegex() throws Throwable {
  BaseHandler.CONFIG_FILE = "/com/nextdoor/bender/handler/config_s3_source.json";

  TestContext ctx = new TestContext();
  ctx.setFunctionName("unittest");
  ctx.setInvokedFunctionArn("arn:aws:lambda:us-east-1:123:function:test-function:staging");

  BaseHandler<S3EventNotification> handler = (BaseHandler) getHandler();
  handler.init(ctx);

  handler.handler(getTestEvent(), ctx);
  assertEquals(1, DummyTransportHelper.BufferedTransporter.output.size());
}
 

import com.amazonaws.services.s3.event.S3EventNotification; //导入依赖的package包/类
public static S3EventNotification getS3Notification(String key, String bucket, long size) {
  S3ObjectEntity objEntity = new S3ObjectEntity(key, size, null, null);
  S3BucketEntity bucketEntity = new S3BucketEntity(bucket, null, null);
  S3Entity entity = new S3Entity(null, bucketEntity, objEntity, null);

  String timestamp = formatter.print(System.currentTimeMillis());
  S3EventNotificationRecord rec =
      new S3EventNotificationRecord(null, null, null, timestamp, null, null, null, entity, null);

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

  return new S3EventNotification(notifications);
}