Merge pull request #2 from picwelltimjones/master

Fix 'Error: write after end at' seen in Lambda

Author: srisub-amzn

src/s3_lambda_es.js

@@ -42,40 +42,11 @@ var numDocsAdded = 0;   // Number of log lines added to ES so far
  */
 var creds = new AWS.EnvironmentCredentials('AWS');
 
-
-/* == Streams ==
- * To avoid loading an entire (typically large) log file into memory,
- * this is implemented as a pipeline of filters, streaming log data
- * from S3 to ES.
- * Flow: S3 file stream -> Log Line stream -> Log Record stream -> ES
- */
-var lineStream = new LineStream();
-// A stream of log records, from parsing each log line
-var recordStream = new stream.Transform({objectMode: true})
-recordStream._transform = function(line, encoding, done) {
-    var logRecord = parse(line.toString());
-    var serializedRecord = JSON.stringify(logRecord);
-    this.push(serializedRecord);
-    totLogLines ++;
-    done();
-}
-
-
-/* Lambda "main": Execution starts here */
-exports.handler = function(event, context) {
-    console.log('Received event: ', JSON.stringify(event, null, 2));
-    event.Records.forEach(function(record) {
-        var bucket = record.s3.bucket.name;
-        var objKey = decodeURIComponent(record.s3.object.key.replace(/\+/g, ' '));
-        s3LogsToES(bucket, objKey, context);
-    });
-}
-
 /*
  * Get the log file from the given S3 bucket and key.  Parse it and add
  * each log record to the ES domain.
  */
-function s3LogsToES(bucket, key, context) {
+function s3LogsToES(bucket, key, context, lineStream, recordStream) {
     // Note: The Lambda function should be configured to filter for .log files
     // (as part of the Event Source "suffix" setting).
 
@@ -137,3 +108,31 @@ function postDocumentToES(doc, context) {
         context.fail();
     });
 }
+
+/* Lambda "main": Execution starts here */
+exports.handler = function(event, context) {
+    console.log('Received event: ', JSON.stringify(event, null, 2));
+    
+    /* == Streams ==
+    * To avoid loading an entire (typically large) log file into memory,
+    * this is implemented as a pipeline of filters, streaming log data
+    * from S3 to ES.
+    * Flow: S3 file stream -> Log Line stream -> Log Record stream -> ES
+    */
+    var lineStream = new LineStream();
+    // A stream of log records, from parsing each log line
+    var recordStream = new stream.Transform({objectMode: true})
+    recordStream._transform = function(line, encoding, done) {
+        var logRecord = parse(line.toString());
+        var serializedRecord = JSON.stringify(logRecord);
+        this.push(serializedRecord);
+        totLogLines ++;
+        done();
+    }
+
+    event.Records.forEach(function(record) {
+        var bucket = record.s3.bucket.name;
+        var objKey = decodeURIComponent(record.s3.object.key.replace(/\+/g, ' '));
+        s3LogsToES(bucket, objKey, context, lineStream, recordStream);
+    });
+}