diff --git a/.gitignore b/.gitignore
index 39bb3a5..e643d4d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,6 @@
+.gradle
 launchConfigurations/
 *.pyc
+action/*.zip
+action/package.json
+tests/build
diff --git a/action/kafkaFeed.js b/action/kafkaFeed.js
index fc785c0..3cd42f7 100644
--- a/action/kafkaFeed.js
+++ b/action/kafkaFeed.js
@@ -1,143 +1,30 @@
-var request = require('request');
+const common = require('./lib/common');
 
 /**
  *   Feed to listen to Kafka messages
  *  @param {string} brokers - array of Kafka brokers
  *  @param {string} topic - topic to subscribe to
  *  @param {bool}   isJSONData - attempt to parse messages as JSON
- *  @param {bool}   isBinaryKey - encode message as Base64
- *  @param {bool}   isBinaryValue - encode key as Base64
- *  @param {string} endpoint - address to OpenWhisk deployment
+ *  @param {bool}   isBinaryKey - encode key as Base64
+ *  @param {bool}   isBinaryValue - encode message as Base64
+ *  @param {string} endpoint - address to OpenWhisk deployment (expected to be bound at deployment)
  */
 function main(params) {
-    var promise = new Promise(function(resolve, reject) {
-        if(!params.package_endpoint) {
-            reject('Could not find the package_endpoint parameter.');
-            return;
-        }
+    const endpoint = params.endpoint;
+    const webActionName = 'kafkaFeedWeb';
 
-        var triggerComponents = params.triggerName.split("/");
-        var namespace = encodeURIComponent(process.env['__OW_NAMESPACE']);
-        var trigger = encodeURIComponent(triggerComponents[2]);
-        var feedServiceURL = 'http://' + params.package_endpoint + '/triggers/' + namespace + '/' + trigger;
+    var massagedParams = common.massageParamsForWeb(params);
+    massagedParams.triggerName = common.getTriggerFQN(params.triggerName);
 
-        if (params.lifecycleEvent === 'CREATE') {
-            // makes a PUT request to create the trigger in the feed service provider
-            var putTrigger = function(validatedParams) {
-                var body = validatedParams;
+    if (params.lifecycleEvent === 'CREATE') {
+        return common.createTrigger(endpoint, massagedParams, webActionName);
+    } else if (params.lifecycleEvent === 'DELETE') {
+        return common.deleteTrigger(endpoint, massagedParams, webActionName);
+    }
 
-                // params.endpoint may already include the protocol - if so,
-                // strip it out
-                var massagedAPIHost = params.endpoint.replace(/https?:\/\/(.*)/, "$1");
-                body.triggerURL = 'https://' + process.env['__OW_API_KEY'] + "@" + massagedAPIHost + '/api/v1/namespaces/' + namespace + '/triggers/' + trigger;
-
-                var options = {
-                    method: 'PUT',
-                    url: feedServiceURL,
-                    body: JSON.stringify(body),
-                    headers: {
-                        'Content-Type': 'application/json',
-                        'User-Agent': 'whisk'
-                    }
-                };
-
-                return doRequest(options);
-            };
-
-            validateParameters(params)
-                .then(putTrigger)
-                .then(resolve)
-                .catch(reject);
-        } else if (params.lifecycleEvent === 'DELETE') {
-            var authorizationHeader = 'Basic ' + new Buffer(process.env['__OW_API_KEY']).toString('base64');
-
-            var options = {
-                method: 'DELETE',
-                url: feedServiceURL,
-                headers: {
-                    'Content-Type': 'application/json',
-                    'Authorization': authorizationHeader,
-                    'User-Agent': 'whisk'
-                }
-            };
-
-            doRequest(options)
-                .then(resolve)
-                .catch(reject);
-        }
-    });
-
-    return promise;
+    return {
+        error: 'unsupported lifecycleEvent'
+    };
 }
 
