Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Event source mapping support #189

Merged
merged 1 commit into from
Mar 27, 2017
Merged

Event source mapping support #189

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions bin/node-lambda
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ dotenv.load();

var AWS_ENVIRONMENT = process.env.AWS_ENVIRONMENT || '';
var CONFIG_FILE = process.env.CONFIG_FILE || '';
var EVENT_SOURCE_FILE = process.env.EVENT_SOURCE_FILE || 'event_sources.json';
var EXCLUDE_GLOBS = process.env.EXCLUDE_GLOBS || '';
var AWS_ACCESS_KEY_ID = process.env.AWS_ACCESS_KEY_ID;
var AWS_SECRET_ACCESS_KEY = process.env.AWS_SECRET_ACCESS_KEY;
Expand Down Expand Up @@ -61,6 +62,8 @@ program
.option('-A, --packageDirectory [' + PACKAGE_DIRECTORY + ']', 'Local Package Directory', PACKAGE_DIRECTORY)
.option('-f, --configFile [' + CONFIG_FILE + ']',
'Path to file holding secret environment variables (e.g. "deploy.env")', CONFIG_FILE)
.option('-S, --eventSourceFile [' + EVENT_SOURCE_FILE + ']',
'Path to file holding event source mapping variables (e.g. "event_sources.json")', EVENT_SOURCE_FILE)
.option('-x, --excludeGlobs [' + EXCLUDE_GLOBS + ']',
'Space-separated glob pattern(s) for additional exclude files (e.g. "event.json dotenv.sample")', EXCLUDE_GLOBS)
.option('-D, --prebuiltDirectory [' + PREBUILT_DIRECTORY + ']', 'Prebuilt directory', PREBUILT_DIRECTORY)
Expand Down
8 changes: 8 additions & 0 deletions lib/event_sources.json.example
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
[
{
"EventSourceArn": "your event source arn",
"StartingPosition": "LATEST",
"BatchSize": 100,
"Enabled": true
}
]
157 changes: 151 additions & 6 deletions lib/main.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ Lambda.prototype.setup = function (program) {
this._createSampleFile(program.eventFile, 'event.json');
this._createSampleFile('deploy.env', 'deploy.env');
this._createSampleFile(program.contextFile, 'context.json');
this._createSampleFile('event_sources.json', 'event_sources.json');
console.log('Setup done. Edit the .env, deploy.env, ' + program.contextFile + ' and ' + program.eventFile +
' files as needed.');
};
Expand Down Expand Up @@ -146,10 +147,24 @@ Lambda.prototype._params = function (program, buffer) {
Variables: config
}
}

return params;
};

Lambda.prototype._eventSourceList = function (program) {
var eventSourceList = []
if (program.eventSourceFile) {
try {
eventSourceList = fs.readJsonSync(program.eventSourceFile);
} catch(err) {
eventSourceList = [];
throw err;
}
}

return eventSourceList;
};

/**
* @deprecated
*/
Expand Down Expand Up @@ -402,6 +417,93 @@ Lambda.prototype._buildAndArchive = function (program, archive_callback) {
});
};

Lambda.prototype._listEventSourceMappings = function (lambda, params, cb) {
return lambda.listEventSourceMappings(params, function (err, data) {
var eventSourceMappings = [];
if (!err && data && data.EventSourceMappings) {
eventSourceMappings = data.EventSourceMappings;
}
return cb(err, eventSourceMappings);
});
};

