@Test public void TestLambdaFunction() { S3Event event = new S3Event(S3Event().getRecords()); new ElasticLambdaForwarder().handleRequest(event, context()); }
@Override public Void handleRequest(S3Event event, Context context) { logger = context.getLogger(); try { initialize(); logger.log(String.format( "Parameters: hostname [%s:%d] index [%s], username [%s], pipeline [%s], ssl/tls [%b]", ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_INDEX, ELASTIC_USERNAME, ELASTIC_PIPELINE, ELASTIC_HTTPS)); RestHighLevelClient es = client(ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_USERNAME, ELASTIC_PASSWORD, ELASTIC_HTTPS); AmazonS3 s3Client = getS3Client(); BulkProcessor processor = processor(es, BULK_ACTIONS, BULK_CONCURRENCY); try { process(es, s3Client, event, processor, ELASTIC_INDEX, ELASTIC_PIPELINE); } finally { logger.log("Finished processing; flushing any remaining logs..."); processor.flush(); processor.awaitClose(60L, TimeUnit.SECONDS); logger.log("Elasticsearch processor shut down"); } } catch (Exception e) { logger.log(String.format("%s\n%s", e.getMessage(), trace(e))); } return null; }
ElasticLambdaForwarder implements RequestHandler<S3Event, Void> { @Override public Void handleRequest(S3Event event, Context context) { logger = context.getLogger(); try { initialize(); logger.log(String.format( "Parameters: hostname [%s:%d] index [%s], username [%s], pipeline [%s], ssl/tls [%b]", ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_INDEX, ELASTIC_USERNAME, ELASTIC_PIPELINE, ELASTIC_HTTPS)); RestHighLevelClient es = client(ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_USERNAME, ELASTIC_PASSWORD, ELASTIC_HTTPS); AmazonS3 s3Client = getS3Client(); BulkProcessor processor = processor(es, BULK_ACTIONS, BULK_CONCURRENCY); try { process(es, s3Client, event, processor, ELASTIC_INDEX, ELASTIC_PIPELINE); } finally { logger.log("Finished processing; flushing any remaining logs..."); processor.flush(); processor.awaitClose(60L, TimeUnit.SECONDS); logger.log("Elasticsearch processor shut down"); } } catch (Exception e) { logger.log(String.format("%s\n%s", e.getMessage(), trace(e))); } return null; } }
ElasticLambdaForwarder implements RequestHandler<S3Event, Void> { @Override public Void handleRequest(S3Event event, Context context) { logger = context.getLogger(); try { initialize(); logger.log(String.format( "Parameters: hostname [%s:%d] index [%s], username [%s], pipeline [%s], ssl/tls [%b]", ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_INDEX, ELASTIC_USERNAME, ELASTIC_PIPELINE, ELASTIC_HTTPS)); RestHighLevelClient es = client(ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_USERNAME, ELASTIC_PASSWORD, ELASTIC_HTTPS); AmazonS3 s3Client = getS3Client(); BulkProcessor processor = processor(es, BULK_ACTIONS, BULK_CONCURRENCY); try { process(es, s3Client, event, processor, ELASTIC_INDEX, ELASTIC_PIPELINE); } finally { logger.log("Finished processing; flushing any remaining logs..."); processor.flush(); processor.awaitClose(60L, TimeUnit.SECONDS); logger.log("Elasticsearch processor shut down"); } } catch (Exception e) { logger.log(String.format("%s\n%s", e.getMessage(), trace(e))); } return null; }  }
ElasticLambdaForwarder implements RequestHandler<S3Event, Void> { @Override public Void handleRequest(S3Event event, Context context) { logger = context.getLogger(); try { initialize(); logger.log(String.format( "Parameters: hostname [%s:%d] index [%s], username [%s], pipeline [%s], ssl/tls [%b]", ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_INDEX, ELASTIC_USERNAME, ELASTIC_PIPELINE, ELASTIC_HTTPS)); RestHighLevelClient es = client(ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_USERNAME, ELASTIC_PASSWORD, ELASTIC_HTTPS); AmazonS3 s3Client = getS3Client(); BulkProcessor processor = processor(es, BULK_ACTIONS, BULK_CONCURRENCY); try { process(es, s3Client, event, processor, ELASTIC_INDEX, ELASTIC_PIPELINE); } finally { logger.log("Finished processing; flushing any remaining logs..."); processor.flush(); processor.awaitClose(60L, TimeUnit.SECONDS); logger.log("Elasticsearch processor shut down"); } } catch (Exception e) { logger.log(String.format("%s\n%s", e.getMessage(), trace(e))); } return null; }  @Override Void handleRequest(S3Event event, Context context); }
ElasticLambdaForwarder implements RequestHandler<S3Event, Void> { @Override public Void handleRequest(S3Event event, Context context) { logger = context.getLogger(); try { initialize(); logger.log(String.format( "Parameters: hostname [%s:%d] index [%s], username [%s], pipeline [%s], ssl/tls [%b]", ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_INDEX, ELASTIC_USERNAME, ELASTIC_PIPELINE, ELASTIC_HTTPS)); RestHighLevelClient es = client(ELASTIC_HOSTNAME, ELASTIC_PORT, ELASTIC_USERNAME, ELASTIC_PASSWORD, ELASTIC_HTTPS); AmazonS3 s3Client = getS3Client(); BulkProcessor processor = processor(es, BULK_ACTIONS, BULK_CONCURRENCY); try { process(es, s3Client, event, processor, ELASTIC_INDEX, ELASTIC_PIPELINE); } finally { logger.log("Finished processing; flushing any remaining logs..."); processor.flush(); processor.awaitClose(60L, TimeUnit.SECONDS); logger.log("Elasticsearch processor shut down"); } } catch (Exception e) { logger.log(String.format("%s\n%s", e.getMessage(), trace(e))); } return null; }  @Override Void handleRequest(S3Event event, Context context);  }