Add message queue trigger support (#218)

Adds support for message queue triggers based on the NATS-Streaming message queue.

This lets the user map a queue topic to a function, and optionally send the function's response into another queue topic.

The message queue trigger manager is designed to be message queue independent, so we should be able add support for more queues by adding implementations for the relatively simple mqtrigger.MessageQueue interface.

Compiling.md

@@ -0,0 +1,55 @@
+Compiling Fission
+=================
+
+[You only need to do this if you're making Fission changes; if you're
+just deploying Fission, use fission.yaml which points to prebuilt
+images.]
+
+You'll need the `go` compiler and tools installed, along with the
+[glide dependency management
+tool](https://github.com/Masterminds/glide#install).  You'll also need
+docker for building images.
+
+The server side is compiled as one binary ("fission-bundle") which
+contains controller, poolmgr and router; it invokes the right one
+based on command-line arguments.
+
+To build fission-bundle: clone this repo to
+`$GOPATH/src/github.com/fission/fission`, then from the top level
+directory (if you want to build the image with the docker inside
+minikube, you'll need to set the proper environment variables with
+`eval $(minikube docker-env)`):
+
+```
+  # Get dependencies
+  $ glide install
+
+  # Build fission server and an image
+  $ pushd fission-bundle
+  $ ./build.sh
+```
+
+You now need to build the docker image for fission. You can use
+`push.sh` and push it to a docker hub account. But it's easiest to use
+minikube and its built-in docker daemon:
+
+```
+  $ eval $(minikube docker-env)
+  $ docker build -t minikube/fission-bundle .
+```
+
+Next, install fission on your kubernetes cluster:
+
+```
+  # To install, update fission.yaml to point to the image you just made: "minikube/fission-bundle"
+  $ $EDITOR fission.yaml
+  $ kubectl create -f fission.yaml
+  $ kubectl create -f fission-nodeport.yaml
+```
+
+And if you're changing the CLI too, you can build it with:
+
+```
+  # Build Fission CLI
+  $ cd fission && go install
+```

Documentation/Architecture.md

@@ -95,6 +95,19 @@ While a few simple retries are done, there isn't yet a reliable
 message bus between Kubewatcher and the function.  Work for this is
 tracked in issue #64.
 
+Message Queue Trigger
+---------------------
+
+A message queue trigger binds a message queue topic to a function:
+events from that topic cause the function to be invoked with the
+message as the body of the request. The trigger may also contain a
+response topic: if specified, the function's output is sent to this
+response.
+
+Here's a diagram of the components:
+
+![Message queue trigger Diagram](https://user-images.githubusercontent.com/202578/27012344-9457cb24-4f00-11e7-8d6b-926ff01637b3.jpg)
+
 Environment Container
 ---------------------
 
@@ -113,7 +126,7 @@ volume shared between fetcher and this environment container.  Poolmgr
 then requests the container to load the function.
 
 Logger
------------
+------
 
 Logger helps to forward function logs to centralized db service for log
 persistence. Currently only influxdb is supported to store logs.

Documentation/wip/env-api.md

@@ -0,0 +1,120 @@
+# Environment V2 Fission-Environment API
+
+Fission Environments are the language-specific component of fission.
+
+They must satisfy the interface in this spec.
+
+## Meta
+
+This is version 2.0-alpha of the Fission-Environment API.
+
+(It's unstable and may change without warning until 2.0-beta.)
+
+## Overview
+
+Fission V2 Environments consist of:
+
+ * Metadata
+ * A runtime image
+ * A builder image (optional)
+ * Function Interface Specification
+ * User Documentation
+ * Examples
+
+### Metadata 
+
+See EnvironmentSpec, Runtime, and Builder in types.go.
+
+## Runtime Image
+
+An environment runtime image is a docker container image. It must run
+a server that has two jobs:
+
+(a) Loading a "function" from a file path on demand
+(b) Invoking that function on request
+
+### Function Loading
+
+The environment must expose a single HTTP endpoint (at the port and
+URL specified in the metadata) that loads a function. The function
+load request is a JSON-serialized `FunctionLoadRequest`.
+
+The function load request contains a filepath to load the function
+from. Fission does not define in any way the contents of the path; it
+is completely environment-dependent. It may be a single file, or a
+directory (in case of deployment packages).
+
+The load request may contain an EntryPoint. If it does, the loader
+must interpret this; usually it's the name of a function in a module
+or package containing multiple functions. If there is no entrypoint,
+the environment must use a default; again, the value of thsi default
+is environment-specific.
+
+The load request may contain a URL. If it does, requests to that URL
+should be routed to the function. It defaults to "/".
+
+### Function Invocation
+
+Functions are invoked on HTTP request to the server. The port for the
+request on the runtime container is defined in the Runtime metadata,
+and the URL for the request is specified in the FunctionLoadRequest.
+
+The interface of the function is environment specific; the environment
+must come with a spec for this interface.
+
+## Builder
+
+The builder is a container image that contains tools to build a
+function from source. The source may be a single file or a directory
+of files.
+
+The builder container is invoked with the specified command, with the
+following params:
+
+ 1. File path of the source
+ 2. File path where the output should go
+ 3. Other env or function-specific params passed by the user
+
+The first two parameters are file paths, and all remaining params are
+environment-specific.
+
+The output of the builder should be something that the runtime can
+load and run -- there should be no intermediate steps that need user
+intervention.
+
+### Errors
+
+## Function Interface Spec
+
+The function interface spec is a document that specifies the interface
+of functions and their semantics.  It must specify:
+
+ * How functions are invoked (sync, async)
+ * How the request context is provided to the function (URL, headers, request type, request body)
+ * Function logging
+ * Semantics of function errors and exceptions
+
+
+## Documentation
+
+The docs should contain everything necessary to use the environment:
+
+ * How to add it to a fission cluster
+ * How to write and build functions for this environment (link to the interface spec)
+ * How to modify and rebuild the environment itself
+
+## Examples
+
+Suggested examples to provide:
+
+ * A simple "Hello world"
+ * A function that demonstrates use of the request context: url params,
+   request headers, request body 
+ * A function that does logging
+ * A multi-file function package
+ * A function with dependencies
+ * Functions with shared code
+
+## Compatibility with v1
+
+V1 environment images can be used as v2 environment runtime images. 

INSTALL.md

@@ -1,14 +1,18 @@
-  * [Running Fission on your Cluster](#running-fission-on-your-cluster)
+
-     * [Setup Kubernetes](#setup-kubernetes)
+- [Running Fission on your Cluster](#running-fission-on-your-cluster)
-        * [Mac](#install-and-start-kubernetes-on-osx)
+  * [Setup Kubernetes](#setup-kubernetes)
-        * [Linux](#or-install-and-start-kubernetes-on-linux)
+    + [Install and start Kubernetes on OSX:](#install-and-start-kubernetes-on-osx)
-     * [Verify access to the cluster](#verify-access-to-the-cluster)
+    + [Or, install and start Kubernetes on Linux:](#or-install-and-start-kubernetes-on-linux)
-     * [Get and Run Fission: Minikube or Local cluster](#get-and-run-fission-minikube-or-local-cluster)
+  * [Verify access to the cluster](#verify-access-to-the-cluster)
-     * [Get and Run Fission: GKE or other Cloud](#get-and-run-fission-gke-or-other-cloud)
+  * [Get and Run Fission: Minikube or Local cluster](#get-and-run-fission-minikube-or-local-cluster)
-     * [Install the client CLI](#install-the-client-cli)
+  * [Get and Run Fission: GKE or other Cloud](#get-and-run-fission-gke-or-other-cloud)
-     * [Run an example](#run-an-example)
+  * [Get and Run Fission: OpenShift](#get-and-run-fission-openshift)
-     * [Enable Persistent Function Logs (Optional)](#enable-persistent-function-logs-optional)
+    + [Using Minishift or Local Cluster](#using-minishift-or-local-cluster)
-     * [Use the web based Fission-ui (Optional)](#use-the-web-based-fission-ui-optional)
+    + [Using other clouds](#using-other-clouds)
+  * [Install the client CLI](#install-the-client-cli)
+  * [Run an example](#run-an-example)
+  * [Enable Persistent Function Logs (Optional)](#enable-persistent-function-logs-optional)
+  * [Use the web based Fission-ui (Optional)](#use-the-web-based-fission-ui-optional)
 
 ## Running Fission on your Cluster
 
@@ -45,8 +49,8 @@ set up services with NodePort.  This exposes fission on ports 31313
 and 31314.
 
 ```
-  $ kubectl create -f http://fission.io/fission.yaml
+  $ kubectl create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-rbac.yaml
-  $ kubectl create -f http://fission.io/fission-nodeport.yaml
+  $ kubectl create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-nodeport.yaml
 ```
 
 Set the FISSION_URL and FISSION_ROUTER environment variables.
@@ -68,8 +72,8 @@ If you're using GKE or any other cloud provider that supports the
 LoadBalancer service type, use these commands:
 
 ```
-  $ kubectl create -f http://fission.io/fission.yaml
+  $ kubectl create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-rbac.yaml
-  $ kubectl create -f http://fission.io/fission-cloud.yaml
+  $ kubectl create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-cloud.yaml
 ```
 
 Save the external IP addresses of controller and router services in
@@ -84,8 +88,11 @@ svc```).  Then:
 
 ### Get and Run Fission: OpenShift
 
-If you're using OpenShift, it's possible to run Fission on it! The deployment
+If you're using OpenShift, it's possible to run Fission on it! The
-template needs to be deployed as a user with cluster-admin permissions (like `system:admin`), as it needs to create a `ClusterRole` for deploying function containers from the `fission` namespace/project.
+deployment template needs to be deployed as a user with cluster-admin
+permissions (like `system:admin`), as it needs to create a
+`ClusterRole` for deploying function containers from the `fission`
+namespace/project.
 
 Identically as with Kubernetes, you need to set the FISSION_URL and FISSION_ROUTER environment variables. If you're using minishift, use these commands:
 
@@ -93,21 +100,48 @@ Identically as with Kubernetes, you need to set the FISSION_URL and FISSION_ROUT
   $ export FISSION_URL=http://$(minishift ip):31313¬
   $ export FISSION_ROUTER=$(minishift ip):31314¬
 ```
+#### Using Minishift or Local Cluster
+
+If you're using minishift or no cloud provider, use these commands to set up services with NodePort. This exposes fission on ports 31313 and 31314.
+
+```
+  $ oc login -u system:admin
+  $ oc create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-openshift.yaml
+  $ oc create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-nodeport.yaml
+```
+
+#### Using other clouds
+
+If you're using any cloud provider that supports the LoadBalancer service type, use these commands:
+
+```
+$ oc login -u system:admin
+$ oc create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-openshift.yaml
+$ oc create -f https://github.com/fission/fission/releases/download/nightly20170621/fission-cloud.yaml
+```
+After these steps, you should be able to run fission client as with kubernetes.
 
 ### Install the client CLI
 
+#### Mac OS
+
 Get the CLI binary for Mac:
 
 ```
-  $ curl http://fission.io/mac/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
+  $ curl -Lo fission https://github.com/fission/fission/releases/download/nightly20170621/fission-cli-osx && chmod +x fission && sudo mv fission /usr/local/bin/
 ```
 
-Or Linux:
+#### Linux
 
 ```
-  $ curl http://fission.io/linux/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
+  $ curl -Lo fission https://github.com/fission/fission/releases/download/nightly20170621/fission-cli-linux && chmod +x fission && sudo mv fission /usr/local/bin/
 ```
 
+#### Windows
+
+For Windows, you can use the linux binary on WSL. Or you can download
+this windows executable: [fission.exe](https://github.com/fission/fission/releases/download/nightly20170621/fission-cli-windows.exe)
+
 ### Run an example
 
 Finally, you're ready to use Fission!
@@ -159,10 +193,29 @@ It allows users to observe and manage fission. It also provides a simple online
 To setup Fission-ui with fission in k8s is simple:
 
 ```bash
-  # After Fission deployed
+  # Run this after fission is deployed
   $ kubectl create -f https://raw.githubusercontent.com/fission/fission-ui/master/docker/fission-ui.yaml
 ```
 
 Then open `http://node-ip:31319` to use Fission-ui.
 
 For more infomation, please check out [Fission-ui Readme](https://github.com/fission/fission-ui/blob/master/README.md).
+
+### Install NATS for message-queue based triggers (Optional)
+
+Fission supports message queue triggers that allow you to invoke
+functions based on events in a queue.  For now, NATS-Streaming is the
+only supported message queue.
+
+You can install NATS Streaming on your Kubernetes cluster with:
+
+```
+  $ kubectl create -f fission-nats.yaml
+```
+
+You can subscribe to a NATS Streaming queue with a command like this:
+(See `fission mqtrigger --help` for details)
+
+```
+  $ fission mqtrigger create --name myQueueTrigger --function processEvent --topic "myQueue.request" 
+```

README.md

@@ -81,187 +81,12 @@ See the [examples](examples) directory for more.
 Running Fission on your Cluster
 ===============================
 
-### Setup Kubernetes
+See the [installation guide](INSTALL.md).
-
-You can install Kubernetes on your laptop with [minikube](https://github.com/kubernetes/minikube):
-
-#### Install and start Kubernetes on OSX:
-```bash
-  $ curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/darwin/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin
-  $ curl -Lo minikube https://storage.googleapis.com/minikube/releases/v0.16.0/minikube-darwin-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
-  $ minikube start
-```
-
-#### Or, install and start Kubernetes on Linux:
-```bash
-  $ curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin
-  $ curl -Lo minikube https://storage.googleapis.com/minikube/releases/v0.16.0/minikube-linux-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
-  $ minikube start
-```
-
-Or, you can use [Google Container Engine's](https://cloud.google.com/container-engine/) free trial to get a 3 node cluster.
-
-### Verify access to the cluster
-
-```
-  $ kubectl version
-```
-
-### Get and Run Fission: Minikube or Local cluster
-
-If you're using minikube or no cloud provider, use these commands to
-set up services with NodePort.  This exposes fission on ports 31313
-and 31314.
-
-```
-  $ kubectl create -f http://fission.io/fission.yaml
-  $ kubectl create -f http://fission.io/fission-nodeport.yaml
-```
-
-Set the FISSION_URL and FISSION_ROUTER environment variables.
-FISSION_URL is used by the fission CLI to find the server.
-FISSION_URL should be prefixed with a `http://`.  (FISSION_ROUTER is
-only needed for the examples below to work.)
-
-If you're using minikube, use these commands:
-
-```
-  $ export FISSION_URL=http://$(minikube ip):31313
-  $ export FISSION_ROUTER=$(minikube ip):31314
-```
-
-
-### Get and Run Fission: GKE or other Cloud
-
-If you're using GKE or any other cloud provider that supports the
-LoadBalancer service type, use these commands:
-
-```
-  $ kubectl create -f http://fission.io/fission.yaml
-  $ kubectl create -f http://fission.io/fission-cloud.yaml
-```
-
-Save the external IP addresses of controller and router services in
-FISSION_URL and FISSION_ROUTER, respectively.  Wait for services to
-get IP addresses (check this with ```kubectl --namespace fission get
-svc```).  Then:
-
-```
-  $ export FISSION_URL=http://$(kubectl --namespace fission get svc controller -o=jsonpath='{..ip}')
-  $ export FISSION_ROUTER=$(kubectl --namespace fission get svc router -o=jsonpath='{..ip}')
-```
-### Get and run fission: OpenShift
-
-If you're using OpenShift, it's possible to run Fission on it! The deployment
-template needs to be deployed as a user with cluster-admin permissions (like `system:admin`), as it needs to create a `ClusterRole` for deploying function containers from the `fission` namespace/project.
-
-Identically as with Kubernetes, you need to set the FISSION_URL and FISSION_ROUTER environment variables. If you're using minishift, use these commands:
-
-```
-  $ export FISSION_URL=http://$(minishift ip):31313¬
-  $ export FISSION_ROUTER=$(minishift ip):31314¬
-```
-
-#### Using Minishift or Local Cluster
-
-If you're using minishift or no cloud provider, use these commands to set up services with NodePort. This exposes fission on ports 31313 and 31314.
-
-```
-  $ oc login -u system:admin
-  $ oc create -f http://fission.io/fission-openshift.yaml
-  $ oc create -f http://fission.io/fission-nodeport.yaml
-```
-
-#### Using other clouds
-If you're using any cloud provider that supports the LoadBalancer service type, use these commands:
-
-```
-$ oc login -u system:admin
-$ oc create -f http://fission.io/fission-openshift.yaml
-$ oc create -f http://fission.io/fission-cloud.yaml
-```
-After these steps, you should be able to run fission client as with kubernetes.
-
-### Install the client CLI
-
-Get the CLI binary for Mac:
-
-```
-  $ curl http://fission.io/mac/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
-```
-
-Or Linux:
-
-```
-  $ curl http://fission.io/linux/fission > fission && chmod +x fission && sudo mv fission /usr/local/bin/
-```
-
-### Run an example
-
-Finally, you're ready to use Fission!
-
-```
-  $ fission env create --name nodejs --image fission/node-env
-
-  $ curl https://raw.githubusercontent.com/fission/fission/master/examples/nodejs/hello.js > hello.js
-
-  $ fission function create --name hello --env nodejs --code hello.js
-  
-  $ fission route create --method GET --url /hello --function hello
-  
-  $ curl http://$FISSION_ROUTER/hello
-  Hello, world!
-```
-
-You can also set up persistence for logs: [instructions here](INSTALL.md).
-
 
 Compiling Fission
 =================
 
-[You only need to do this if you're making Fission changes; if you're
+See the [compilation guide](Compiling.md).
-just deploying Fission, use fission.yaml which points to prebuilt
-images.]
-
-You'll need go installed, along with the [glide dependency management
-tool](https://github.com/Masterminds/glide#install).
-You'll also need docker for building images.
-
-The server side is compiled as one binary ("fission-bundle") which
-contains controller, poolmgr and router; it invokes the right one
-based on command-line arguments.
-
-To build fission-bundle: clone this repo to
-`$GOPATH/src/github.com/fission/fission`, then from the top level
-directory (if you want to build the image with the docker inside
-minikube, you'll need to set the proper environment variables with
-`eval $(minikube docker-env)`):
-
-```
-  # Get dependencies
-  $ glide install
-
-  # Build fission server and an image
-  $ pushd fission-bundle
-  $ ./build.sh
-
-  # Edit push.sh to point to your registry, or comment out the `docker push`
-  # line if building into your local minikube for dev purposes
-  $ $EDITOR push.sh
-  $ ./push.sh
-  $ popd
-
-  # To install, update fission.yaml to point to your compiled image
-  $ $EDITOR fission.yaml
-  $ kubectl create -f fission.yaml
-```
-
-If you're changing the CLI:
-
-```
-  # Build Fission CLI
-  $ cd fission && go install
-```
 
 Status
 ======

controller/api.go

@@ -36,6 +36,7 @@ type (
 		FunctionStore
 		HTTPTriggerStore
 		TimeTriggerStore
+		MessageQueueTriggerStore
 		EnvironmentStore
 		WatchStore
 	}
@@ -49,11 +50,12 @@ type (
 
 func MakeAPI(rs *ResourceStore) *API {
 	api := &API{
-		FunctionStore:    FunctionStore{ResourceStore: *rs},
+		FunctionStore:            FunctionStore{ResourceStore: *rs},
-		HTTPTriggerStore: HTTPTriggerStore{ResourceStore: *rs},
+		HTTPTriggerStore:         HTTPTriggerStore{ResourceStore: *rs},
-		TimeTriggerStore: TimeTriggerStore{ResourceStore: *rs},
+		TimeTriggerStore:         TimeTriggerStore{ResourceStore: *rs},
-		EnvironmentStore: EnvironmentStore{ResourceStore: *rs},
+		MessageQueueTriggerStore: MessageQueueTriggerStore{ResourceStore: *rs},
-		WatchStore:       WatchStore{ResourceStore: *rs},
+		EnvironmentStore:         EnvironmentStore{ResourceStore: *rs},
+		WatchStore:               WatchStore{ResourceStore: *rs},
 	}
 	return api
 }
@@ -130,6 +132,12 @@ func (api *API) Serve(port int) {
 	r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiUpdate).Methods("PUT")
 	r.HandleFunc("/v1/triggers/time/{timeTrigger}", api.TimeTriggerApiDelete).Methods("DELETE")
 
+	r.HandleFunc("/v1/triggers/messagequeue", api.MessageQueueTriggerApiList).Methods("GET")
+	r.HandleFunc("/v1/triggers/messagequeue", api.MessageQueueApiCreate).Methods("POST")
+	r.HandleFunc("/v1/triggers/messagequeue/{mqTrigger}", api.MessageQueueApiGet).Methods("GET")
+	r.HandleFunc("/v1/triggers/messagequeue/{mqTrigger}", api.MessageQueueApiUpdate).Methods("PUT")
+	r.HandleFunc("/v1/triggers/messagequeue/{mqTrigger}", api.MessageQueueApiDelete).Methods("DELETE")
+
 	r.HandleFunc("/proxy/{dbType}", api.FunctionLogsApiPost).Methods("POST")
 
 	address := fmt.Sprintf(":%v", port)

controller/client/client.go

@@ -640,3 +640,116 @@ func (c *Client) TimeTriggerList() ([]fission.TimeTrigger, error) {
 
 	return triggers, nil
 }
+
+func (c *Client) MessageQueueTriggerCreate(t *fission.MessageQueueTrigger) (*fission.Metadata, error) {
+	reqbody, err := json.Marshal(t)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.Post(c.url("triggers/messagequeue"), "application/json", bytes.NewReader(reqbody))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleCreateResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var m fission.Metadata
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		return nil, err
+	}
+
+	return &m, nil
+}
+
+func (c *Client) MessageQueueTriggerGet(m *fission.Metadata) (*fission.MessageQueueTrigger, error) {
+	relativeUrl := fmt.Sprintf("triggers/messagequeue/%v", m.Name)
+	if len(m.Uid) > 0 {
+		relativeUrl += fmt.Sprintf("?uid=%v", m.Uid)
+	}
+
+	resp, err := http.Get(c.url(relativeUrl))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var t fission.MessageQueueTrigger
+	err = json.Unmarshal(body, &t)
+	if err != nil {
+		return nil, err
+	}
+
+	return &t, nil
+}
+
+func (c *Client) MessageQueueTriggerUpdate(mqTrigger *fission.MessageQueueTrigger) (*fission.Metadata, error) {
+	reqbody, err := json.Marshal(mqTrigger)
+	if err != nil {
+		return nil, err
+	}
+	relativeUrl := fmt.Sprintf("triggers/messagequeue/%v", mqTrigger.Metadata.Name)
+
+	resp, err := c.put(relativeUrl, "application/json", reqbody)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	var m fission.Metadata
+	err = json.Unmarshal(body, &m)
+	if err != nil {
+		return nil, err
+	}
+	return &m, nil
+}
+
+func (c *Client) MessageQueueTriggerDelete(m *fission.Metadata) error {
+	relativeUrl := fmt.Sprintf("triggers/messagequeue/%v", m.Name)
+	if len(m.Uid) > 0 {
+		relativeUrl += fmt.Sprintf("?uid=%v", m.Uid)
+	}
+	err := c.delete(relativeUrl)
+	return err
+}
+
+func (c *Client) MessageQueueTriggerList(mqType string) ([]fission.MessageQueueTrigger, error) {
+	relativeUrl := "triggers/messagequeue"
+	if len(mqType) > 0 {
+		relativeUrl += fmt.Sprintf("?mqtype=%v", mqType)
+	}
+
+	resp, err := http.Get(c.url(relativeUrl))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := c.handleResponse(resp)
+	if err != nil {
+		return nil, err
+	}
+
+	triggers := make([]fission.MessageQueueTrigger, 0)
+	err = json.Unmarshal(body, &triggers)
+	if err != nil {
+		return nil, err
+	}
+
+	return triggers, nil
+}

controller/mqTriggerApi.go

@@ -0,0 +1,166 @@
+/*
+Copyright 2017 The Fission Authors.
+
+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 controller
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/gorilla/mux"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/fission/fission"
+)
+
+func (api *API) MessageQueueTriggerApiList(w http.ResponseWriter, r *http.Request) {
+	mqType := r.FormValue("mqtype")
+	triggers, err := api.MessageQueueTriggerStore.List(mqType)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	resp, err := json.Marshal(triggers)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) MessageQueueApiCreate(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	var mqTrigger fission.MessageQueueTrigger
+	err = json.Unmarshal(body, &mqTrigger)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	// trigger name must not conflict with any other trigger
+	// even they are different message queue type
+	triggers, err := api.MessageQueueTriggerStore.List("")
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	for _, trigger := range triggers {
+		if trigger.Name == mqTrigger.Name {
+			err = fission.MakeError(fission.ErrorNameExists,
+				"Message queue trigger with same name already exists")
+			api.respondWithError(w, err)
+			return
+		}
+	}
+
+	// save trigger info
+	uid, err := api.MessageQueueTriggerStore.Create(&mqTrigger)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	mqTriggerMeta := fission.Metadata{Name: mqTrigger.Metadata.Name, Uid: uid}
+	resp, err := json.Marshal(mqTriggerMeta)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	w.WriteHeader(http.StatusCreated)
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) MessageQueueApiGet(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	mqTriggerMeta := fission.Metadata{
+		Name: vars["mqTrigger"],
+		Uid:  r.FormValue("uid"), // empty if uid is absent
+	}
+	mqTrigger, err := api.MessageQueueTriggerStore.Get(&mqTriggerMeta)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	resp, err := json.Marshal(mqTrigger)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) MessageQueueApiUpdate(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	mqtName := vars["mqTrigger"]
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	var mqTrigger fission.MessageQueueTrigger
+	err = json.Unmarshal(body, &mqTrigger)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	if mqtName != mqTrigger.Metadata.Name {
+		err = fission.MakeError(fission.ErrorInvalidArgument, "Message queue trigger name doesn't match URL")
+		api.respondWithError(w, err)
+		return
+	}
+
+	uid, err := api.MessageQueueTriggerStore.Update(&mqTrigger)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+
+	mqTriggerMeta := fission.Metadata{Name: mqTrigger.Metadata.Name, Uid: uid}
+	resp, err := json.Marshal(mqTriggerMeta)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	api.respondWithSuccess(w, resp)
+}
+
+func (api *API) MessageQueueApiDelete(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	mqTriggerMeta := fission.Metadata{
+		Name: vars["mqTrigger"],
+		Uid:  r.FormValue("uid"), // empty if uid is absent
+	}
+
+	if len(mqTriggerMeta.Uid) == 0 {
+		log.WithFields(log.Fields{"mqTrigger": mqTriggerMeta.Name}).Info("Deleting all versions")
+	}
+
+	err := api.MessageQueueTriggerStore.Delete(mqTriggerMeta)
+	if err != nil {
+		api.respondWithError(w, err)
+		return
+	}
+	api.respondWithSuccess(w, []byte(""))
+}

controller/mqTriggerStore.go

@@ -0,0 +1,80 @@
+/*
+CopyrigmqTrigger 2017 The Fission Authors.
+
+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
+
+    mqTriggertp://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 controller
+
+import (
+	"github.com/fission/fission"
+	uuid "github.com/satori/go.uuid"
+)
+
+type MessageQueueTriggerStore struct {
+	ResourceStore
+}
+
+func (mqs *MessageQueueTriggerStore) Create(mqTrigger *fission.MessageQueueTrigger) (string, error) {
+	mqTrigger.Metadata.Uid = uuid.NewV4().String()
+	return mqTrigger.Metadata.Uid, mqs.ResourceStore.create(mqTrigger)
+}
+
+func (mqs *MessageQueueTriggerStore) Get(m *fission.Metadata) (*fission.MessageQueueTrigger, error) {
+	var mqTrigger fission.MessageQueueTrigger
+	err := mqs.ResourceStore.read(m.Name, &mqTrigger)
+	if err != nil {
+		return nil, err
+	}
+	return &mqTrigger, nil
+}
+
+func (mqs *MessageQueueTriggerStore) Update(mqTrigger *fission.MessageQueueTrigger) (string, error) {
+	mqTrigger.Metadata.Uid = uuid.NewV4().String()
+	return mqTrigger.Metadata.Uid, mqs.ResourceStore.update(mqTrigger)
+}
+
+func (mqs *MessageQueueTriggerStore) Delete(m fission.Metadata) error {
+	typeName, err := getTypeName(fission.MessageQueueTrigger{})
+	if err != nil {
+		return err
+	}
+	return mqs.ResourceStore.delete(typeName, m.Name)
+}
+
+func (mqs *MessageQueueTriggerStore) List(mqType string) ([]fission.MessageQueueTrigger, error) {
+	typeName, err := getTypeName(fission.MessageQueueTrigger{})
+	if err != nil {
+		return nil, err
+	}
+	bufs, err := mqs.ResourceStore.getAll(typeName)
+	if err != nil {
+		return nil, err
+	}
+
+	triggers := make([]fission.MessageQueueTrigger, 0, len(bufs))
+	js := JsonSerializer{}
+	for _, buf := range bufs {
+		var mqTrigger fission.MessageQueueTrigger
+		err = js.deserialize([]byte(buf), &mqTrigger)
+		if err != nil {
+			return nil, err
+		}
+		if len(mqType) > 0 && mqType != mqTrigger.MessageQueueType {
+			continue
+		}
+		triggers = append(triggers, mqTrigger)
+	}
+
+	return triggers, nil
+}

environments/ruby/README.md

@@ -2,9 +2,19 @@
 
 This is the Ruby environment for Fission.
 
-It's a Docker image containing a Ruby 2.4.1 runtime, along with a
+It's a Docker image containing a Ruby 2.4.1 runtime. The image uses
-dynamic loader.  A few common dependencies are included in the
+Rack with WEBrick to host the internal web server.
-Gemfile.
+
+The environment works via convention where you create a Ruby method
+called `handler` with a single optional argument, a `Fission::Context`
+object.
+
+The `Fission::Context` object gives access to the Rack env, and a
+request object. Please see `fission/context.rb` for the public api.
+
+The `Fission::Request` object is a subclass of `Rack::Request` and
+provides access to parameters and headers. See `fission/request.rb`
+for the public api.
 
 ## Customizing this image
 
@@ -39,3 +49,7 @@ Or, if you already have an environment, you can update its image:
 
 After this, fission functions that have the env parameter set to the
 same environment name as this command will use this environment.
+
+## Creating functions to use this image
+
+See the [examples README](examples/ruby/README.md).

environments/ruby/fission/context.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+require_relative 'request'
+
+module Fission
+  class Context
+    attr_reader :env
+
+    def initialize(env)
+      @env = env
+    end
+
+    def request
+      @request ||= Request.new(env)
+    end
+  end
+end

environments/ruby/fission/handler.rb

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+require_relative 'context'
+
+module Fission
+  module Handler
+    def self.call(env)
+      response = if method(:handler).arity > 0
+        handler(Context.new(env))
+      else
+        handler
+      end
+
+      response.is_a?(Array) ? response : Rack::Response.new([response]).finish
+    rescue
+      Rack::Response.new(['500 Internal Server Error'], 500, {}).finish
+    end
+  end
+end

environments/ruby/fission/request.rb

@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+module Fission
+  HEADER_PREFIX = 'HTTP_'
+  PARAM_HEADER_PREFIX = 'HTTP_X_FISSION_PARAMS_'
+  PARAMETERS_KEY = 'fission.request.parameters'
+
+  class Request < Rack::Request
+    def headers
+      Hash[
+        *env.select { |k,v| k.start_with?(HEADER_PREFIX) }
+             .map { |k,v| [k.sub(/\A#{HEADER_PREFIX}/, '').split('_').map(&:capitalize).join('-'), v] }
+             .sort
+             .flatten
+      ]
+    end
+
+    def params
+      env[PARAMETERS_KEY] ||= super.merge(path_parameters)
+    end
+
+    def path_parameters
+      Hash[
+        *env.select { |k,v| k.start_with?(PARAM_HEADER_PREFIX) }
+             .map { |k,v| [k.sub(/\A#{PARAM_HEADER_PREFIX}/, '').downcase, v] }
+             .sort
+             .flatten
+      ]
+    end
+  end
+end

environments/ruby/fission/specializer.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+module Fission
+  CODE_PATH = '/userfunc/user'
+
+  module Specializer
+    def self.call(env)
+      load CODE_PATH
+      Rack::Response.new([], 201).finish
+    rescue
+      Rack::Response.new(['500 Internal Server Error'], 500, {}).finish
+    end
+  end
+end

environments/ruby/server.rb

@@ -1,55 +1,9 @@
 # frozen_string_literal: true
+
 require 'rack'
 
-CODEPATH = '/userfunc/user'
+require_relative 'fission/specializer'
-
+require_relative 'fission/handler'
-module Fission
-  class Request < Rack::Request
-    def headers
-      Hash[
-        *env.select { |k,v| k.start_with?('HTTP_') }
-             .map { |k,v| [k.sub(/\AHTTP_/, '').split('_').map(&:capitalize).join('-'), v] }
-             .sort
-             .flatten
-      ]
-    end
-  end
-
-  class Context
-    attr_reader :env
-
-    def initialize(env)
-      @env = env
-    end
-
-    def request
-      @request ||= Request.new(env)
-    end
-  end
-
-  module Specializer
-    def self.call(env)
-      load CODEPATH
-      Rack::Response.new([], 201).finish
-    rescue
-      Rack::Response.new(['500 Internal Server Error'], 500, {}).finish
-    end
-  end
-
-  module Handler
-    def self.call(env)
-      response = if method(:handler).arity > 0
-        handler(Context.new(env))
-      else
-        handler
-      end
-
-      response.is_a?(Array) ? response : Rack::Response.new([response]).finish
-    rescue
-      Rack::Response.new(['500 Internal Server Error'], 500, {}).finish
-    end
-  end
-end
 
 app = Rack::Builder.new do
   use Rack::CommonLogger, $stderr

examples/ruby/README.md

@@ -0,0 +1,107 @@
+# Ruby examples
+
+This directory contains several examples to get you started using Ruby
+with Fission.
+
+Before running any of these functions, make sure you have created a
+`ruby` Fission environment:
+
+```
+$ fission env create --name ruby --image USER/ruby-env
+```
+
+## Method signature
+
+A standard Ruby function has the basic form:
+
+```ruby
+def handler(context)
+  return [200, {}, []]
+end
+```
+
+If the fission context is not required, the function can be simplified:
+
+```ruby
+def handler
+  [200, {}, ["Hello, world!\n"]]
+end
+```
+
+If a simple text response is to be returned, with a status of 200, this
+can be further simplified.
+
+```ruby
+def handler
+  "Hello, world!\n"
+end
+```
+
+## Hello example (`hello.rb`)
+
+This example is the simplest possible Ruby function, as described above.
+
+To run the example:
+
+```
+$ fission function create --name hello --env ruby --code examples/ruby/hello.rb
+
+$ fission route create --method GET --url /hello --function hello
+
+$ curl http://$FISSION_ROUTER/hello
+  Hello, world!
+```
+
+## Request data example (`request_data.rb`)
+
+This example shows basic use of the `Fission::Context` and
+`Fission::Request` objects.
+
+To run the example:
+
+```
+$ fission function create --name request --env ruby --code examples/ruby/request_data.rb
+
+$ fission route create --method GET --url /request/{id} --function request
+
+$ curl http://$FISSION_ROUTER/request/123?key=abc
+  ---ENV---
+  GATEWAY_INTERFACE=CGI/1.1
+  PATH_INFO=/
+  QUERY_STRING=key=abc
+  REMOTE_ADDR=172.17.0.8
+  REMOTE_HOST=172.17.0.8
+  REQUEST_METHOD=GET
+  REQUEST_URI=http://192.168.64.200:31314/?key=abc
+  SCRIPT_NAME=
+  SERVER_NAME=192.168.64.200
+  SERVER_PORT=31314
+  SERVER_PROTOCOL=HTTP/1.1
+  SERVER_SOFTWARE=WEBrick/1.3.1 (Ruby/2.4.1/2017-03-22)
+  HTTP_HOST=192.168.64.200:31314
+  HTTP_USER_AGENT=curl/7.52.1
+  HTTP_ACCEPT=*/*
+  HTTP_X_FISSION_PARAMS_ID=123
+  HTTP_X_FORWARDED_FOR=172.17.0.1
+  HTTP_ACCEPT_ENCODING=gzip
+  rack.version=1=3
+  ...
+  HTTP_VERSION=HTTP/1.1
+  REQUEST_PATH=/
+
+  ---HEADERS---
+  Accept: */*
+  Accept-Encoding: gzip
+  Host: 192.168.64.200:31314
+  User-Agent: curl/7.52.1
+  Version: HTTP/1.1
+  X-Fission-Params-Id: 123
+  X-Forwarded-For: 172.17.0.1
+
+  ---PARAMS---
+  key=abc
+  id=123
+
+  --BODY--
+
+```

fission-bundle/main.go

@@ -8,6 +8,7 @@ import (
 	"github.com/fission/fission/controller"
 	"github.com/fission/fission/kubewatcher"
 	"github.com/fission/fission/logger"
+	"github.com/fission/fission/mqtrigger"
 	"github.com/fission/fission/poolmgr"
 	"github.com/fission/fission/router"
 	"github.com/fission/fission/timer"
@@ -61,6 +62,13 @@ func runTimer(controllerUrl, routerUrl string) {
 	}
 }
 
+func runMessageQueueMgr(controllerUrl, routerUrl string) {
+	err := messagequeue.Start(controllerUrl, routerUrl)
+	if err != nil {
+		log.Fatalf("Error starting timer: %v", err)
+	}
+}
+
 func getPort(portArg interface{}) int {
 	portArgStr := portArg.(string)
 	port, err := strconv.Atoi(portArgStr)
@@ -96,6 +104,7 @@ Usage:
   fission-bundle --kubewatcher [--controllerUrl=<url> --routerUrl=<url>]
   fission-bundle --logger
   fission-bundle --timer [--controllerUrl=<url> --routerUrl=<url>]
+  fission-bundle --mqt   [--controllerUrl=<url> --routerUrl=<url>]
 Options:
   --controllerPort=<port>  Port that the controller should listen on.
   --routerPort=<port>      Port that the router should listen on.
@@ -109,6 +118,7 @@ Options:
   --kubewatcher            Start Kubernetes events watcher.
   --logger                 Start logger.
   --timer 		           Start Timer.
+  --mqt 		           Start message queue trigger.
 `
 	arguments, err := docopt.Parse(usage, nil, true, "fission-bundle", false)
 	if err != nil {
@@ -149,5 +159,9 @@ Options:
 		runTimer(controllerUrl, routerUrl)
 	}
 
+	if arguments["--mqt"] == true {
+		runMessageQueueMgr(controllerUrl, routerUrl)
+	}
+
 	select {}
 }

fission-cloud.yaml

@@ -49,3 +49,21 @@ spec:
     nodePort: 31315
   selector:
     svc: influxdb
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: nats-streaming
+  namespace: fission
+  labels:
+    svc: nats-streaming
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 4222
+    targetPort: 4222
+    nodePort: 31316
+  selector:
+    svc: nats-streaming

fission-nats.yaml

@@ -0,0 +1,60 @@
+# To enable nats authentication, please follow the instruction described in
+# http://nats.io/documentation/server/gnatsd-authentication/.
+# And dont forget to change the MESSAGE_QUEUE_URL in mqtrigger deployment.
+
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: mqtrigger
+  namespace: fission
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        svc: mqtrigger
+    spec:
+      containers:
+      - name: mqtrigger
+        image: fission/fission-bundle
+        command: ["/fission-bundle"]
+        args: ["--mqt"]
+        env:
+        - name: MESSAGE_QUEUE_TYPE
+          value: nats-streaming
+        - name: MESSAGE_QUEUE_URL
+          value: nats://nats-streaming:4222
+
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  labels:
+    svc: nats-streaming
+  name: nats-streaming
+  namespace: fission
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      svc: nats-streaming
+  strategy:
+    rollingUpdate:
+      maxSurge: 1
+      maxUnavailable: 1
+    type: RollingUpdate
+  template:
+    metadata:
+      creationTimestamp: null
+      labels:
+        svc: nats-streaming
+    spec:
+      containers:
+      - name: nats-streaming
+        image: nats-streaming
+        args: ["--cluster_id", "fissionMQTrigger"]
+        ports:
+        - containerPort: 4222
+          hostPort: 4222
+          protocol: TCP
+

fission-nodeport.yaml

@@ -49,3 +49,21 @@ spec:
     nodePort: 31315
   selector:
     svc: influxdb
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: nats-streaming
+  namespace: fission
+  labels:
+    svc: nats-streaming
+spec:
+  type: NodePort
+  ports:
+  - port: 4222
+    targetPort: 4222
+    nodePort: 31316
+  selector:
+    svc: nats-streaming

fission/function.go

@@ -179,9 +179,16 @@ func fnUpdate(c *cli.Context) error {
 	}
 
 	fileName := c.String("code")
+	if len(fileName) == 0 {
+		fileName = c.String("package")
+	}
+
+	if len(envName) == 0 && len(fileName) == 0 {
+		fatal("Need --env or --code or --package argument.")
+	}
+
 	if len(fileName) > 0 {
 		code := fnFetchCode(fileName)
-
 		function.Code = string(code)
 	}
 

fission/main.go

@@ -82,6 +82,21 @@ func main() {
 		{Name: "list", Usage: "List Time triggers", Flags: []cli.Flag{}, Action: ttList},
 	}
 
+	// Message queue trigger
+	mqtNameFlag := cli.StringFlag{Name: "name", Usage: "Message queue Trigger name"}
+	mqtFnNameFlag := cli.StringFlag{Name: "function", Usage: "Function name"}
+	mqtFnUidFlag := cli.StringFlag{Name: "uid", Usage: "Function UID (optional; uses latest if unspecified)"}
+	mqtMQTypeFlag := cli.StringFlag{Name: "mqtype", Usage: "Message queue type, e.g. nats-streaming  (optional; uses \"nats-streaming\" if unspecified)"}
+	mqtTopicFlag := cli.StringFlag{Name: "topic", Usage: "Message queue Topic the trigger listens on"}
+	mqtRespTopicFlag := cli.StringFlag{Name: "resptopic", Usage: "Topic that the function response is sent on (optional; response discarded if unspecified)"}
+	mqtSubcommands := []cli.Command{
+		{Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, mqtFnUidFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag}, Action: mqtCreate},
+		{Name: "get", Usage: "Get message queue trigger", Flags: []cli.Flag{}, Action: mqtGet},
+		{Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtTopicFlag, mqtRespTopicFlag}, Action: mqtUpdate},
+		{Name: "delete", Usage: "Delete message queue trigger", Flags: []cli.Flag{mqtNameFlag}, Action: mqtDelete},
+		{Name: "list", Usage: "List message queue triggers", Flags: []cli.Flag{mqtMQTypeFlag}, Action: mqtList},
+	}
+
 	// environments
 	envNameFlag := cli.StringFlag{Name: "name", Usage: "Environment name"}
 	envImageFlag := cli.StringFlag{Name: "image", Usage: "Environment image URL"}
@@ -112,6 +127,7 @@ func main() {
 		{Name: "function", Aliases: []string{"fn"}, Usage: "Create, update and manage functions", Subcommands: fnSubcommands},
 		{Name: "httptrigger", Aliases: []string{"ht", "route"}, Usage: "Manage HTTP triggers (routes) for functions", Subcommands: htSubcommands},
 		{Name: "timetrigger", Aliases: []string{"tt", "timer"}, Usage: "Manage Time triggers (timers) for functions", Subcommands: ttSubcommands},
+		{Name: "mqtrigger", Aliases: []string{"mqt", "messagequeue"}, Usage: "Manage message queue triggers for functions", Subcommands: mqtSubcommands},
 		{Name: "environment", Aliases: []string{"env"}, Usage: "Manage environments", Subcommands: envSubcommands},
 		{Name: "watch", Aliases: []string{"w"}, Usage: "Manage watches", Subcommands: wSubCommands},
 

fission/mqtrigger.go

@@ -0,0 +1,155 @@
+/*
+Copyrigtt 2017 The Fission Authors.
+
+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
+
+    tttp://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 main
+
+import (
+	"fmt"
+	"os"
+	"text/tabwriter"
+
+	"github.com/satori/go.uuid"
+	"github.com/urfave/cli"
+
+	"github.com/fission/fission"
+	"github.com/fission/fission/mqtrigger/messageQueue"
+)
+
+func mqtCreate(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	mqtName := c.String("name")
+	if len(mqtName) == 0 {
+		mqtName = uuid.NewV4().String()
+	}
+	fnName := c.String("function")
+	if len(fnName) == 0 {
+		fatal("Need a function name to create a trigger, use --function")
+	}
+	fnUid := c.String("uid")
+	mqType := c.String("mqtype")
+	switch mqType {
+	case "":
+		mqType = messageQueue.NATS
+	case messageQueue.NATS:
+		mqType = messageQueue.NATS
+	default:
+		fatal("Unknown message queue type, currently only \"nats-streaming\" is supported")
+	}
+
+	// TODO: check topic availability
+	topic := c.String("topic")
+	if len(topic) == 0 {
+		fatal("Listen topic cannot be empty")
+	}
+	respTopic := c.String("resptopic")
+
+	if topic == respTopic {
+		fatal("Listen topic should not equal to response topic")
+	}
+
+	checkMQTopicAvailability(mqType, topic, respTopic)
+
+	fnMeta := fission.Metadata{
+		Name: fnName,
+		Uid:  fnUid,
+	}
+
+	mqt := fission.MessageQueueTrigger{
+		Metadata: fission.Metadata{
+			Name: mqtName,
+		},
+		Function:         fnMeta,
+		MessageQueueType: mqType,
+		Topic:            topic,
+		ResponseTopic:    respTopic,
+	}
+
+	_, err := client.MessageQueueTriggerCreate(&mqt)
+	checkErr(err, "create message queue trigger")
+
+	fmt.Printf("trigger '%s' created\n", mqtName)
+	return err
+}
+
+func mqtGet(c *cli.Context) error {
+	return nil
+}
+
+func mqtUpdate(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+	mqtName := c.String("name")
+	if len(mqtName) == 0 {
+		fatal("Need name of trigger, use --name")
+	}
+	topic := c.String("topic")
+	respTopic := c.String("resptopic")
+
+	mqt, err := client.MessageQueueTriggerGet(&fission.Metadata{Name: mqtName})
+	checkErr(err, "get Time trigger")
+
+	checkMQTopicAvailability(mqt.MessageQueueType, topic, respTopic)
+
+	mqt.Topic = topic
+	mqt.ResponseTopic = respTopic
+
+	_, err = client.MessageQueueTriggerUpdate(mqt)
+	checkErr(err, "update Time trigger")
+
+	fmt.Printf("trigger '%v' updated\n", mqtName)
+	return nil
+}
+
+func mqtDelete(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+	mqtName := c.String("name")
+	if len(mqtName) == 0 {
+		fatal("Need name of trigger to delete, use --name")
+	}
+
+	err := client.MessageQueueTriggerDelete(&fission.Metadata{Name: mqtName})
+	checkErr(err, "delete trigger")
+
+	fmt.Printf("trigger '%v' deleted\n", mqtName)
+	return nil
+}
+
+func mqtList(c *cli.Context) error {
+	client := getClient(c.GlobalString("server"))
+
+	mqts, err := client.MessageQueueTriggerList(c.String("mqtype"))
+	checkErr(err, "list message queue triggers")
+
+	w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
+
+	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
+		"NAME", "FUNCTION_NAME", "FUNCTION_UID", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC")
+	for _, mqt := range mqts {
+		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
+			mqt.Metadata.Name, mqt.Function.Name, mqt.Function.Uid, mqt.MessageQueueType, mqt.Topic, mqt.ResponseTopic)
+	}
+	w.Flush()
+
+	return nil
+}
+
+func checkMQTopicAvailability(mqType string, topics ...string) {
+	for _, t := range topics {
+		if len(t) > 0 && !messageQueue.IsTopicValid(mqType, t) {
+			fatal(fmt.Sprintf("Invalid topic for %s: %s", mqType, t))
+		}
+	}
+}

glide.lock

@@ -88,6 +88,21 @@ imports:
   - buffer
   - jlexer
   - jwriter
+- name: github.com/nats-io/go-nats
+  version: 6949c8e06a246e4177961aab22940b5c411e48f0
+  subpackages:
+  - encoders/builtin
+  - util
+- name: github.com/nats-io/go-nats-streaming
+  version: 6e620057a207bd61e992c1c5b6a2de7b6a4cb010
+  subpackages:
+  - pb
+- name: github.com/nats-io/nats-streaming-server
+  version: 7a922646104d066c98527959455993231fadd168
+  subpackages:
+  - util
+- name: github.com/nats-io/nuid
+  version: 289cccf02c178dc782430d534e3c1f5b72af807f
 - name: github.com/pborman/uuid
   version: 3d4f2ba23642d3cfd06bd4b54cf03d99d95c0f1b
 - name: github.com/PuerkitoBio/purell

glide.yaml

@@ -34,3 +34,7 @@ import:
   subpackages:
   - client/v2
 - package: github.com/robfig/cron
+- package: github.com/nats-io/go-nats-streaming
+  version: ^v0.3.4
+- package: github.com/nats-io/nats-streaming-server
+  version: ^v0.4.0

hack/release.sh

@@ -0,0 +1,187 @@
+#!/bin/sh
+
+set -e
+#set -x
+
+DIR=$(realpath $(dirname $0))/../
+BUILDDIR=$(realpath $DIR)/build
+
+# Ensure we're on the master branch
+check_branch() {
+    curr_branch=$(git rev-parse --abbrev-ref HEAD)
+    if $curr_branch != "master"
+    then
+	echo "Not on master branch."
+	exit 1
+    fi
+}
+
+# Ensure working dir is clean
+check_clean() {    
+    if ! git diff-index --quiet HEAD --
+    then
+	echo "Unclean tree"
+	exit 1
+    fi
+}
+
+# Build CLI binaries for mac/linux/windows
+build_all_cli() {
+    build_cli "linux" "linux"
+    build_cli "darwin" "osx"
+    build_cli "windows" "windows"
+}
+
+# Build cli binary for one OS, and put it in $BUILDDIR/cli/<os>/
+build_cli() {
+    os=$1
+    osName=$2
+    arch="amd64" # parameterize if/when we need to
+    
+    pushd $DIR/fission
+    GOOS=$os GOARCH=$arch go build .
+
+    if [ "$os" == "windows" ]
+    then
+	binary=fission.exe
+    else
+	binary=fission
+    fi
+
+    outdir=$BUILDDIR/cli/$osName/
+    mkdir -p $outdir
+    mv $binary $outdir
+    
+    popd
+}
+
+# Build fission-bundle image
+build_fission_bundle_image() {
+    version=$1
+    tag=fission/fission-bundle:$version
+
+    pushd $DIR/fission-bundle
+
+    GOOS=linux go build 
+    docker build -t $tag .
+    
+    popd
+}
+
+# Push fission-bundle image
+push_fission_bundle_image() {
+    version=$1
+    tag=fission/fission-bundle:$version
+    docker push $tag
+}
+
+#
+# Create fission.yaml
+#
+# TODO: get rid of this in favour of the helm chart
+#
+build_yaml() {
+    version=$1
+    tag=fission/fission-bundle:$version
+
+    outdir=$BUILDDIR/yaml/
+    mkdir -p $outdir
+    
+    pushd $DIR
+    
+    cat fission.yaml | sed "s#fission/fission-bundle#$tag#g" > $outdir/fission.yaml
+    cat fission-logger.yaml | sed "s#fission/fission-bundle#$tag#g" > $outdir/fission-logger.yaml
+    cat fission-openshift.yaml | sed "s#fission/fission-bundle#$tag#g" > $outdir/fission-openshift.yaml
+    cat fission-rbac.yaml | sed "s#fission/fission-bundle#$tag#g" > $outdir/fission-rbac.yaml
+    cat fission-nats.yaml | sed "s#fission/fission-bundle#$tag#g" > $outdir/fission-nats.yaml
+
+    cp fission-nodeport.yaml $outdir
+    cp fission-cloud.yaml    $outdir
+    
+    popd
+}
+
+build_all() {
+    version=$1
+    if [ -z "$version" ]
+    then
+	echo "Version unspecified"
+	exit 1
+    fi
+    
+    if [ -e $BUILDDIR ]
+    then
+	echo "Removing existing build dir ($BUILDDIR)."
+	rm -rf $BUILDDIR
+    fi
+    
+    mkdir -p $BUILDDIR
+    
+    build_fission_bundle_image $version
+    build_yaml $version
+    build_all_cli
+}
+
+make_github_release() {
+    version=$1
+    gittag=nightly$(date +%Y%m%d)
+
+    # tag the release
+    git tag $gittag
+    
+    # push tag
+    git push --tags
+
+    # create gh release
+    gothub release \
+	   --user fission \
+	   --repo fission \
+	   --tag $gittag \
+	   --name "Nightly release for $(date +%Y-%b-%d)" \
+	   --description "Nightly release for $(date +%Y-%b-%d)" \
+
+    # attach files
+
+    # cli
+    gothub upload \
+	   --user fission \
+	   --repo fission \
+	   --tag $gittag \
+	   --name fission-cli-osx \
+	   --file $BUILDDIR/cli/osx/fission
+
+    gothub upload \
+	   --user fission \
+	   --repo fission \
+	   --tag $gittag \
+	   --name fission-cli-linux \
+	   --file $BUILDDIR/cli/linux/fission
+
+    gothub upload \
+	   --user fission \
+	   --repo fission \
+	   --tag $gittag \
+	   --name fission-cli-windows.exe \
+	   --file $BUILDDIR/cli/windows/fission.exe
+
+    # yamls
+    yaml_files="fission.yaml fission-logger.yaml fission-rbac.yaml fission-openshift.yaml fission-nodeport.yaml fission-cloud.yaml fission-nats.yaml"
+    for f in $yaml_files
+    do
+	gothub upload \
+	       --user fission \
+	       --repo fission \
+	       --tag $gittag \
+	       --name $f \
+	       --file $BUILDDIR/yaml/$f
+    done
+    
+}
+
+
+# check_master
+check_clean
+version=$1
+build_all $version
+push_fission_bundle_image $version
+make_github_release $version

mqtrigger/main.go

@@ -0,0 +1,38 @@
+/*
+Copyright 2016 The Fission Authors.
+
+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 messagequeue
+
+import (
+	"os"
+
+	controllerClient "github.com/fission/fission/controller/client"
+	"github.com/fission/fission/mqtrigger/messageQueue"
+)
+
+func Start(controllerUrl string, routerUrl string) error {
+	controller := controllerClient.MakeClient(controllerUrl)
+
+	// Message queue type: nats is the only supported one for now
+	mqType := os.Getenv("MESSAGE_QUEUE_TYPE")
+	mqUrl := os.Getenv("MESSAGE_QUEUE_URL")
+	mqCfg := messageQueue.MessageQueueConfig{
+		MQType: mqType,
+		Url:    mqUrl,
+	}
+	messageQueue.MakeMessageQueueTriggerManager(controller, routerUrl, mqCfg)
+	return nil
+}

mqtrigger/messageQueue/messageQueue.go

@@ -0,0 +1,226 @@
+/*
+Copyright 2016 The Fission Authors.
+
+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 messageQueue
+
+import (
+	"errors"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/fission/fission"
+	controllerClient "github.com/fission/fission/controller/client"
+)
+
+const (
+	NATS string = "nats-streaming"
+)
+
+const (
+	ADD_TRIGGER requestType = iota
+	DELETE_TRIGGER
+	GET_ALL_TRIGGERS
+)
+
+type (
+	messageQueueSubscription interface{}
+
+	requestType int
+
+	MessageQueueConfig struct {
+		MQType string
+		Url    string
+	}
+
+	MessageQueue interface {
+		subscribe(trigger fission.MessageQueueTrigger) (messageQueueSubscription, error)
+		unsubscribe(triggerSub messageQueueSubscription) error
+	}
+
+	MessageQueueTriggerManager struct {
+		reqChan      chan request
+		mqCfg        MessageQueueConfig
+		triggers     map[string]*triggerSubscription
+		controller   *controllerClient.Client
+		messageQueue MessageQueue
+	}
+
+	triggerSubscription struct {
+		fission.Metadata
+		funcMeta     fission.Metadata
+		subscription messageQueueSubscription
+	}
+
+	request struct {
+		requestType
+		triggerSub *triggerSubscription
+		respChan   chan response
+	}
+	response struct {
+		err      error
+		triggers *map[string]messageQueueSubscription
+	}
+)
+
+func MakeMessageQueueTriggerManager(ctrlClient *controllerClient.Client,
+	routerUrl string, mqConfig MessageQueueConfig) *MessageQueueTriggerManager {
+
+	var messageQueue MessageQueue
+	var err error
+
+	mqTriggerMgr := MessageQueueTriggerManager{
+		reqChan:    make(chan request),
+		triggers:   make(map[string]*triggerSubscription),
+		controller: ctrlClient,
+	}
+	switch mqConfig.MQType {
+	case NATS:
+		messageQueue, err = makeNatsMessageQueue(routerUrl, mqConfig)
+	default:
+		err = errors.New("No matched message queue type found")
+	}
+	if err != nil {
+		log.Fatalf("Failed to connect to remote message queue server: %v", err)
+	}
+	mqTriggerMgr.messageQueue = messageQueue
+	go mqTriggerMgr.service()
+	go mqTriggerMgr.syncTriggers()
+	return &mqTriggerMgr
+}
+
+func (mqt *MessageQueueTriggerManager) service() {
+	for {
+		req := <-mqt.reqChan
+		switch req.requestType {
+		case ADD_TRIGGER:
+			var err error
+			triggerUid := req.triggerSub.Uid
+			if _, ok := mqt.triggers[triggerUid]; ok {
+				err = errors.New("Trigger already exists")
+			} else {
+				mqt.triggers[triggerUid] = req.triggerSub
+			}
+			req.respChan <- response{err: err}
+		case GET_ALL_TRIGGERS:
+			copyTriggers := make(map[string]messageQueueSubscription)
+			for key, val := range mqt.triggers {
+				copyTriggers[key] = val
+			}
+			req.respChan <- response{triggers: &copyTriggers}
+		case DELETE_TRIGGER:
+			triggerUid := req.triggerSub.Uid
+			delete(mqt.triggers, triggerUid)
+		}
+	}
+}
+
+func (mqt *MessageQueueTriggerManager) addTrigger(triggerSub *triggerSubscription) error {
+	respChan := make(chan response)
+	mqt.reqChan <- request{
+		requestType: ADD_TRIGGER,
+		triggerSub:  triggerSub,
+		respChan:    respChan,
+	}
+	r := <-respChan
+	return r.err
+}
+
+func (mqt *MessageQueueTriggerManager) getAllTriggers() *map[string]messageQueueSubscription {
+	respChan := make(chan response)
+	mqt.reqChan <- request{
+		requestType: GET_ALL_TRIGGERS,
+		respChan:    respChan,
+	}
+	r := <-respChan
+	return r.triggers
+}
+
+func (mqt *MessageQueueTriggerManager) delTrigger(triggerUid string) {
+	mqt.reqChan <- request{
+		requestType: DELETE_TRIGGER,
+		triggerSub: &triggerSubscription{
+			Metadata: fission.Metadata{
+				Uid: triggerUid,
+			},
+		},
+	}
+}
+
+func (mqt *MessageQueueTriggerManager) syncTriggers() {
+	for {
+		// TODO: handle error
+		newTriggers, err := mqt.controller.MessageQueueTriggerList(mqt.mqCfg.MQType)
+		if err != nil {
+			log.Warnf("Failed to sync message queue trigger from controller: %v", err)
+		}
+		// sync trigger from controller
+		newTriggerMap := map[string]fission.MessageQueueTrigger{}
+		for _, trigger := range newTriggers {
+			newTriggerMap[trigger.Uid] = trigger
+		}
+		currentTriggerSubMap := mqt.getAllTriggers()
+
+		// register new triggers
+		for key, trigger := range newTriggerMap {
+			if _, ok := (*currentTriggerSubMap)[key]; ok {
+				continue
+			}
+			sub, err := mqt.messageQueue.subscribe(trigger)
+			if err != nil {
+				log.Warnf("Message queue trigger %s created failed: %v", trigger.Name, err)
+				continue
+			}
+			triggerSub := triggerSubscription{
+				Metadata: fission.Metadata{
+					Name: trigger.Name,
+					Uid:  trigger.Uid,
+				},
+				funcMeta:     trigger.Function,
+				subscription: sub,
+			}
+			err = mqt.addTrigger(&triggerSub)
+			if err != nil {
+				log.Warnf("Message queue trigger %s created failed: %v", trigger.Name, err)
+				continue
+			}
+			log.Infof("Message queue trigger %s created", trigger.Name)
+		}
+
+		// remove old triggers
+		for _, ts := range *currentTriggerSubMap {
+			triggerSub := ts.(*triggerSubscription)
+			if _, ok := newTriggerMap[triggerSub.Uid]; ok {
+				continue
+			}
+			if err := mqt.messageQueue.unsubscribe(triggerSub.subscription); err != nil {
+				log.Warnf("Message queue trigger %s deleted failed: %v", triggerSub.Name, err)
+			} else {
+				mqt.delTrigger(triggerSub.Uid)
+				log.Infof("Message queue trigger %s deleted", triggerSub.Name)
+			}
+		}
+		time.Sleep(3 * time.Second)
+	}
+}
+
+func IsTopicValid(mqType string, topic string) bool {
+	switch mqType {
+	case NATS:
+		return isTopicValidForNats(topic)
+	}
+	return false
+}

mqtrigger/messageQueue/nats.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2016 The Fission Authors.
+
+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 messageQueue
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
+	ns "github.com/nats-io/go-nats-streaming"
+	nsUtil "github.com/nats-io/nats-streaming-server/util"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/fission/fission"
+)
+
+const (
+	natsClusterID  = "fissionMQTrigger"
+	natsProtocol   = "nats://"
+	natsClientID   = "fission"
+	natsQueueGroup = "fission-messageQueueNatsTrigger"
+)
+
+type (
+	Nats struct {
+		nsConn    ns.Conn
+		routerUrl string
+	}
+)
+
+func makeNatsMessageQueue(routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
+	conn, err := ns.Connect(natsClusterID, natsClientID, ns.NatsURL(mqCfg.Url))
+	if err != nil {
+		return nil, err
+	}
+	nats := Nats{
+		nsConn:    conn,
+		routerUrl: routerUrl,
+	}
+	return nats, nil
+}
+
+func (nats Nats) subscribe(trigger fission.MessageQueueTrigger) (messageQueueSubscription, error) {
+	subj := trigger.Topic
+
+	if !isTopicValidForNats(subj) {
+		return nil, errors.New(fmt.Sprintf("Not a valid topic: %s", trigger.Topic))
+	}
+
+	opts := []ns.SubscriptionOption{
+		// Create a durable subscription to nats, so that triggers could retrieve last unack message.
+		// https://github.com/nats-io/go-nats-streaming#durable-subscriptions
+		ns.DurableName(trigger.Uid),
+
+		// Nats-streaming server is auto-ack mode by default. Since we want nats-streaming server to
+		// resend a message if the trigger does not ack it, we need to enable the manual ack mode, so that
+		// trigger could choose to ack message or simply drop it depend on the response of function pod.
+		ns.SetManualAckMode(),
+	}
+	sub, err := nats.nsConn.Subscribe(subj, msgHandler(&nats, trigger), opts...)
+	if err != nil {
+		return nil, err
+	}
+	return sub, nil
+}
+
+func (nats Nats) unsubscribe(subscription messageQueueSubscription) error {
+	return subscription.(ns.Subscription).Close()
+}
+
+func isTopicValidForNats(topic string) bool {
+	// nats-streaming does not support wildcard channel.
+	return nsUtil.IsSubjectValid(topic)
+}
+
+func msgHandler(nats *Nats, trigger fission.MessageQueueTrigger) func(*ns.Msg) {
+	return func(msg *ns.Msg) {
+		url := nats.routerUrl + "/" + strings.TrimPrefix(fission.UrlForFunction(&trigger.Function), "/")
+		log.Printf("Making HTTP request to %v", url)
+
+		headers := map[string]string{
+			"X-Fission-MQTrigger-Topic":     trigger.Topic,
+			"X-Fission-MQTrigger-RespTopic": trigger.ResponseTopic,
+		}
+
+		// Create request
+		req, err := http.NewRequest("POST", url, bytes.NewReader(msg.Data))
+		for k, v := range headers {
+			req.Header.Add(k, v)
+		}
+
+		// Make the request
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			log.Warningf("Request failed: %v", url)
+			return
+		}
+		defer resp.Body.Close()
+
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			log.Warningf("Request body error: %v", string(body))
+			return
+		}
+		if resp.StatusCode != 200 {
+			log.Printf("Request returned failure: %v", resp.StatusCode)
+			return
+		}
+		// trigger acks message only if a request done successfully
+		err = msg.Ack()
+		if err != nil {
+			log.Warningf("Failed to ack message: %v", err)
+		}
+		err = nats.nsConn.Publish(trigger.ResponseTopic, body)
+		if err != nil {
+			log.Warningf("Failed to publish message to topic %s: %v", trigger.ResponseTopic, err)
+		}
+	}
+}

resource.go

@@ -28,6 +28,10 @@ func (ht HTTPTrigger) Key() string {
 	return ht.Metadata.Name
 }
 
+func (mqt MessageQueueTrigger) Key() string {
+	return mqt.Metadata.Name
+}
+
 func (tt TimeTrigger) Key() string {
 	return tt.Metadata.Name
 }

test/mqtrigger/main.js

@@ -0,0 +1,6 @@
+module.exports = async function(context) {
+    return {
+        status: 200,
+        body: "Hello, World!"
+    };
+}

test/mqtrigger/stan-pub.go

@@ -0,0 +1,112 @@
+// This file originally came from official Nats.io GitHub repository.
+// You can reach original file with the following link:
+// https://github.com/nats-io/go-nats-streaming/tree/master/examples
+
+// Copyright 2012-2016 Apcera Inc. All rights reserved.
+// +build ignore
+
+package main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/nats-io/go-nats-streaming"
+)
+
+var usageStr = `
+Usage: stan-pub [options] <subject> <message>
+
+Options:
+	-s, --server   <url>            NATS Streaming server URL(s)
+	-c, --cluster  <cluster name>   NATS Streaming cluster name
+	-id,--clientid <client ID>      NATS Streaming client ID
+	-a, --async                     Asynchronous publish mode
+`
+
+// NOTE: Use tls scheme for TLS, e.g. stan-pub -s tls://demo.nats.io:4443 foo hello
+func usage() {
+	fmt.Printf("%s\n", usageStr)
+	os.Exit(0)
+}
+
+func main() {
+	var clusterID string
+	var clientID string
+	var async bool
+	var URL string
+
+	flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
+	flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
+	flag.StringVar(&clusterID, "c", "test-cluster", "The NATS Streaming cluster ID")
+	flag.StringVar(&clusterID, "cluster", "test-cluster", "The NATS Streaming cluster ID")
+	flag.StringVar(&clientID, "id", "stan-pub", "The NATS Streaming client ID to connect with")
+	flag.StringVar(&clientID, "clientid", "stan-pub", "The NATS Streaming client ID to connect with")
+	flag.BoolVar(&async, "a", false, "Publish asynchronously")
+	flag.BoolVar(&async, "async", false, "Publish asynchronously")
+
+	log.SetFlags(0)
+	flag.Usage = usage
+	flag.Parse()
+
+	args := flag.Args()
+
+	if len(args) < 1 {
+		usage()
+	}
+
+	sc, err := stan.Connect(clusterID, clientID, stan.NatsURL(URL))
+	if err != nil {
+		log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
+	}
+	defer sc.Close()
+
+	subj, msg := args[0], []byte(args[1])
+
+	ch := make(chan bool)
+	var glock sync.Mutex
+	var guid string
+	acb := func(lguid string, err error) {
+		glock.Lock()
+		log.Printf("Received ACK for guid %s\n", lguid)
+		defer glock.Unlock()
+		if err != nil {
+			log.Fatalf("Error in server ack for guid %s: %v\n", lguid, err)
+		}
+		if lguid != guid {
+			log.Fatalf("Expected a matching guid in ack callback, got %s vs %s\n", lguid, guid)
+		}
+		ch <- true
+	}
+
+	if async != true {
+		err = sc.Publish(subj, msg)
+		if err != nil {
+			log.Fatalf("Error during publish: %v\n", err)
+		}
+		log.Printf("Published [%s] : '%s'\n", subj, msg)
+	} else {
+		glock.Lock()
+		guid, err = sc.PublishAsync(subj, msg, acb)
+		if err != nil {
+			log.Fatalf("Error during async publish: %v\n", err)
+		}
+		glock.Unlock()
+		if guid == "" {
+			log.Fatal("Expected non-empty guid to be returned.")
+		}
+		log.Printf("Published [%s] : '%s' [guid: %s]\n", subj, msg, guid)
+
+		select {
+		case <-ch:
+			break
+		case <-time.After(5 * time.Second):
+			log.Fatal("timeout")
+		}
+
+	}
+}

test/mqtrigger/stan-sub.go

@@ -0,0 +1,133 @@
+// This file originally came from official Nats.io GitHub repository.
+// You can reach original file with the following link:
+// https://github.com/nats-io/go-nats-streaming/tree/master/examples
+
+// Copyright 2012-2016 Apcera Inc. All rights reserved.
+// +build ignore
+
+package main
+
+import (
+	"flag"
+	"log"
+	"time"
+
+	"github.com/nats-io/go-nats-streaming"
+	"github.com/nats-io/go-nats-streaming/pb"
+)
+
+var usageStr = `
+Usage: stan-sub [options] <subject>
+
+Options:
+	-s, --server   <url>            NATS Streaming server URL(s)
+	-c, --cluster  <cluster name>   NATS Streaming cluster name
+	-id,--clientid <client ID>      NATS Streaming client ID
+
+Subscription Options:
+	--qgroup <name>                 Queue group
+	--seq <seqno>                   Start at seqno
+	--all                           Deliver all available messages
+	--last                          Deliver starting with last published message
+	--since <duration>              Deliver messages in last interval (e.g. 1s, 1hr)
+	         (for more information: https://golang.org/pkg/time/#ParseDuration)
+	--durable <name>                Durable subscriber name
+	--unsubscribe                   Unsubscribe the durable on exit
+`
+
+// NOTE: Use tls scheme for TLS, e.g. stan-sub -s tls://demo.nats.io:4443 foo
+func usage() {
+	log.Fatalf(usageStr)
+}
+
+func printMsg(m *stan.Msg) {
+	log.Printf("[%s]: '%s'", m.Subject, m.Data)
+}
+
+func main() {
+	var clusterID string
+	var clientID string
+	var showTime bool
+	var startSeq uint64
+	var startDelta string
+	var deliverAll bool
+	var deliverLast bool
+	var durable string
+	var qgroup string
+	var unsubscribe bool
+	var URL string
+
+	//	defaultID := fmt.Sprintf("client.%s", nuid.Next())
+
+	flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
+	flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
+	flag.StringVar(&clusterID, "c", "test-cluster", "The NATS Streaming cluster ID")
+	flag.StringVar(&clusterID, "cluster", "test-cluster", "The NATS Streaming cluster ID")
+	flag.StringVar(&clientID, "id", "", "The NATS Streaming client ID to connect with")
+	flag.StringVar(&clientID, "clientid", "", "The NATS Streaming client ID to connect with")
+	flag.BoolVar(&showTime, "t", false, "Display timestamps")
+	// Subscription options
+	flag.Uint64Var(&startSeq, "seq", 0, "Start at sequence no.")
+	flag.BoolVar(&deliverAll, "all", false, "Deliver all")
+	flag.BoolVar(&deliverLast, "last", false, "Start with last value")
+	flag.StringVar(&startDelta, "since", "", "Deliver messages since specified time offset")
+	flag.StringVar(&durable, "durable", "", "Durable subscriber name")
+	flag.StringVar(&qgroup, "qgroup", "", "Queue group name")
+	flag.BoolVar(&unsubscribe, "unsubscribe", false, "Unsubscribe the durable on exit")
+
+	log.SetFlags(0)
+	flag.Usage = usage
+	flag.Parse()
+
+	args := flag.Args()
+
+	if clientID == "" {
+		log.Printf("Error: A unique client ID must be specified.")
+		usage()
+	}
+	if len(args) < 1 {
+		log.Printf("Error: A subject must be specified.")
+		usage()
+	}
+
+	sc, err := stan.Connect(clusterID, clientID, stan.NatsURL(URL))
+	if err != nil {
+		log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
+	}
+	// log.Printf("Connected to %s clusterID: [%s] clientID: [%s]\n", URL, clusterID, clientID)
+
+	subj := args[0]
+
+	exit := make(chan struct{})
+	mcb := func(msg *stan.Msg) {
+		printMsg(msg)
+		exit <- struct{}{}
+	}
+
+	startOpt := stan.StartAt(pb.StartPosition_NewOnly)
+
+	if startSeq != 0 {
+		startOpt = stan.StartAtSequence(startSeq)
+	} else if deliverLast == true {
+		startOpt = stan.StartWithLastReceived()
+	} else if deliverAll == true {
+		log.Print("subscribing with DeliverAllAvailable")
+		startOpt = stan.DeliverAllAvailable()
+	} else if startDelta != "" {
+		ago, err := time.ParseDuration(startDelta)
+		if err != nil {
+			sc.Close()
+			log.Fatal(err)
+		}
+		startOpt = stan.StartAtTimeDelta(ago)
+	}
+
+	sub, err := sc.QueueSubscribe(subj, qgroup, mcb, startOpt, stan.DurableName(durable))
+	if err != nil {
+		sc.Close()
+		log.Fatal(err)
+	}
+
+	<-exit
+	sub.Unsubscribe()
+}

test/mqtrigger/test.sh

@@ -0,0 +1,45 @@
+#!/bin/bash
+
+set -e
+
+clusterID="fissionMQTrigger"
+topic="foo.bar"
+resptopic="foo.foo"
+expectedRespOutput="[foo.foo]: 'Hello, World!'"
+FISSIONDIR=$GOPATH"/src/github.com/fission/fission"
+
+if [[ -z $NATS_STREAMING_URL ]]; then
+    echo "'NATS_STREAMING_URL' must not be empty. For example: export NATS_STREAMING_URL=nats://192.168.0.1:4222"
+    exit 1
+fi
+
+if [[ -z $FISSION_URL ]]; then
+    echo "'FISSION_URL' must not be empty. For example: export FISSION_URL=http://10.10.10.10"
+    exit 1
+fi
+
+cd $FISSIONDIR"/fission/"
+go build
+mv fission $FISSIONDIR"/test/mqtrigger"
+cd $FISSIONDIR"/test/mqtrigger"
+
+./fission env create --name nodejs --image fission/node-env
+./fission fn create --name hello1 --env nodejs --code main.js --method GET
+./fission route create --method GET --url /h1 --function hello1
+./fission mqtrigger create --name h1 --function hello1 --mqtype "nats-streaming" --topic "foo.bar" --resptopic "foo.foo"
+
+# wait until nats trigger is created
+sleep 5
+
+go run ./stan-pub.go -s $NATS_STREAMING_URL -c $clusterID -id clientPub $topic "" || exit 1
+
+response=$(go run ./stan-sub.go --last -s $NATS_STREAMING_URL -c $clusterID -id clientSub $resptopic 2>&1)
+
+if [[ "$response" != "$expectedRespOutput" ]]; then
+    echo "$response is not equal to $expectedRespOutput"
+    exit 1
+fi
+
+echo "Subscriber received expected response: $response"
+
+exit 0

types.go

@@ -53,6 +53,14 @@ type (
 		Function   Metadata `json:"function"`
 	}
 
+	MessageQueueTrigger struct {
+		Metadata         `json:"metadata"`
+		Function         Metadata `json:"function"`
+		MessageQueueType string   `json:"messageQueueType"`
+		Topic            string   `json:"topic"`
+		ResponseTopic    string   `json:"respTopic,omitempty"`
+	}
+
 	// Watch is a specification of Kubernetes watch along with a URL to post events to.
 	Watch struct {
 		Metadata `json:"metadata"`