-function doRequest(options) {
-    var requestPromise = new Promise(function (resolve, reject) {
-        request(options, function (error, response, body) {
-            if (error) {
-                reject({
-                    response: response,
-                    error: error,
-                    body: JSON.parse(body)
-                });
-            } else {
-                console.log("Status code: " + response.statusCode);
-
-                if (response.statusCode >= 400) {
-                    console.log("Response from Kafka feed service: " + body);
-                    reject({
-                        statusCode: response.statusCode,
-                        response: JSON.parse(body)
-                    });
-                } else {
-                    resolve({
-                        response: JSON.parse(body)
-                    });
-                }
-            }
-        });
-    });
-
-    return requestPromise;
-}
-
-function validateParameters(rawParams) {
-    var promise = new Promise(function(resolve, reject) {
-        var validatedParams = {};
-
-        validatedParams.isMessageHub = false;
-        validatedParams.isJSONData = getBooleanFromArgs(rawParams, 'isJSONData');
-        validatedParams.isBinaryValue = getBooleanFromArgs(rawParams, 'isBinaryValue');
-        validatedParams.isBinaryKey = getBooleanFromArgs(rawParams, 'isBinaryKey');
-
-        if (validatedParams.isJSONData && validatedParams.isBinaryValue) {
-            reject('isJSONData and isBinaryValue cannot both be enabled.');
-            return;
-        }
-
-        if (rawParams.topic && rawParams.topic.length > 0) {
-            validatedParams.topic = rawParams.topic;
-        } else {
-            reject('You must supply a "topic" parameter.');
-            return;
-        }
-
-        if (isNonEmptyArray(rawParams.brokers)) {
-            validatedParams.brokers = rawParams.brokers;
-        } else {
-            reject('You must supply a "brokers" parameter as an array of Kafka brokers.');
-            return;
-        }
-
-        resolve(validatedParams);
-    });
-
-    return promise;
-}
-
-function getBooleanFromArgs(args, key) {
-  return (typeof args[key] !== 'undefined' && args[key] && (args[key] === true || args[key].toString().trim().toLowerCase() === 'true'));
-}
-
-function isNonEmptyArray(obj) {
-    return obj && Array.isArray(obj) && obj.length !== 0;
-}
+exports.main = main;
diff --git a/action/kafkaFeedWeb.js b/action/kafkaFeedWeb.js
new file mode 100644
index 0000000..6975677
--- /dev/null
+++ b/action/kafkaFeedWeb.js
@@ -0,0 +1,115 @@
+const common = require('./lib/common');
+const Database = require('./lib/Database');
+
+/**
+ *   Feed to listen to Kafka messages
+ *  @param {string} brokers - array of Kafka brokers
+ *  @param {string} topic - topic to subscribe to
+ *  @param {bool}   isJSONData - attempt to parse messages as JSON
+ *  @param {bool}   isBinaryKey - encode key as Base64
+ *  @param {bool}   isBinaryValue - encode message as Base64
+ *  @param {string} endpoint - address to OpenWhisk deployment (expected to be bound at deployment)
+ *  @param {string} DB_URL - URL for the DB, must include authentication (expected to be bound at deployment)
+ *  @param {string} DB_NAME - DB name (expected to be bound at deployment)
+ */
+function main(params) {
+    var promise = new Promise((resolve, reject) => {
+        // hold off initializing this until definitely needed
+        var db;
+
+        if (params.__ow_method === "put") {
+            return validateParameters(params)
+                .then(validatedParams => {
+                    console.log(`VALIDATED: ${JSON.stringify(validatedParams, null, 2)}`);
+                    db = new Database(params.DB_URL, params.DB_NAME);
+
+                    var promises = [];
+
+                    // do these in parallel!
+                    promises.push(db.ensureTriggerIsUnique(validatedParams.triggerName));
+                    promises.push(common.verifyTriggerAuth(validatedParams.triggerURL));
+
+                    return Promise.all(promises)
+                        .then(result => validatedParams);
+                })
+                .then(validatedParams => db.recordTrigger(validatedParams))
+                .then(result => {
+                    console.log('successfully wrote the trigger');
+                    resolve();
+                })
+                .catch(error => {
+                    console.log(`Failed to write the trigger ${error}`);
+
+                    // defaults to potentially be overridden
+                    var statusCode = 500;
+                    var body = error.toString();
+
+                    if(error.validationError) {
+                        statusCode = 400;
+                        body = error.validationError;
+                    } else if(error.authError) {
+                        statusCode = 401;
+                        body = error.authError;
+                    }
+
+                    resolve({
+                        statusCode: statusCode,
+                        headers: {'Content-Type': 'text/plain'},
+                        body: body
+                    });
+                });
+        } else if (params.__ow_method === "delete") {
+            const triggerURL = common.getTriggerURL(params.authKey, params.endpoint, params.triggerName);
+
+            return common.verifyTriggerAuth(triggerURL)
+                .then(() => {
+                    db = new Database(params.DB_URL, params.DB_NAME);
+                    return db.deleteTrigger(params.triggerName);
+                })
+                .then(resolve)
+                .catch(reject);
+        } else {
+            resolve({
+                statusCode: 400,
+                headers: {'Content-Type': 'text/plain'},
+                body: 'unsupported lifecycleEvent'
+            });
+        }
+    });
+
+    return promise;
+}
+
+function validateParameters(rawParams) {
+    var promise = new Promise((resolve, reject) => {
+        var validatedParams;
+
+        var commonValidationResult = common.performCommonParameterValidation(rawParams);
+        if(commonValidationResult.validationError) {
+            reject(commonValidationResult);
+            return;
+        } else {
+            validatedParams = commonValidationResult.validatedParams;
+        }
+
+        // brokers
+        if (rawParams.brokers) {
+            validatedParams.brokers = common.validateBrokerParam(rawParams.brokers);
+            if (!validatedParams.brokers) {
+                reject( { validationError: "You must supply a 'brokers' parameter as an array of Message Hub brokers." });
+                return;
+            }
+        } else {
+            reject( { validationError: "You must supply a 'brokers' parameter." });
+            return;
+        }
+
+        validatedParams.isMessageHub = false;
+
+        resolve(validatedParams);
+    });
+
+    return promise;
+}
+
+exports.main = main;
diff --git a/action/kafkaFeedWeb_package.json b/action/kafkaFeedWeb_package.json
new file mode 100644
index 0000000..78b5c57
--- /dev/null
+++ b/action/kafkaFeedWeb_package.json
@@ -0,0 +1,5 @@
+{
+  "name": "kafkaFeedWeb",
+  "version": "1.0.0",
+  "main": "kafkaFeedWeb.js"
+}
diff --git a/action/kafkaFeed_package.json b/action/kafkaFeed_package.json
new file mode 100644
index 0000000..02a1614
--- /dev/null
+++ b/action/kafkaFeed_package.json
@@ -0,0 +1,5 @@
+{
+  "name": "kafkaFeed",
+  "version": "1.0.0",
+  "main": "kafkaFeed.js"
+}
diff --git a/action/lib/Database.js b/action/lib/Database.js
new file mode 100644
index 0000000..fe8f778
--- /dev/null
+++ b/action/lib/Database.js
@@ -0,0 +1,63 @@
+// constructor for DB object - a thin, promise-loving wrapper around nano
+module.exports = function(dbURL, dbName) {
+    var nano = require('nano')(dbURL);
+    this.db = nano.db.use(dbName);
+
+    this.getTrigger = function(triggerFQN) {
+        return new Promise((resolve, reject) => {
+            this.db.get(triggerFQN, (err, result) => {
+                if(err) {
+                    reject(err);
+                } else {
+                    resolve(result);
+                }
+            });
+        });
+    };
+
+    this.ensureTriggerIsUnique = function(triggerFQN) {
+        return this.getTrigger(this.db, triggerFQN)
+            .then(result => {
+                return Promise.reject('Trigger already exists');
+            })
+            .catch(err => {
+                // turn that frown upside-down!
+                return true;
+            });
+    };
+
+    this.recordTrigger = function(params) {
+        console.log('recording trigger');
+
+        params['_id'] = params.triggerName;
+        params['status'] = {
+            'active': true,
+            'dateChanged': Math.round(new Date().getTime() / 1000)
+        };
+
+        return new Promise((resolve, reject) => {
+            this.db.insert(params, (err, result) => {
+                if(err) {
+                    reject(err);
+                } else {
+                    resolve(result);
+                }
+            });
+        });
+    };
+
+    this.deleteTrigger = function(triggerFQN) {
+        return this.getTrigger(triggerFQN)
+            .then(doc => {
+                return new Promise((resolve, reject) => {
+                    this.db.destroy(doc._id, doc._rev, (err, result) => {
+                        if(err) {
+                            reject(err);
+                        } else {
+                            resolve(result);
+                        }
+                    });
+                });
+            })
+    };
+};
diff --git a/action/lib/common.js b/action/lib/common.js
new file mode 100644
index 0000000..def07e4
--- /dev/null
+++ b/action/lib/common.js
@@ -0,0 +1,177 @@
+const request = require('request-promise');
+
+function triggerComponents(triggerName) {
+    var split = triggerName.split("/");
+
+    return {
+        namespace: split[1],
+        triggerName: split[2]
+    };
+}
+
+function getTriggerURL(authKey, endpoint, triggerName) {
+    var massagedAPIHost = endpoint.replace(/https?:\/\/(.*)/, "$1");
+
+    var components = triggerComponents(triggerName);
+    var namespace = components.namespace;
+    var trigger = components.triggerName;
+
+    var url = `https://${authKey}@${massagedAPIHost}/api/v1/namespaces/${encodeURIComponent(namespace)}/triggers/${encodeURIComponent(trigger)}`;
+
+    return url;
+}
+
+function verifyTriggerAuth(triggerURL) {
+    var options = {
+        method: 'GET',
+        url: triggerURL,
+        rejectUnauthorized: false,
+        headers: {
+            'Content-Type': 'application/json',
+            'User-Agent': 'whisk'
+        }
+    };
+
+    return request(options)
+        .catch(err => {
+            console.log(`Trigger auth error: ${JSON.stringify(err)}`);
+            return Promise.reject({ authError: 'You are not authorized for this trigger.'});
+        });
+}
+
+function validateBrokerParam(brokerParam) {
+    if (isNonEmptyArray(brokerParam)) {
+        return brokerParam;
+    } else if (typeof brokerParam === 'string') {
+        return brokerParam.split(',');
+    } else {
+        return undefined;
+    }
+}
+
+function getBooleanFromArgs(args, key) {
+    return (typeof args[key] !== 'undefined' && args[key] && (args[key] === true || args[key].toString().trim().toLowerCase() === 'true'));
+}
+
+function isNonEmptyArray(obj) {
+    return obj && Array.isArray(obj) && obj.length !== 0;
+}
+
+// Return the trigger FQN with the resolved namespace
+// This is required to avoid naming conflicts when using the default namespace "_"
+function getTriggerFQN(triggerName) {
+    var components = triggerName.split('/');
+    return `/${process.env['__OW_NAMESPACE']}/${components[2]}`;
+}
+
+function massageParamsForWeb(rawParams) {
+    var massagedParams = Object.assign({}, rawParams);
+
+    // remove these parameters as they may conflict with bound parameters of the web action
+    delete massagedParams.endpoint;
+    delete massagedParams.bluemixServiceName;
+    delete massagedParams.lifecycleEvent;
+
+    return massagedParams;
+}
+
+function getWebActionURL(endpoint, actionName) {
+    return `https://${endpoint}/api/v1/web/whisk.system/messagingWeb/${actionName}.http`;
+}
+
+function createTrigger(endpoint, params, actionName) {
+    var options = {
+        method: 'PUT',
+        url: getWebActionURL(endpoint, actionName),
+        rejectUnauthorized: false,
+        json: true,
+        body: params,
+        headers: {
+            'Content-Type': 'application/json',
+            'Accept': 'text/plain',
+            'User-Agent': 'whisk'
+        }
+    };
+
+    return request(options)
+        .then(response => {
+            // do not pass the response back to the caller, its contents are secret
+            return;
+        })
+        .catch(error => {
+            console.log(`Error creating trigger: ${JSON.stringify(error, null, 2)}`);
+            return Promise.reject(error.response.body);
+        });
+}
+
+function deleteTrigger(endpoint, params, actionName) {
+    var options = {
+        method: 'DELETE',
+        url: getWebActionURL(endpoint, actionName),
+        rejectUnauthorized: false,
+        json: true,
+        body: params,
+        headers: {
+            'Content-Type': 'application/json',
+            'Accept': 'text/plain',
+            'User-Agent': 'whisk'
+        }
+    };
+
+    return request(options)
+        .then(response => {
+            // do not pass the response back to the caller, its contents are secret
+            return;
+        }).catch(error => {
+            console.log(`Error deleting trigger: ${JSON.stringify(error, null, 2)}`);
+            return Promise.reject(error.response.body);
+        });
+}
+
+// perform parameter validation that is common to both feed actions
+function performCommonParameterValidation(rawParams) {
+    var validatedParams = {};
+
+    // topic
+    if (rawParams.topic && rawParams.topic.length > 0) {
+        validatedParams.topic = rawParams.topic;
+    } else {
+        return { validationError: "You must supply a 'topic' parameter." };
+    }
+
+    // triggerName
+    if (rawParams.triggerName) {
+        validatedParams.triggerName = rawParams.triggerName;
+    } else {
+        return { validationError: "You must supply a 'triggerName' parameter." };
+    }
+
+    validatedParams.isJSONData = getBooleanFromArgs(rawParams, 'isJSONData');
+    validatedParams.isBinaryValue = getBooleanFromArgs(rawParams, 'isBinaryValue');
+
+    if (validatedParams.isJSONData && validatedParams.isBinaryValue) {
+        return { validationError: 'isJSONData and isBinaryValue cannot both be enabled.' };
+    }
+
+    // now that everything else is valid, let's add these
+    validatedParams.isBinaryKey = getBooleanFromArgs(rawParams, 'isBinaryKey');
+    validatedParams.authKey = rawParams.authKey;
+    validatedParams.triggerURL = getTriggerURL(validatedParams.authKey, rawParams.endpoint, rawParams.triggerName);
+
+    const uuid = require('uuid');
+    validatedParams.uuid = uuid.v4();
+
+    return { validatedParams: validatedParams };
+}
+
+module.exports = {
+    'createTrigger': createTrigger,
+    'deleteTrigger': deleteTrigger,
+    'getBooleanFromArgs': getBooleanFromArgs,
+    'getTriggerFQN': getTriggerFQN,
+    'getTriggerURL': getTriggerURL,
+    'massageParamsForWeb': massageParamsForWeb,
+    'performCommonParameterValidation': performCommonParameterValidation,
+    'validateBrokerParam': validateBrokerParam,
+    'verifyTriggerAuth': verifyTriggerAuth,
+};
diff --git a/action/messageHubFeed.js b/action/messageHubFeed.js
index ed521d3..818520a 100644
--- a/action/messageHubFeed.js
+++ b/action/messageHubFeed.js
@@ -1,4 +1,4 @@
-var request = require('request');
+const common = require('./lib/common');
 
 /**
  *   Feed to listen to MessageHub messages
@@ -9,213 +9,24 @@
  *  @param {bool}   isJSONData - attempt to parse messages as JSON
  *  @param {bool}   isBinaryKey - encode key as Base64
  *  @param {bool}   isBinaryValue - encode message as Base64
- *  @param {string} endpoint - address to OpenWhisk deployment
+ *  @param {string} endpoint - address to OpenWhisk deployment (expected to be bound at deployment)
  */
 function main(params) {
-    var promise = new Promise(function(resolve, reject) {
-        if(!params.package_endpoint) {
-            reject('Could not find the package_endpoint parameter.');
-            return;
-        }
+    const endpoint = params.endpoint;
+    const webActionName = 'messageHubFeedWeb'
 
-        var triggerComponents = params.triggerName.split("/");
-        var namespace = encodeURIComponent(process.env['__OW_NAMESPACE']);
-        var trigger = encodeURIComponent(triggerComponents[2]);
-        var feedServiceURL = 'http://' + params.package_endpoint + '/triggers/' + namespace + '/' + trigger;
+    var massagedParams = common.massageParamsForWeb(params);
+    massagedParams.triggerName = common.getTriggerFQN(params.triggerName);
 
-        if (params.lifecycleEvent === 'CREATE') {
-            // makes a PUT request to create the trigger in the feed service provider
-            var putTrigger = function(validatedParams) {
-                var body = validatedParams;
-
-                // params.endpoint may already include the protocol - if so,
-                // strip it out
-                var massagedAPIHost = params.endpoint.replace(/https?:\/\/(.*)/, "$1");
-                body.triggerURL = 'https://' + process.env['__OW_API_KEY'] + "@" + massagedAPIHost + '/api/v1/namespaces/' + namespace + '/triggers/' + trigger;
-
-                var options = {
-                    method: 'PUT',
-                    url: feedServiceURL,
-                    body: JSON.stringify(body),
-                    headers: {
-                        'Content-Type': 'application/json',
-                        'User-Agent': 'whisk'
-                    }
-                };
-
-                return doRequest(options);
-            };
-
-            return validateParameters(params)
-                .then(checkMessageHubCredentials)
-                .then(putTrigger)
-                .then(resolve)
-                .catch(reject);
-        } else if (params.lifecycleEvent === 'DELETE') {
-            var authorizationHeader = 'Basic ' + new Buffer(process.env['__OW_API_KEY']).toString('base64');
-
-            var options = {
-                method: 'DELETE',
-                url: feedServiceURL,
-                headers: {
-                    'Content-Type': 'application/json',
-                    'Authorization': authorizationHeader,
-                    'User-Agent': 'whisk'
-                }
-            };
-
-            doRequest(options)
-                .then(resolve)
-                .catch(reject);
-        }
-    });
-
-    return promise;
-}
-
-function checkMessageHubCredentials(params) {
-    // make sure we have valid MH credentials
-    console.log("Checking Message Hub credentials...");
-
-    // listing topics seems to be the simplest way to check auth
-    var topicURL = params.kafka_admin_url + '/admin/topics';
-
-    var options = {
-        method: 'GET',
-        url: topicURL,
-        headers: {
-            'X-Auth-Token': params.username + params.password
-        }
-    };
-
-    return doRequest(options)
-        .then(function(body) {
-            console.log("Successfully authenticated with Message Hub");
-
-            var topicNames = body.response.map(function(topic) {
-                return topic.name
-            });
-
-            if(topicNames.indexOf(params.topic) < 0) {
-                return Promise.reject('Topic does not exist. You must create the topic first: ' + params.topic);
-            } else {
-                console.log("Topic exists.");
-                // return the params so they can be chained into the next function
-                return params;
-            }
-        }, function(authError) {
-            return Promise.reject('Could not authenticate with Message Hub. Please check your credentials.');
-        });
-}
-
-function doRequest(options) {
-    var requestPromise = new Promise(function (resolve, reject) {
-        request(options, function (error, response, body) {
-            if (error) {
-                reject({
-                    response: response,
-                    error: error,
-                    body: JSON.parse(body)
-                });
-            } else {
-                console.log("Status code: " + response.statusCode);
-
-                if (response.statusCode >= 400) {
-                    console.log("Response: " + JSON.stringify(body));
-                    reject({
-                        statusCode: response.statusCode,
-                        response: JSON.parse(body)
-                    });
-                } else {
-                    resolve({
-                        response: JSON.parse(body)
-                    });
-                }
-            }
-        });
-    });
-
-    return requestPromise;
-}
-
-function validateParameters(rawParams) {
-    var promise = new Promise(function(resolve, reject) {
-        var validatedParams = {};
-
-        validatedParams.isMessageHub = true;
-        validatedParams.isJSONData = getBooleanFromArgs(rawParams, 'isJSONData');
-        validatedParams.isBinaryValue = getBooleanFromArgs(rawParams, 'isBinaryValue');
-        validatedParams.isBinaryKey = getBooleanFromArgs(rawParams, 'isBinaryKey');
-
-        if (validatedParams.isJSONData && validatedParams.isBinaryValue) {
-            reject('isJSONData and isBinaryValue cannot both be enabled.');
-            return;
-        }
-
-        // topic
-        if (rawParams.topic && rawParams.topic.length > 0) {
-            validatedParams.topic = rawParams.topic;
-        } else {
-            reject('You must supply a "topic" parameter.');
-            return;
-        }
-
-        // kafka_brokers_sasl
-        if (rawParams.kafka_brokers_sasl) {
-            validatedParams.brokers = validateBrokerParam(rawParams.kafka_brokers_sasl);
-            if(!validatedParams.brokers) {
-                reject('You must supply a "kafka_brokers_sasl" parameter as an array of Message Hub brokers.');
-                return;
-            }
-        } else {
-            reject('You must supply a "kafka_brokers_sasl" parameter as an array of Message Hub brokers.');
-            return;
-        }
-
-        // user
-        if (rawParams.user) {
-            validatedParams.username = rawParams.user;
-        } else {
-            reject('You must supply a "user" parameter to authenticate with Message Hub.');
-            return;
-        }
-
-        // password
-        if (rawParams.password) {
-            validatedParams.password = rawParams.password;
-        } else {
-            reject('You must supply a "password" parameter to authenticate with Message Hub.');
-            return;
-        }
-
-        // kafka_admin_url
-        if(rawParams.kafka_admin_url) {
-            validatedParams.kafka_admin_url = rawParams.kafka_admin_url;
-        } else {
-            reject('You must supply a "kafka_admin_url" parameter.');
-            return;
-        }
-
-        resolve(validatedParams);
-    });
-
-    return promise;
-}
-
-function validateBrokerParam(brokerParam) {
-    if(isNonEmptyArray(brokerParam)) {
-        return brokerParam;
-    } else if (typeof brokerParam === 'string') {
-        return brokerParam.split(',');
-    } else {
-        return undefined;
+    if (params.lifecycleEvent === 'CREATE') {
+        return common.createTrigger(endpoint, massagedParams, webActionName);
+    } else if (params.lifecycleEvent === 'DELETE') {
+        return common.deleteTrigger(endpoint, massagedParams, webActionName);
     }
+
+    return {
+        error: 'unsupported lifecycleEvent'
+    };
 }
 
-function getBooleanFromArgs(args, key) {
-  return (typeof args[key] !== 'undefined' && args[key] && (args[key] === true || args[key].toString().trim().toLowerCase() === 'true'));
-}
-
-function isNonEmptyArray(obj) {
-    return obj && Array.isArray(obj) && obj.length !== 0;
-}
+exports.main = main;
diff --git a/action/messageHubFeedWeb.js b/action/messageHubFeedWeb.js
new file mode 100644
index 0000000..2a7b4e3
--- /dev/null
+++ b/action/messageHubFeedWeb.js
@@ -0,0 +1,174 @@
+const common = require('./lib/common');
+const Database = require('./lib/Database');
+
+/**
+ *   Feed to listen to MessageHub messages
+ *  @param {string} brokers - array of Message Hub brokers
+ *  @param {string} username - Kafka username
+ *  @param {string} password - Kafka password
+ *  @param {string} topic - topic to subscribe to
+ *  @param {bool}   isJSONData - attempt to parse messages as JSON
+ *  @param {bool}   isBinaryKey - encode key as Base64
+ *  @param {bool}   isBinaryValue - encode message as Base64
+ *  @param {string} endpoint - address to OpenWhisk deployment (expected to be bound at deployment)
+ *  @param {string} DB_URL - URL for the DB, must include authentication (expected to be bound at deployment)
+ *  @param {string} DB_NAME - DB name (expected to be bound at deployment)
+ */
+function main(params) {
+    var promise = new Promise((resolve, reject) => {
+        // hold off initializing this until definitely needed
+        var db;
+
+        if (params.__ow_method === "put") {
+            return validateParameters(params)
+                .then(validatedParams => {
+                    console.log(`VALIDATED: ${JSON.stringify(validatedParams, null, 2)}`);
+                    db = new Database(params.DB_URL, params.DB_NAME);
+
+                    var promises = [];
+
+                    // do these in parallel!
+                    promises.push(db.ensureTriggerIsUnique(validatedParams.triggerName));
+                    promises.push(common.verifyTriggerAuth(validatedParams.triggerURL));
+                    promises.push(checkMessageHubCredentials(validatedParams));
+
+                    return Promise.all(promises)
+                        .then(result => validatedParams);
+                })
+                .then(validatedParams => db.recordTrigger(validatedParams))
+                .then(result => {
+                    console.log('successfully wrote the trigger');
+                    resolve();
+                })
+                .catch(error => {
+                    console.log(`Failed to write the trigger ${error}`);
+
+                    // defaults to potentially be overridden
+                    var statusCode = 500;
+                    var body = error.toString();
+
+                    if(error.validationError) {
+                        statusCode = 400;
+                        body = error.validationError;
+                    } else if(error.authError) {
+                        statusCode = 401;
+                        body = error.authError;
+                    }
+
+                    resolve({
+                        statusCode: statusCode,
+                        headers: {'Content-Type': 'text/plain'},
+                        body: body
+                    });
+                });
+        } else if (params.__ow_method === "delete") {
+            const triggerURL = common.getTriggerURL(params.authKey, params.endpoint, params.triggerName);
+
+            return common.verifyTriggerAuth(triggerURL)
+                .then(() => {
+                    db = new Database(params.DB_URL, params.DB_NAME);
+                    return db.deleteTrigger(params.triggerName);
+                })
+                .then(resolve)
+                .catch(reject);
+        } else {
+            resolve({
+                statusCode: 400,
+                headers: {'Content-Type': 'text/plain'},
+                body: 'unsupported lifecycleEvent'
+            });
+        }
+    });
+
+    return promise;
+}
+
+function validateParameters(rawParams) {
+    var promise = new Promise((resolve, reject) => {
+        var validatedParams;
+
+        var commonValidationResult = common.performCommonParameterValidation(rawParams);
+        if(commonValidationResult.validationError) {
+            reject(commonValidationResult);
+            return;
+        } else {
+            validatedParams = commonValidationResult.validatedParams;
+        }
+
+        // kafka_brokers_sasl
+        if (rawParams.kafka_brokers_sasl) {
+            validatedParams.brokers = common.validateBrokerParam(rawParams.kafka_brokers_sasl);
+            if (!validatedParams.brokers) {
+                reject( { validationError: "You must supply a 'kafka_brokers_sasl' parameter as an array of Message Hub brokers." });
+                return;
+            }
+        } else {
+            reject( { validationError: "You must supply a 'kafka_brokers_sasl' parameter." });
+            return;
+        }
+
+        // user
+        if (rawParams.user) {
+            validatedParams.username = rawParams.user;
+        } else {
+            reject( { validationError: "You must supply a 'user' parameter to authenticate with Message Hub." });
+            return;
+        }
+
+        // password
+        if (rawParams.password) {
+            validatedParams.password = rawParams.password;
+        } else {
+            reject( { validationError: "You must supply a 'password' parameter to authenticate with Message Hub." });
+            return;
+        }
+
+        // kafka_admin_url
+        if (rawParams.kafka_admin_url) {
+            validatedParams.kafka_admin_url = rawParams.kafka_admin_url;
+        } else {
+            reject( { validationError: "You must supply a 'kafka_admin_url' parameter." });
+            return;
+        }
+
+        validatedParams.isMessageHub = true;
+
+        resolve(validatedParams);
+    });
+
+    return promise;
+}
+
+function checkMessageHubCredentials(params) {
+    // listing topics seems to be the simplest way to check auth
+    var topicURL = params.kafka_admin_url + '/admin/topics';
+
+    var options = {
+        method: 'GET',
+        url: topicURL,
+        json: true,
+        headers: {
+            'X-Auth-Token': params.username + params.password
+        }
+    };
+
+    const request = require('request-promise');
+
+    return request(options)
+        .then(body => {
+            console.log("Successfully authenticated with Message Hub");
+
+            var topicNames = body.map(topic => {
+                return topic.name
+            });
+
+            if (topicNames.indexOf(params.topic) < 0) {
+                return Promise.reject( 'Topic does not exist. You must create the topic first: ' + params.topic );
+            }
+        }, function (authError) {
+            console.log(`authError: ${JSON.stringify(authError)}`);
+            return Promise.reject( 'Could not authenticate with Message Hub. Please check your credentials.' );
+        });
+}
+
+exports.main = main;
diff --git a/action/messageHubFeedWeb_package.json b/action/messageHubFeedWeb_package.json
new file mode 100644
index 0000000..7735d19
--- /dev/null
+++ b/action/messageHubFeedWeb_package.json
@@ -0,0 +1,5 @@
+{
+  "name": "messageHubFeedWeb",
+  "version": "1.0.0",
+  "main": "messageHubFeedWeb.js"
+}
diff --git a/action/messageHubFeed_package.json b/action/messageHubFeed_package.json
new file mode 100644
index 0000000..a86035a
--- /dev/null
+++ b/action/messageHubFeed_package.json
@@ -0,0 +1,5 @@
+{
+  "name": "messageHubFeed",
+  "version": "1.0.0",
+  "main": "messageHubFeed.js"
+}
diff --git a/installCatalog.sh b/installCatalog.sh
index d265e13..9a459d3 100755
--- a/installCatalog.sh
+++ b/installCatalog.sh
@@ -4,8 +4,8 @@
 # automatically
 #
 # To run this command
-# ./installCatalog.sh  <AUTH> <EDGEHOST> <KAFKA_TRIGGER_HOST> <KAFKA_TRIGGER_PORT> <APIHOST>
-# AUTH and APIHOST are found in $HOME/.wskprops
+# ./installCatalog.sh <authkey> <edgehost> <dburl> <dbprefix> <apihost>
+# authkey and apihost are found in $HOME/.wskprops
 
 set -e
 set -x
@@ -15,16 +15,15 @@
 
 if [ $# -eq 0 ]
 then
-echo "Usage: ./installCatalog.sh <authkey> <apihost> <kafkatriggerhost> <kafkatriggerport>"
+echo "Usage: ./installCatalog.sh <authkey> <edgehost> <dburl> <dbprefix> <apihost>"
 fi
 
 AUTH="$1"
 EDGEHOST="$2"
-KAFKA_TRIGGER_HOST="$3"
-KAFKA_TRIGGER_PORT="$4"
+DB_URL="$3"
+DB_NAME="${4}ow_kafka_triggers"
 APIHOST="$5"
 
-
 # If the auth key file exists, read the key in the file. Otherwise, take the
 # first argument as the key itself.
 if [ -f "$AUTH" ]; then
@@ -34,29 +33,65 @@
 # Make sure that the APIHOST is not empty.
 : ${APIHOST:?"APIHOST must be set and non-empty"}
 
-KAFKA_PROVIDER_ENDPOINT=$KAFKA_TRIGGER_HOST':'$KAFKA_TRIGGER_PORT
-
 PACKAGE_HOME="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 
 export WSK_CONFIG_FILE= # override local property file to avoid namespace clashes
 
-echo Installing the kafka package and feed action.
+echo Installing the Message Hub package and feed action.
 
 $WSK_CLI -i --apihost "$EDGEHOST" package update messaging \
     --auth "$AUTH" \
     --shared yes \
     -a parameters '[ {"name":"kafka_brokers_sasl", "required":true, "description": "Array of Message Hub brokers", "bindTime":true},{"name":"user", "required":true, "description": "Message Hub username", "bindTime":true},{"name":"password", "required":true, "description": "Message Hub password", "bindTime":true, "type":"password"},{"name":"topic", "required":true, "description": "Topic to subscribe to"},{"name":"isJSONData", "required":false, "description": "Attempt to parse message value as JSON"},{"name":"isBinaryKey", "required":false, "description": "Encode key as Base64"},{"name":"isBinaryValue", "required":false, "description": "Encode message value as Base64"},{"name":"endpoint", "required":true, "description": "Hostname and port of OpenWhisk deployment"},{"name":"kafka_admin_url", "required":true, "description": "Your Message Hub admin REST URL", "bindTime":true}]' \
     -p bluemixServiceName 'messagehub' \
-    -p endpoint "$APIHOST" \
-    -p package_endpoint $KAFKA_PROVIDER_ENDPOINT
+    -p endpoint "$APIHOST"
 
-$WSK_CLI -i --apihost "$EDGEHOST" action update messaging/messageHubFeed "$PACKAGE_HOME/action/messageHubFeed.js" \
+# make messageHubFeed.zip
+OLD_PATH=`pwd`
+cd action
+
+if [ -e messageHubFeed.zip ]
+then
+    rm -rf messageHubFeed.zip
+fi
+
+cp -f messageHubFeed_package.json package.json
+zip -r messageHubFeed.zip lib package.json messageHubFeed.js
+
+$WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:6 messaging/messageHubFeed "$PACKAGE_HOME/action/messageHubFeed.zip" \
     --auth "$AUTH" \
     -a feed true \
     -a description 'Feed to list to Message Hub messages' \
     -a parameters '[ {"name":"kafka_brokers_sasl", "required":true, "description": "Array of Message Hub brokers"},{"name":"user", "required":true, "description": "Message Hub username"},{"name":"password", "required":true, "description": "Message Hub password", "type":"password"},{"name":"topic", "required":true, "description": "Topic to subscribe to"},{"name":"isJSONData", "required":false, "description": "Attempt to parse message value as JSON"},{"name":"isBinaryKey", "required":false, "description": "Encode key as Base64"},{"name":"isBinaryValue", "required":false, "description": "Encode message value as Base64"},{"name":"endpoint", "required":true, "description": "Hostname and port of OpenWhisk deployment"},{"name":"kafka_admin_url", "required":true, "description": "Your Message Hub admin REST URL"}]' \
     -a sampleInput '{"kafka_brokers_sasl":"[\"kafka01-prod01.messagehub.services.us-south.bluemix.net:9093\"]", "username":"someUsername", "password":"somePassword", "topic":"mytopic", "isJSONData": "false", "endpoint":"openwhisk.ng.bluemix.net", "kafka_admin_url":"https://kafka-admin-prod01.messagehub.services.us-south.bluemix.net:443"}'
 
+# create messagingWeb package and web version of feed action
+$WSK_CLI -i --apihost "$EDGEHOST" package update messagingWeb \
+    --auth "$AUTH" \
+    --shared no \
+    -p endpoint "$APIHOST" \
+    -p DB_URL "$DB_URL" \
+    -p DB_NAME "$DB_NAME" \
+
+# make messageHubFeedWeb.zip
+
+if [ -e messageHubFeedWeb.zip ]
+then
+    rm -rf messageHubFeedWeb.zip
+fi
+
+cp -f messageHubFeedWeb_package.json package.json
+zip -r messageHubFeedWeb.zip lib package.json messageHubFeedWeb.js
+
+cd $OLD_PATH
+
+
+$WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:6 messagingWeb/messageHubFeedWeb "$PACKAGE_HOME/action/messageHubFeedWeb.zip" \
+    --auth "$AUTH" \
+    --web true \
+    -a description 'Write a new trigger to MH provider DB' \
+    -a parameters '[ {"name":"kafka_brokers_sasl", "required":true, "description": "Array of Message Hub brokers"},{"name":"user", "required":true, "description": "Message Hub username"},{"name":"password", "required":true, "description": "Message Hub password", "type":"password"},{"name":"topic", "required":true, "description": "Topic to subscribe to"},{"name":"isJSONData", "required":false, "description": "Attempt to parse message value as JSON"},{"name":"isBinaryKey", "required":false, "description": "Encode key as Base64"},{"name":"isBinaryValue", "required":false, "description": "Encode message value as Base64"},{"name":"endpoint", "required":true, "description": "Hostname and port of OpenWhisk deployment"},{"name":"kafka_admin_url", "required":true, "description": "Your Message Hub admin REST URL"}]'
+
 $WSK_CLI -i --apihost "$EDGEHOST" action update messaging/messageHubProduce "$PACKAGE_HOME/action/messageHubProduce.py" \
     --auth "$AUTH" \
     --kind python:3 \
diff --git a/installKafka.sh b/installKafka.sh
old mode 100644
new mode 100755
index 96aaa14..cd92d1a
--- a/installKafka.sh
+++ b/installKafka.sh
@@ -4,8 +4,8 @@
 # automatically
 #
 # To run this command
-# ./installKafka.sh  <AUTH> <EDGEHOST> <KAFKA_TRIGGER_HOST> <KAFKA_TRIGGER_PORT> <APIHOST>
-# AUTH and APIHOST are found in $HOME/.wskprops
+# ./installKafka.sh <authkey> <edgehost> <dburl> <dbprefix> <apihost>
+# authkey and apihost are found in $HOME/.wskprops
 
 set -e
 set -x
@@ -15,16 +15,15 @@
 
 if [ $# -eq 0 ]
 then
-echo "Usage: ./installCatalog.sh <authkey> <apihost> <kafkatriggerhost> <kafkatriggerport>"
+echo "Usage: ./installKafka.sh <authkey> <edgehost> <dburl> <dbprefix> <apihost>"
 fi
 
 AUTH="$1"
 EDGEHOST="$2"
-KAFKA_TRIGGER_HOST="$3"
-KAFKA_TRIGGER_PORT="$4"
+DB_URL="$3"
+DB_NAME="${4}ow_kafka_triggers"
 APIHOST="$5"
 
-
 # If the auth key file exists, read the key in the file. Otherwise, take the
 # first argument as the key itself.
 if [ -f "$AUTH" ]; then
@@ -34,20 +33,61 @@
 # Make sure that the APIHOST is not empty.
 : ${APIHOST:?"APIHOST must be set and non-empty"}
 
-KAFKA_PROVIDER_ENDPOINT=$KAFKA_TRIGGER_HOST':'$KAFKA_TRIGGER_PORT
-
 PACKAGE_HOME="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 
 export WSK_CONFIG_FILE= # override local property file to avoid namespace clashes
 
-echo Installing the kafka feed action.
+echo Installing the Kafka package and feed action.
 
-$WSK_CLI -i --apihost "$EDGEHOST" action update messaging/kafkaFeed "$PACKAGE_HOME/action/kafkaFeed.js" \
+# make kafkaFeed.zip
+OLD_PATH=`pwd`
+cd action
+
+if [ -e kafkaFeed.zip ]
+then
+    rm -rf kafkaFeed.zip
+fi
+
+cp -f kafkaFeed_package.json package.json
+zip -r kafkaFeed.zip lib package.json kafkaFeed.js
+cd $OLD_PATH
+
+$WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:6 messaging/kafkaFeed "$PACKAGE_HOME/action/kafkaFeed.zip" \
     --auth "$AUTH" \
+    -a feed true \
     -a description 'Feed to listen to Kafka messages' \
     -a parameters '[ {"name":"brokers", "required":true, "description": "Array of Kafka brokers"}, {"name":"topic", "required":true, "description": "Topic to subscribe to"}, {"name":"isJSONData", "required":false, "description": "Attempt to parse message value as JSON"}, {"name":"isBinaryKey", "required":false, "description": "Encode key as Base64"}, {"name":"isBinaryValue", "required":false, "description": "Encode message value as Base64"}, {"name":"endpoint", "required":true, "description": "Hostname and port of OpenWhisk deployment"}]' \
     -a sampleInput '{"brokers":"[\"127.0.0.1:9093\"]", "topic":"mytopic", "isJSONData":"false", "endpoint": "openwhisk.ng.bluemix.net"}'
 
+# create messagingWeb package and web version of feed action
+$WSK_CLI -i --apihost "$EDGEHOST" package update messagingWeb \
+    --auth "$AUTH" \
+    --shared no \
+    -p endpoint "$APIHOST" \
+    -p DB_URL "$DB_URL" \
+    -p DB_NAME "$DB_NAME" \
+
+# make kafkaFeedWeb.zip
+OLD_PATH=`pwd`
+cd action
+
+if [ -e kafkaFeedWeb.zip ]
+then
+    rm -rf kafkaFeedWeb.zip
+fi
+
+cp -f kafkaFeedWeb_package.json package.json
+zip -r kafkaFeedWeb.zip lib package.json kafkaFeedWeb.js
+
+cd $OLD_PATH
+
+
+$WSK_CLI -i --apihost "$EDGEHOST" action update --kind nodejs:6 messagingWeb/kafkaFeedWeb "$PACKAGE_HOME/action/kafkaFeedWeb.zip" \
+    --auth "$AUTH" \
+    --web true \
+    -a description 'Write a new trigger to Kafka provider DB' \
+    -a parameters '[ {"name":"brokers", "required":true, "description": "Array of Kafka brokers"},{"name":"topic", "required":true, "description": "Topic to subscribe to"},{"name":"isJSONData", "required":false, "description": "Attempt to parse message value as JSON"},{"name":"isBinaryKey", "required":false, "description": "Encode key as Base64"},{"name":"isBinaryValue", "required":false, "description": "Encode message value as Base64"},{"name":"endpoint", "required":true, "description": "Hostname and port of OpenWhisk deployment"}]'
+
 $WSK_CLI -i --apihost "$EDGEHOST" action update messaging/kafkaProduce "$PACKAGE_HOME/action/kafkaProduce.py" \
     --auth "$AUTH" \
     --kind python:3 \
diff --git a/provider/app.py b/provider/app.py
index 7263339..f189981 100644
--- a/provider/app.py
+++ b/provider/app.py
@@ -14,139 +14,23 @@
 
 import logging
 import os
-import requests
-import sys
-import urllib
-import uuid
 
-from datetime import datetime
-from flask import Flask, jsonify, request
-from consumer import Consumer
+from flask import Flask, jsonify
 from consumercollection import ConsumerCollection
 from database import Database
-from threading import Lock
 from thedoctor import TheDoctor
-from urlparse import urlparse
 from health import generateHealthReport
 from gevent.wsgi import WSGIServer
+from service import Service
 
 
 app = Flask(__name__)
 app.debug = False
+
 database = Database()
-
 consumers = ConsumerCollection()
 
 
-@app.route('/triggers/<namespace>/<trigger>', methods=['PUT'])
-def postTrigger(namespace, trigger):
-    body = request.get_json(force=True, silent=True)
-    triggerFQN = '/' + namespace + '/' + trigger
-    expectedRoute = urllib.quote('/namespaces/' + namespace + '/triggers/' + trigger)
-    missing = getMissingPostFields(body)
-
-    if consumers.hasConsumerForTrigger(triggerFQN):
-        logging.warn("[{}] Trigger already exists".format(triggerFQN))
-        response = jsonify({
-            'success': False,
-            'error': "trigger already exists"
-        })
-        response.status_code = 409
-    elif len(missing) > 0:
-        response = jsonify({
-            'success': False,
-            'error': 'missing fields: %s' % ', '.join(missing)
-        })
-        response.status_code = 400
-        return response
-    elif not body["triggerURL"].endswith(expectedRoute):
-        logging.warn("[{}] Trigger and namespace from route must correspond to triggerURL".format(triggerFQN))
-        response = jsonify({
-            'success': False,
-            'error': "trigger and namespace from route must correspond to triggerURL"
-        })
-        response.status_code = 409
-    elif not body["isMessageHub"] and not enable_generic_kafka:
-        # Generic Kafka triggers has been disabled
-        logging.warn("[{}] Attempt to create generic kafka trigger while function is disabled".format(triggerFQN))
-        response = jsonify({
-            'success': False,
-            'error': "only triggers for Message Hub instances are allowed."
-        })
-        response.status_code = 403
-    else:
-        logging.info("[{}] Ensuring user has access rights to post a trigger".format(triggerFQN))
-
-        try:
-            trigger_get_response = requests.get(body["triggerURL"], verify=check_ssl)
-        except:
-            triggerURL = urlparse(body["triggerURL"])
-
-            if triggerURL.port != None:
-                triggerAddress = "{}:{}".format(triggerURL.hostname, triggerURL.port)
-            else:
-                triggerAddress = "{}".format(triggerURL.hostname)
-
-            logging.warn("[{}] Failed to communicate with OpenWhisk server ({}) for authentication".format(triggerFQN, triggerAddress))
-            response = jsonify({
-                'success': False,
-                'error': "failed to communicate with OpenWhisk server ({}) for authentication.".format(triggerAddress)
-            })
-            response.status_code = 500
-            return response
-
-        trigger_get_status_code = trigger_get_response.status_code
-        logging.info("[{}] Repsonse status code from trigger authorization {}".format(triggerFQN,
-                                                                                      trigger_get_status_code))
-        if trigger_get_status_code == 200:
-            logging.info("[{}] User authenticated. About to create consumer {}".format(triggerFQN, str(body)))
-            createAndRunConsumer(triggerFQN, body)
-            logging.info("[{}] Finished creating consumer.".format(triggerFQN))
-            response = jsonify({'success': True})
-            response.status_code = trigger_get_status_code
-        elif trigger_get_status_code == 401:
-            logging.warn("[{}] User not authorized to post trigger".format(triggerFQN))
-            response = jsonify({
-                'success': False,
-                'error': 'not authorized'
-            })
-            response.status_code = trigger_get_status_code
-        else:
-            logging.warn("[{}] Trigger authentication request failed with error code {}".format(triggerFQN,
-                trigger_get_status_code))
-            response = jsonify({'success': False})
-            response.status_code = trigger_get_status_code
-
-    return response
-
-
-@app.route('/triggers/<namespace>/<trigger>', methods=['DELETE'])
-def deleteTrigger(namespace, trigger):
-    auth = request.authorization
-    body = request.get_json(force=True, silent=True)
-
-    triggerFQN = '/' + namespace + '/' + trigger
-    consumer = consumers.getConsumerForTrigger(triggerFQN)
-    if consumer != None:
-        if authorizedForTrigger(auth, consumer):
-            if consumer.desiredState() == Consumer.State.Disabled:
-                # it's already disabled, just delete it
-                database.deleteTrigger(triggerFQN)
-                consumers.removeConsumerForTrigger(triggerFQN)
-            else:
-                consumer.shutdown()
-
-            response = jsonify({'success': True})
-        else:
-            response = jsonify({'error': 'not authorized'})
-            response.status_code = 401
-    else:
-        response = jsonify({'error': 'not found'})
-        response.status_code = 404
-
-    return response
-
-
 @app.route('/')
 def testRoute():
     return jsonify('Hi!')
@@ -157,67 +41,12 @@
     return jsonify(generateHealthReport(consumers))
 
 
-def authorizedForTrigger(auth, consumer):
-    triggerURL = urlparse(consumer.triggerURL)
-
-    return (auth != None and consumer != None and triggerURL.username == auth.username and triggerURL.password == auth.password)
-
-
-def createAndRunConsumer(triggerFQN, params, record=True):
-    # generate a random uuid for new triggers
-    if not 'uuid' in params:
-        params['uuid'] = str(uuid.uuid4())
-
-    # Create a representation for this trigger, even if it is disabled
-    # This allows it to appear in /health as well as allow it to be deleted
-    # Creating this object is lightweight and does not initialize any connections
-    consumer = Consumer(triggerFQN, params)
-    consumers.addConsumerForTrigger(triggerFQN, consumer)
-
-    if 'status' not in params or params['status']['active'] == True:
-        logging.info('{} Trigger was determined to be active, starting...'.format(triggerFQN))
-        consumer.start()
-    else:
-        logging.info('{} Trigger was determined to be disabled, not starting...'.format(triggerFQN))
-
-    if record:
-        database.recordTrigger(triggerFQN, params)
-
-
-def restoreTriggers():
-    for triggerDoc in database.triggers():
-        triggerFQN = triggerDoc['_id']
-        logging.debug('Restoring trigger {}'.format(triggerFQN))
-
-        try:
-            createAndRunConsumer(triggerFQN, triggerDoc, record=False)
-        except:
-            logging.warn('Skipping consumer due to caught exception: {}'.format(triggerDoc))
-
-
-def getMissingPostFields(fields):
-    missing = []
-    requiredFields = ['brokers', 'topic', 'triggerURL', 'isMessageHub']
-
-    if fields is None:
-        missing.extend(requiredFields)
-    else:
-        # Message Hub also requires 'username' and 'password fields'
-        if 'isMessageHub' in fields and fields['isMessageHub'] == True:
-            requiredFields.extend(['username', 'password'])
-
-        missing.extend([i for i in requiredFields if i not in fields])
-
-    if len(missing) > 0:
-        logging.info("Required fields are missing {}".format(missing))
-
-    return missing
-
-
 def main():
     logger = logging.getLogger()
     logger.setLevel(logging.INFO)
 
+    component = os.getenv('INSTANCE', 'messageHubTrigger-0')
+
     # Make sure we log to the console
     streamHandler = logging.StreamHandler()
     formatter = logging.Formatter('[%(asctime)s.%(msecs)03dZ] [%(levelname)s] [??] [kafkatriggers] %(message)s', datefmt="%Y-%m-%dT%H:%M:%S")
@@ -226,7 +55,7 @@
 
     # also log to file if /logs is present
     if os.path.isdir('/logs'):
-        fh = logging.FileHandler('/logs/kafkatriggers_logs.log')
+        fh = logging.FileHandler('/logs/{}_logs.log'.format(component))
         fh.setFormatter(formatter)
         logger.addHandler(fh)
 
@@ -246,7 +75,7 @@
 
     TheDoctor(consumers).start()
 
-    restoreTriggers()
+    Service(consumers).start()
 
     port = int(os.getenv('PORT', 5000))
     server = WSGIServer(('', port), app, log=logging.getLogger())
diff --git a/provider/consumer.py b/provider/consumer.py
index 0f0b69c..44a4321 100644
--- a/provider/consumer.py
+++ b/provider/consumer.py
@@ -61,6 +61,9 @@
     def shutdown(self):
         self.thread.shutdown()
 
+    def disable(self):
+        self.thread.setDesiredState(Consumer.State.Disabled)
+
     def start(self):
         self.thread.start()
 
@@ -216,7 +219,6 @@
 
         if self.desiredState() == Consumer.State.Dead:
             logging.info('[{}] Permanently killing consumer because desired state is Dead'.format(self.trigger))
-            self.database.deleteTrigger(self.trigger)
         elif self.desiredState() == Consumer.State.Restart:
             logging.info('[{}] Quietly letting the consumer thread stop in order to allow restart.'.format(self.trigger))
             # nothing else to do because this Thread is about to go away
diff --git a/provider/database.py b/provider/database.py
index 996bc16..73ba241 100644
--- a/provider/database.py
+++ b/provider/database.py
@@ -17,18 +17,18 @@
 import time
 import uuid
 
-from cloudant import Cloudant
+from cloudant.client import CouchDB
 from cloudant.result import Result
-from threading import Lock
 
 class Database:
     db_prefix = os.getenv('DB_PREFIX', '')
     dbname = db_prefix + 'ow_kafka_triggers'
-    username = os.environ['CLOUDANT_USER']
-    password = os.environ['CLOUDANT_PASS']
 
-    lock = Lock()
-    client = Cloudant(username, password, account=username)
+    username = os.environ['DB_USER']
+    password = os.environ['DB_PASS']
+    url = os.environ['DB_URL']
+
+    client = CouchDB(username, password, url=url)
     client.connect()
 
     filters_design_doc_id = '_design/filters'
@@ -41,79 +41,35 @@
         logging.warn('Database does not exist - creating it.')
         database = client.create_database(dbname)
 
-    def recordTrigger(self, triggerFQN, doc):
-        with self.lock:
-            try:
-                document = dict(doc)
-                document['_id'] = triggerFQN
-
-                if 'status' not in document:
-                    # set the status as active
-                    status = {
-                        'active': True,
-                        'dateChanged': time.time()
-                    }
-
-                    document['status'] = status
-
-                logging.info('Writing trigger {} to DB'.format(triggerFQN))
-                result = self.database.create_document(document)
-                logging.info('Successfully wrote trigger {} to DB'.format(triggerFQN))
-
-                return result
-            except Exception as e:
-                logging.error('[{}] Uncaught exception while recording trigger to database: {}'.format(triggerFQN, e))
-
-
-    def deleteTrigger(self, triggerFQN):
-        with self.lock:
-            try:
-                document = self.database[triggerFQN]
-
-                if document.exists():
-                    logging.info('Found trigger to delete from DB: {}'.format(triggerFQN))
-                    document.delete()
-                    logging.info('Successfully deleted trigger from DB: {}'.format(triggerFQN))
-                else:
-                    logging.warn('Attempted to delete non-existent trigger from DB: {}'.format(triggerFQN))
-            except Exception as e:
-                logging.error('[{}] Uncaught exception while deleting trigger from database: {}'.format(triggerFQN, e))
 
     def disableTrigger(self, triggerFQN, status_code):
-        with self.lock:
-            try:
-                document = self.database[triggerFQN]
+        try:
+            document = self.database[triggerFQN]
 
-                if document.exists():
-                    logging.info('Found trigger to disable from DB: {}'.format(triggerFQN))
+            if document.exists():
+                logging.info('Found trigger to disable from DB: {}'.format(triggerFQN))
 
-                    status = {
-                        'active': False,
-                        'dateChanged': time.time(),
-                        'reason': {
-                            'kind': 'AUTO',
-                            'statusCode': status_code,
-                            'message': 'Automatically disabled after receiving a {} status code when firing the trigger.'.format(status_code)
-                        }
+                status = {
+                    'active': False,
+                    'dateChanged': time.time(),
+                    'reason': {
+                        'kind': 'AUTO',
+                        'statusCode': status_code,
+                        'message': 'Automatically disabled after receiving a {} status code when firing the trigger.'.format(status_code)
                     }
+                }
 
-                    document['status'] = status
-                    document.save()
+                document['status'] = status
+                document.save()
 
-                    logging.info('{} Successfully recorded trigger as disabled.'.format(triggerFQN))
-            except Exception as e:
-                logging.error('[{}] Uncaught exception while disabling trigger: {}'.format(triggerFQN, e))
+                logging.info('{} Successfully recorded trigger as disabled.'.format(triggerFQN))
+        except Exception as e:
+            logging.error('[{}] Uncaught exception while disabling trigger: {}'.format(triggerFQN, e))
 
 
-    def triggers(self):
-        allDocs = []
+    def changesFeed(self):
+        return self.database.infinite_changes(include_docs=True)
 
-        logging.info('Fetching all triggers from DB')
-        for document in self.database.get_view_result(self.filters_design_doc_id, self.only_triggers_view_id, include_docs=True):
-            allDocs.append(document['doc'])
-
-        logging.info('Successfully retrieved {} triggers'.format(len(allDocs)))
-        return allDocs
 
     def migrate(self):
         logging.info('Starting DB migration')
diff --git a/provider/health.py b/provider/health.py
index 3eb19ec..2024d7b 100644
--- a/provider/health.py
+++ b/provider/health.py
@@ -14,7 +14,6 @@
 
 import psutil   # https://pythonhosted.org/psutil/
 
-from consumercollection import ConsumerCollection
 from datetime import datetime
 
 MILLISECONDS_IN_SECOND = 1000
@@ -118,7 +117,6 @@
 
 def getConsumers(consumers):
     consumerReports = []
-    currentTime = datetime.now()
 
     consumerCopyRO = consumers.getCopyForRead()
     for consumerId in consumerCopyRO:
diff --git a/provider/service.py b/provider/service.py
new file mode 100644
index 0000000..9b24517
--- /dev/null
+++ b/provider/service.py
@@ -0,0 +1,89 @@
+# Copyright 2016 IBM Corp. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from consumer import Consumer
+from database import Database
+from threading import Thread
+
+class Service (Thread):
+    def __init__(self, consumers):
+        Thread.__init__(self)
+        self.daemon = True
+
+        self.changes = Database().changesFeed()
+        self.consumers = consumers
+
+    def run(self):
+        while True:
+            for change in self.changes:
+                if "deleted" in change and change["deleted"] == True:
+                    logging.info('[changes] Found a delete')
+                    consumer = self.consumers.getConsumerForTrigger(change['id'])
+                    if consumer != None:
+                        if consumer.desiredState() == Consumer.State.Disabled:
+                            # just remove it from memory
+                            logging.info('[{}] Removing disabled trigger'.format(consumer.trigger))
+                            self.consumers.removeConsumerForTrigger(consumer.trigger)
+                        else:
+                            logging.info('[{}] Shutting down running trigger'.format(consumer.trigger))
+                            consumer.shutdown()
+                # since we can't use a filter function for the feed (then
+                # you don't get deletes) we need to manually verify this
+                # is a valid trigger doc that has changed
+                elif 'triggerURL' in change['doc']:
+                    logging.info('[changes] Found a change in a trigger document')
+                    document = change['doc']
+
+                    if not self.consumers.hasConsumerForTrigger(change["id"]):
+                        logging.info('[{}] Found a new trigger to create'.format(change["id"]))
+                        self.createAndRunConsumer(document)
+                    else:
+                        logging.info('[{}] Found a change to an existing trigger'.format(change["id"]))
+                        existingConsumer = self.consumers.getConsumerForTrigger(change["id"])
+
+                        if existingConsumer.desiredState() == Consumer.State.Disabled and self.__isTriggerDocActive(document):
+                            # disabled trigger has become active
+                            logging.info('[{}] Existing disabled trigger should become active'.format(change["id"]))
+                            self.createAndRunConsumer(document)
+                        elif existingConsumer.desiredState() == Consumer.State.Running and not self.__isTriggerDocActive(document):
+                            # running trigger should become disabled
+                            logging.info('[{}] Existing running trigger should become disabled'.format(change["id"]))
+                            existingConsumer.disable()
+                        else:
+                            logging.debug('[changes] Found non-interesting trigger change: \n{}\n{}'.format(existingConsumer.desiredState(), document))
+                else:
+                    logging.debug('[changes] Found a change for a non-trigger document')
+
+            logging.error("[changes] uh-oh! I made it out of the changes for loop!")
+
+
+    def createAndRunConsumer(self, doc):
+        triggerFQN = doc['_id']
+
+        # Create a representation for this trigger, even if it is disabled
+        # This allows it to appear in /health as well as allow it to be deleted
+        # Creating this object is lightweight and does not initialize any connections
+        consumer = Consumer(triggerFQN, doc)
+        self.consumers.addConsumerForTrigger(triggerFQN, consumer)
+
+        if self.__isTriggerDocActive(doc):
+            logging.info('[{}] Trigger was determined to be active, starting...'.format(triggerFQN))
+            consumer.start()
+        else:
+            logging.info('[{}] Trigger was determined to be disabled, not starting...'.format(triggerFQN))
+
+    def __isTriggerDocActive(self, doc):
+        return ('status' not in doc or doc['status']['active'] == True)
diff --git a/provider/thedoctor.py b/provider/thedoctor.py
index c861518..a36dc90 100644
--- a/provider/thedoctor.py
+++ b/provider/thedoctor.py
@@ -2,7 +2,6 @@
 import time
 
 from consumer import Consumer
-from consumercollection import ConsumerCollection
 from threading import Thread
 
 class TheDoctor (Thread):
diff --git a/tests/dat/multipleValueTypes.json b/tests/dat/multipleValueTypes.json
index 2f3b082..e22e0c4 100644
--- a/tests/dat/multipleValueTypes.json
+++ b/tests/dat/multipleValueTypes.json
@@ -2,6 +2,9 @@
   "kafka_brokers_sasl": [
     "someBroker"
   ],
+  "brokers": [
+    "someBroker"
+  ],
   "topic": "someTopic",
   "user": "someUser",
   "password": "somePassword",
diff --git a/tests/src/test/scala/system/packages/KafkaFeedTests.scala b/tests/src/test/scala/system/packages/KafkaFeedTests.scala
index cc59a36..07ea59d 100644
--- a/tests/src/test/scala/system/packages/KafkaFeedTests.scala
+++ b/tests/src/test/scala/system/packages/KafkaFeedTests.scala
@@ -42,24 +42,16 @@
 
   implicit val wskprops = WskProps()
   val wsk = new Wsk()
-  val actionName = "kafkaFeedAction"
-  val actionFile = "../action/kafkaFeed.js"
+
+  val messagingPackage = "/whisk.system/messaging"
+  val kafkaFeed = "kafkaFeed"
+  val actionName = s"${messagingPackage}/${kafkaFeed}"
 
   behavior of "Kafka feed action"
 
-  override def beforeAll() {
-    wsk.action.create(actionName, Some(actionFile))
-    super.beforeAll()
-  }
-
-  override def afterAll()  {
-    wsk.action.delete(actionName)
-    super.afterAll()
-  }
-
   it should "reject invocation when topic argument is missing" in {
     val expectedOutput = JsObject(
-      "error" -> JsString("You must supply a \"topic\" parameter.")
+      "error" -> JsString("You must supply a 'topic' parameter.")
     )
 
     runActionWithExpectedResult(actionName, "dat/missingTopic.json", expectedOutput, false)
@@ -67,20 +59,12 @@
 
   it should "reject invocation when brokers argument is missing" in  {
     val expectedOutput = JsObject(
-      "error" -> JsString("You must supply a \"brokers\" parameter as an array of Kafka brokers.")
+      "error" -> JsString("You must supply a 'brokers' parameter.")
     )
 
     runActionWithExpectedResult(actionName, "dat/missingBrokers.json", expectedOutput, false)
   }
 
-  it should "reject invocation when package_endpoint argument is missing" in {
-    val expectedOutput = JsObject(
-      "error" -> JsString("Could not find the package_endpoint parameter.")
-    )
-
-    runActionWithExpectedResult(actionName, "dat/missingPackageEndpoint.json", expectedOutput, false)
-  }
-
   it should "reject invocation when isJSONData and isBinaryValue are both enable" in {
     val expectedOutput = JsObject(
       "error" -> JsString("isJSONData and isBinaryValue cannot both be enabled.")
diff --git a/tests/src/test/scala/system/packages/KafkaFeedWebTests.scala b/tests/src/test/scala/system/packages/KafkaFeedWebTests.scala
new file mode 100644
index 0000000..3ff26a8
--- /dev/null
+++ b/tests/src/test/scala/system/packages/KafkaFeedWebTests.scala
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2015-2016 IBM Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package system.packages
+
+import org.junit.runner.RunWith
+
+import org.scalatest.BeforeAndAfter
+import org.scalatest.FlatSpec
+import org.scalatest.Matchers
+import org.scalatest.junit.JUnitRunner
+
+import com.jayway.restassured.RestAssured
+import com.jayway.restassured.config.SSLConfig
+
+import common.Wsk
+import common.WskProps
+import common.TestUtils.FORBIDDEN
+
+import spray.json._
+
+@RunWith(classOf[JUnitRunner])
+class KafkaFeedWebTests
+  extends FlatSpec
+    with BeforeAndAfter
+    with Matchers {
+
+  val wskprops = WskProps()
+
+  val webAction = "/whisk.system/messagingWeb/kafkaFeedWeb"
+  val webActionURL = s"https://${wskprops.apihost}/api/v1/web${webAction}.http"
+
+  val completeParams = JsObject(
+    "triggerName" -> JsString("/invalidNamespace/invalidTrigger"),
+    "topic" -> JsString("someTopic"),
+    "brokers" -> JsArray(JsString("someBroker")),
+    "user" -> JsString("someUsername"),
+    "password" -> JsString("somePassword"),
+    "kafka_admin_url" -> JsString("https://kafka-admin-prod01.messagehub.services.us-south.bluemix.net:443"),
+    "authKey" -> JsString("DoesNotWork")
+  )
+
+  def makePutCallWithExpectedResult(params: JsObject, expectedResult: String, expectedCode: Int) = {
+    val response = RestAssured.given()
+        .contentType("application/json\r\n")
+        .config(RestAssured.config().sslConfig(new SSLConfig().relaxedHTTPSValidation()))
+        .body(params.toString())
+        .put(webActionURL)
+    assert(response.statusCode() == expectedCode)
+    response.body.asString shouldBe expectedResult
+  }
+
+  def makeDeleteCallWithExpectedResult(expectedResult: String, expectedCode: Int) = {
+    val response = RestAssured.given().contentType("application/json\r\n").config(RestAssured.config().sslConfig(new SSLConfig().relaxedHTTPSValidation())).delete(webActionURL)
+    assert(response.statusCode() == expectedCode)
+    response.body.asString shouldBe expectedResult
+  }
+
+  behavior of "Kafka feed web action"
+
+  it should "not be obtainable using the CLI" in {
+      val wsk = new Wsk()
+      implicit val wp = wskprops
+
+      wsk.action.get(webAction, FORBIDDEN)
+  }
+
+  it should "reject post of a trigger due to missing brokers argument" in {
+    val params = JsObject(completeParams.fields - "brokers")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'brokers' parameter.", 400)
+  }
+
+  it should "reject post of a trigger due to missing topic argument" in {
+    val params = JsObject(completeParams.fields - "topic")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'topic' parameter.", 400)
+  }
+
+  it should "reject post of a trigger due to missing triggerName argument" in {
+    val params = JsObject(completeParams.fields - "triggerName")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'triggerName' parameter.", 400)
+  }
+
+  it should "reject put of a trigger when authentication fails" in {
+    makePutCallWithExpectedResult(completeParams, "You are not authorized for this trigger.", 401)
+  }
+
+  // it should "reject delete of a trigger that does not exist" in {
+  //   val expectedJSON = JsObject(
+  //     "triggerName" -> JsString("/invalidNamespace/invalidTrigger"),
+  //     "error" -> JsString("not found")
+  //   )
+  //
+  //   makeDeleteCallWithExpectedResult(expectedJSON, 404)
+  // }
+}
diff --git a/tests/src/test/scala/system/packages/MessageHubFeedTests.scala b/tests/src/test/scala/system/packages/MessageHubFeedTests.scala
index d47d56c..299087a 100644
--- a/tests/src/test/scala/system/packages/MessageHubFeedTests.scala
+++ b/tests/src/test/scala/system/packages/MessageHubFeedTests.scala
@@ -67,63 +67,50 @@
 
   implicit val wskprops = WskProps()
   val wsk = new Wsk()
-  val actionName = "messageHubFeedAction"
-  val actionFile = "../action/messageHubFeed.js"
+  val actionName = s"${messagingPackage}/${messageHubFeed}"
 
   behavior of "Message Hub feed action"
 
-  override def beforeAll() {
-    wsk.action.create(actionName, Some(actionFile))
-    super.beforeAll()
-  }
-
-  override def afterAll() {
-    wsk.action.delete(actionName)
-    super.afterAll()
-  }
-
   it should "reject invocation when topic argument is missing" in {
     val expectedOutput = JsObject(
-      "error" -> JsString("You must supply a \"topic\" parameter."))
+      "error" -> JsString("You must supply a 'topic' parameter.")
+    )
 
     runActionWithExpectedResult(actionName, "dat/missingTopic.json", expectedOutput, false)
   }
 
   it should "reject invocation when kafka_brokers_sasl argument is missing" in {
     val expectedOutput = JsObject(
-      "error" -> JsString("You must supply a \"kafka_brokers_sasl\" parameter as an array of Message Hub brokers."))
+      "error" -> JsString("You must supply a 'kafka_brokers_sasl' parameter.")
+    )
 
     runActionWithExpectedResult(actionName, "dat/missingBrokers.json", expectedOutput, false)
   }
 
   it should "reject invocation when kafka_admin_url argument is missing" in {
     val expectedOutput = JsObject("error" -> JsString(
-      "You must supply a \"kafka_admin_url\" parameter."))
+      "You must supply a 'kafka_admin_url' parameter.")
+    )
 
     runActionWithExpectedResult(actionName, "dat/missingAdminURL.json", expectedOutput, false)
   }
 
   it should "reject invocation when user argument is missing" in {
     val expectedOutput = JsObject(
-      "error" -> JsString("You must supply a \"user\" parameter to authenticate with Message Hub."))
+      "error" -> JsString("You must supply a 'user' parameter to authenticate with Message Hub.")
+    )
 
     runActionWithExpectedResult(actionName, "dat/missingUser.json", expectedOutput, false)
   }
 
   it should "reject invocation when password argument is missing" in {
     val expectedOutput = JsObject(
-      "error" -> JsString("You must supply a \"password\" parameter to authenticate with Message Hub."))
+      "error" -> JsString("You must supply a 'password' parameter to authenticate with Message Hub.")
+    )
 
     runActionWithExpectedResult(actionName, "dat/missingPassword.json", expectedOutput, false)
   }
 
-  it should "reject invocation when package_endpoint argument is missing" in {
-    val expectedOutput = JsObject(
-      "error" -> JsString("Could not find the package_endpoint parameter."))
-
-    runActionWithExpectedResult(actionName, "dat/missingPackageEndpoint.json", expectedOutput, false)
-  }
-
   it should "reject invocation when isJSONData and isBinaryValue are both enable" in {
     val expectedOutput = JsObject(
       "error" -> JsString("isJSONData and isBinaryValue cannot both be enabled."))
diff --git a/tests/src/test/scala/system/packages/MessageHubFeedWebTests.scala b/tests/src/test/scala/system/packages/MessageHubFeedWebTests.scala
new file mode 100644
index 0000000..b8aa723
--- /dev/null
+++ b/tests/src/test/scala/system/packages/MessageHubFeedWebTests.scala
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2015-2016 IBM Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package system.packages
+
+import org.junit.runner.RunWith
+
+import org.scalatest.BeforeAndAfter
+import org.scalatest.FlatSpec
+import org.scalatest.Matchers
+import org.scalatest.junit.JUnitRunner
+
+import com.jayway.restassured.RestAssured
+import com.jayway.restassured.config.SSLConfig
+
+import common.Wsk
+import common.WskProps
+import common.TestUtils.FORBIDDEN
+
+import spray.json._
+
+@RunWith(classOf[JUnitRunner])
+class MessageHubFeedWebTests
+  extends FlatSpec
+    with BeforeAndAfter
+    with Matchers {
+
+  val wskprops = WskProps()
+
+  val webAction = "/whisk.system/messagingWeb/messageHubFeedWeb"
+  val webActionURL = s"https://${wskprops.apihost}/api/v1/web${webAction}.http"
+
+  val completeParams = JsObject(
+    "triggerName" -> JsString("/invalidNamespace/invalidTrigger"),
+    "topic" -> JsString("someTopic"),
+    "kafka_brokers_sasl" -> JsArray(JsString("someBroker")),
+    "user" -> JsString("someUsername"),
+    "password" -> JsString("somePassword"),
+    "kafka_admin_url" -> JsString("https://kafka-admin-prod01.messagehub.services.us-south.bluemix.net:443"),
+    "authKey" -> JsString("DoesNotWork")
+  )
+
+  def makePutCallWithExpectedResult(params: JsObject, expectedResult: String, expectedCode: Int) = {
+    val response = RestAssured.given()
+        .contentType("application/json\r\n")
+        .config(RestAssured.config().sslConfig(new SSLConfig().relaxedHTTPSValidation()))
+        .body(params.toString())
+        .put(webActionURL)
+    assert(response.statusCode() == expectedCode)
+    response.body.asString shouldBe expectedResult
+  }
+
+  def makeDeleteCallWithExpectedResult(expectedResult: String, expectedCode: Int) = {
+    val response = RestAssured.given().contentType("application/json\r\n").config(RestAssured.config().sslConfig(new SSLConfig().relaxedHTTPSValidation())).delete(webActionURL)
+    assert(response.statusCode() == expectedCode)
+    response.body.asString shouldBe expectedResult
+  }
+
+  behavior of "Message Hub feed web action"
+
+  it should "not be obtainable using the CLI" in {
+      val wsk = new Wsk()
+      implicit val wp = wskprops
+
+      wsk.action.get(webAction, FORBIDDEN)
+  }
+
+  it should "reject post of a trigger due to missing kafka_brokers_sasl argument" in {
+    val params = JsObject(completeParams.fields - "kafka_brokers_sasl")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'kafka_brokers_sasl' parameter.", 400)
+  }
+
+  it should "reject post of a trigger due to missing topic argument" in {
+    val params = JsObject(completeParams.fields - "topic")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'topic' parameter.", 400)
+  }
+
+  it should "reject post of a trigger due to missing triggerName argument" in {
+    val params = JsObject(completeParams.fields - "triggerName")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'triggerName' parameter.", 400)
+  }
+
+  it should "reject post of a trigger due to missing user argument" in {
+    val params = JsObject(completeParams.fields - "user")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'user' parameter to authenticate with Message Hub.", 400)
+  }
+
+  it should "reject post of a trigger due to missing password argument" in {
+    val params = JsObject(completeParams.fields - "password")
+
+    makePutCallWithExpectedResult(params, "You must supply a 'password' parameter to authenticate with Message Hub.", 400)
+  }
+
+  it should "reject put of a trigger when authentication fails" in {
+    makePutCallWithExpectedResult(completeParams, "You are not authorized for this trigger.", 401)
+  }
+
+  // it should "reject delete of a trigger that does not exist" in {
+  //   val expectedJSON = JsObject(
+  //     "triggerName" -> JsString("/invalidNamespace/invalidTrigger"),
+  //     "error" -> JsString("not found")
+  //   )
+  //
+  //   makeDeleteCallWithExpectedResult(expectedJSON, 404)
+  // }
+}
diff --git a/tests/src/test/scala/system/packages/MessagingServiceTests.scala b/tests/src/test/scala/system/packages/MessagingServiceTests.scala
index 5ddd4a5..0e531f8 100644
--- a/tests/src/test/scala/system/packages/MessagingServiceTests.scala
+++ b/tests/src/test/scala/system/packages/MessagingServiceTests.scala
@@ -21,8 +21,6 @@
 import org.scalatest.Matchers
 import org.scalatest.junit.JUnitRunner
 import com.jayway.restassured.RestAssured
-import common.WhiskProperties
-import spray.json._
 
 @RunWith(classOf[JUnitRunner])
 class MessagingServiceTests
@@ -30,25 +28,13 @@
     with BeforeAndAfter
     with Matchers {
 
-  val healthEndpoint = s"/health"
+  val healthEndpoint = "/health"
 
   val getMessagingAddress =
     if (System.getProperty("host") != "" && System.getProperty("port") != "") {
       "http://" + System.getProperty("host") + ":" + System.getProperty("port")
     }
 
-  def makePutCallWithExpectedResult(url: String, params: JsObject, expectedResult: JsObject, expectedCode: Int) = {
-    val response = RestAssured.given().body(params.toString()).put(url)
-    assert(response.statusCode() == expectedCode)
-    response.body.asString.parseJson.asJsObject shouldBe expectedResult
-  }
-
-  def makeDeleteCallWithExpectedResult(url: String, expectedResult: JsObject, expectedCode: Int) = {
-    val response = RestAssured.given().delete(url)
-    assert(response.statusCode() == expectedCode)
-    response.body.asString.parseJson.asJsObject shouldBe expectedResult
-  }
-
   behavior of "Messaging feed provider endpoint"
 
   it should "return status code HTTP 200 OK from /health endpoint" in {
@@ -56,196 +42,4 @@
 
     assert(response.statusCode() == 200 && response.asString().contains("consumers"))
   }
-
-  it should "reject post of a trigger when missing all arguments" in {
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: brokers, topic, triggerURL, isMessageHub"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", JsObject(), expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to missing brokers argument" in {
-    val params = JsObject(
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString("someURL"),
-      "isMessageHub" -> JsBoolean(false)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: brokers"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to missing topic argument" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "triggerURL" -> JsString("someURL"),
-      "isMessageHub" -> JsBoolean(false)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: topic"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to missing triggerURL argument" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "topic" -> JsString("someTopic"),
-      "isMessageHub" -> JsBoolean(false)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: triggerURL"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to missing isMessageHub argument" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "triggerURL" -> JsString("someURL"),
-      "topic" -> JsString("someTopic")
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: isMessageHub"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to missing username argument" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "password" -> JsString("somePassword"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString("someURL"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: username"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to missing password argument" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "username" -> JsString("someUsername"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString("someURL"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("missing fields: password"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 400)
-  }
-
-  it should "reject post of a trigger due to mismatch between triggerURL and trigger name" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "username" -> JsString("someUsername"),
-      "password" -> JsString("somePassword"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString(s"https://someKey@${WhiskProperties.getEdgeHost}/api/v1/namespaces/invalidNamespace/triggers/someTrigger"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("trigger and namespace from route must correspond to triggerURL"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress +  "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 409)
-  }
-
-  it should "reject post of a trigger due to mismatch between triggerURL and namespace" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "username" -> JsString("someUsername"),
-      "password" -> JsString("somePassword"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString(s"https://someKey@${WhiskProperties.getEdgeHost}/api/v1/namespaces/someNamespace/triggers/invalidTrigger"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("trigger and namespace from route must correspond to triggerURL"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 409)
-  }
-
-  it should "reject post of a trigger when authentication fails" in {
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "username" -> JsString("someUsername"),
-      "password" -> JsString("somePassword"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString(s"https://someKey@${WhiskProperties.getEdgeHost}/api/v1/namespaces/invalidNamespace/triggers/invalidTrigger"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString("not authorized"),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 401)
-  }
-
-  it should "reject delete of a trigger that does not exist" in {
-    val expectedJSON = JsObject(
-      "error" -> JsString("not found")
-    )
-
-    makeDeleteCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", expectedJSON, 404)
-  }
-
-  it should "reject post of a trigger when OpenWhisk host and port fails" in {
-    val address = "0.0.0.0.0.0:9999"
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "username" -> JsString("someUsername"),
-      "password" -> JsString("somePassword"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString(s"https://someKey@$address/api/v1/namespaces/invalidNamespace/triggers/invalidTrigger"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString(s"failed to communicate with OpenWhisk server ($address) for authentication."),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 500)
-  }
-
-  it should "reject post of a trigger when OpenWhisk host fails" in {
-    val address = "0.0.0.0.0.0"
-    val params = JsObject(
-      "brokers" -> JsArray(JsString("someBroker")),
-      "username" -> JsString("someUsername"),
-      "password" -> JsString("somePassword"),
-      "topic" -> JsString("someTopic"),
-      "triggerURL" -> JsString(s"https://someKey@$address/api/v1/namespaces/invalidNamespace/triggers/invalidTrigger"),
-      "isMessageHub" -> JsBoolean(true)
-    )
-    val expectedJSON = JsObject(
-      "error" -> JsString(s"failed to communicate with OpenWhisk server ($address) for authentication."),
-      "success" -> JsBoolean(false)
-    )
-
-    makePutCallWithExpectedResult(getMessagingAddress + "/triggers/invalidNamespace/invalidTrigger", params, expectedJSON, 500)
-  }
 }