Lambda.prototype._updateEventSources = function (lambda, functionName, existingEventSourceList, eventSourceList, cb) {
var updateEventSourceList = [];
// Checking new and update event sources
for (var i in eventSourceList) {
var isExisting = false;
for (var j in existingEventSourceList) {
if (eventSourceList[i]['EventSourceArn'] === existingEventSourceList[j]['EventSourceArn']) {
isExisting = true;
updateEventSourceList.push({
'type': 'update',
'FunctionName': functionName,
'Enabled': eventSourceList[i]['Enabled'],
'BatchSize': eventSourceList[i]['BatchSize'],
'UUID': existingEventSourceList[j]['UUID']
});
break;
}
}

// If it is new source
if (!isExisting) {
updateEventSourceList.push({
'type': 'create',
'FunctionName': functionName,
'EventSourceArn': eventSourceList[i]['EventSourceArn'],
'Enabled': eventSourceList[i]['Enabled'] ? eventSourceList[i]['Enabled'] : false,
'BatchSize': eventSourceList[i]['BatchSize'] ? eventSourceList[i]['BatchSize'] : 100,
'StartingPosition': eventSourceList[i]['StartingPosition'] ? eventSourceList[i]['StartingPosition'] : 'LATEST',
});
}
}

// Checking delete event sources
for (var i in existingEventSourceList) {
var isExisting = false;
for (var j in eventSourceList) {
if (eventSourceList[j]['EventSourceArn'] === existingEventSourceList[i]['EventSourceArn']) {
isExisting = true;
break;
}
}

// If delete the source
if (!isExisting) {
updateEventSourceList.push({
'type': 'delete',
'UUID': existingEventSourceList[i]['UUID']
});
}
}

return async.map(updateEventSourceList, function (updateEventSource, _cb) {
switch(updateEventSource['type']) {
case 'create':
delete updateEventSource['type'];
lambda.createEventSourceMapping(updateEventSource, function (err, data) {
return _cb(err, data);
});
break;
case 'update':
delete updateEventSource['type'];
lambda.updateEventSourceMapping(updateEventSource, function (err, data) {
return _cb(err, data);
});
break;
case 'delete':
delete updateEventSource['type'];
lambda.deleteEventSourceMapping(updateEventSource, function (err, data) {
return _cb(err, data);
});
break;
}
}, function(err, results) {
return cb(err, results);
});
};

Lambda.prototype.package = function (program) {
var _this = this;
if (!program.packageDirectory) {
Expand Down Expand Up @@ -451,6 +553,9 @@ Lambda.prototype.deploy = function (program) {
console.log('=> Reading zip file to memory');
var params = _this._params(program, buffer);

console.log('=> Reading event source file to memory');
var eventSourceList = _this._eventSourceList(program);

async.map(regions, function (region, cb) {
console.log('=> Uploading zip file to AWS Lambda ' + region + ' with parameters:');
console.log(params);
Expand Down Expand Up @@ -478,20 +583,60 @@ Lambda.prototype.deploy = function (program) {
apiVersion: '2015-03-31'
});

// Checking function
return lambda.getFunction({
'FunctionName': params.FunctionName
}, function (err) {
if(err) {
return _this._uploadNew(lambda, params, cb);
if (err) {
return _this._uploadNew(lambda, params, function(err, results) {
if (err) {
throw err;
} else {
console.log('=> Zip file(s) done uploading. Results follow: ');
console.log(results);

// Updating event source(s)
_this._updateEventSources(lambda, params.FunctionName, [], eventSourceList, function(err, results) {
cb(null, results);
});
}
});
} else {
_this._listEventSourceMappings(lambda, {
'FunctionName': params.FunctionName
}, function(err, existingEventSourceList) {
if (err) {
throw err;
} else {
return async.parallel([
function(_callback) {
_this._uploadExisting(lambda, params, function(err, results) {
if (err) {
throw err;
} else {
console.log('=> Zip file(s) done uploading. Results follow: ');
console.log(results);
_callback(err, results);
}
})
},
function(_callback) {
_this._updateEventSources(lambda, params.FunctionName, existingEventSourceList, eventSourceList, function(err, results) {
_callback(err, results)
})
}
], function(err, results) {
cb(err, results);
});
}
});
}

return _this._uploadExisting(lambda, params, cb);
});
}, function (err, results) {
if (err) {
throw err;
} else {
console.log('=> Zip file(s) done uploading. Results follow: ');
console.log('=> All tasks done. Results follow: ');
console.log(results);
}
});
Expand Down