Move packages to proejct/pkg to follow go project folder structure convention (#1190)

pkg/mqtrigger/messageQueue/asq.go

@@ -0,0 +1,397 @@
+/*
+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 messageQueue
+
+import (
+	"bytes"
+	"encoding/base64"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/storage"
+	"github.com/fission/fission/pkg/types"
+	"github.com/fission/fission/pkg/utils"
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+)
+
+// TODO: some of these constants should probably be environment variables
+const (
+	// AzureQueuePollingInterval is the polling interval (default is 1 minute).
+	AzureQueuePollingInterval = time.Minute
+	// AzureQueueRetryLimit is the limit for attempts to retry invoking a function.
+	AzureQueueRetryLimit = 3
+	// AzureMessageFetchCount is the number of messages to fetch at a time.
+	AzureMessageFetchCount = 10
+	// AzureMessageVisibilityTimeout is the visibility timeout for dequeued messages.
+	AzureMessageVisibilityTimeout = time.Minute
+	// AzurePoisonQueueSuffix is the suffix used for poison queues.
+	AzurePoisonQueueSuffix = "-poison"
+	// AzureFunctionInvocationTimeout is the amount of time to wait for a triggered function to execute.
+	AzureFunctionInvocationTimeout = 10 * time.Minute
+)
+
+// AzureStorageConnection represents an Azure storage connection.
+type AzureStorageConnection struct {
+	logger     *zap.Logger
+	routerURL  string
+	service    AzureQueueService
+	httpClient AzureHTTPClient
+}
+
+// AzureQueueSubscription represents an Azure storage message queue subscription.
+type AzureQueueSubscription struct {
+	queue           AzureQueue
+	queueName       string
+	outputQueueName string
+	functionURL     string
+	contentType     string
+	unsubscribe     chan bool
+	done            chan bool
+}
+
+// AzureQueueService is the interface that abstracts the Azure storage service.
+// This exists to enable unit testing.
+type AzureQueueService interface {
+	GetQueue(name string) AzureQueue
+}
+
+// AzureQueue is the interface that abstracts Azure storage queues.
+// This exists to enable unit testing.
+type AzureQueue interface {
+	Create(options *storage.QueueServiceOptions) error
+	NewMessage(text string) AzureMessage
+	GetMessages(options *storage.GetMessagesOptions) ([]AzureMessage, error)
+}
+
+// AzureMessage is the interface that abstracts Azure storage messages.
+// This exists to enable unit testing.
+type AzureMessage interface {
+	Bytes() []byte
+	Put(options *storage.PutMessageOptions) error
+	Delete(options *storage.QueueServiceOptions) error
+}
+
+// AzureHTTPClient is the interface that abstract HTTP requests made by the trigger.
+// This exists to enable unit testing.
+type AzureHTTPClient interface {
+	Do(req *http.Request) (*http.Response, error)
+}
+
+type azureQueueService struct {
+	service storage.QueueServiceClient
+}
+
+func (qs azureQueueService) GetQueue(name string) AzureQueue {
+	return azureQueue{
+		ref: qs.service.GetQueueReference(name),
+	}
+}
+
+type azureQueue struct {
+	ref *storage.Queue
+}
+
+func (qr azureQueue) Create(options *storage.QueueServiceOptions) error {
+	exists, err := qr.ref.Exists()
+	if err != nil {
+		return err
+	}
+	if exists {
+		return nil
+	}
+	return qr.ref.Create(options)
+}
+
+func (qr azureQueue) NewMessage(text string) AzureMessage {
+	return azureMessage{
+		ref:   qr.ref.GetMessageReference(text),
+		bytes: []byte(text),
+	}
+}
+
+func (qr azureQueue) GetMessages(options *storage.GetMessagesOptions) ([]AzureMessage, error) {
+	msgs, err := qr.ref.GetMessages(options)
+	if err != nil {
+		return nil, err
+	}
+	messages := make([]AzureMessage, len(msgs))
+	for i := range msgs {
+		bytes, err := base64.StdEncoding.DecodeString(msgs[i].Text)
+		if err != nil {
+			return nil, err
+		}
+		messages[i] = azureMessage{
+			ref:   &msgs[i],
+			bytes: bytes,
+		}
+	}
+	return messages, nil
+}
+
+type azureMessage struct {
+	ref   *storage.Message
+	bytes []byte
+}
+
+func (m azureMessage) Bytes() []byte {
+	return m.bytes
+}
+
+func (m azureMessage) Put(options *storage.PutMessageOptions) error {
+	return m.ref.Put(options)
+}
+
+func (m azureMessage) Delete(options *storage.QueueServiceOptions) error {
+	return m.ref.Delete(options)
+}
+
+func newAzureQueueService(client storage.Client) AzureQueueService {
+	return azureQueueService{
+		service: client.GetQueueService(),
+	}
+}
+
+func newAzureStorageConnection(logger *zap.Logger, routerURL string, config MessageQueueConfig) (MessageQueue, error) {
+	account := os.Getenv("AZURE_STORAGE_ACCOUNT_NAME")
+	if len(account) == 0 {
+		return nil, errors.New("Required environment variable 'AZURE_STORAGE_ACCOUNT_NAME' is not set")
+	}
+
+	key := os.Getenv("AZURE_STORAGE_ACCOUNT_KEY")
+	if len(key) == 0 {
+		return nil, errors.New("Required environment variable 'AZURE_STORAGE_ACCOUNT_KEY' is not set")
+	}
+
+	logger.Info("creating Azure storage connection to storage account", zap.String("account", account))
+
+	client, err := storage.NewBasicClient(account, key)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create Azure storage client")
+	}
+	return &AzureStorageConnection{
+		logger:    logger.Named("azue_storage"),
+		routerURL: routerURL,
+		service:   newAzureQueueService(client),
+		httpClient: &http.Client{
+			Timeout: AzureFunctionInvocationTimeout,
+		},
+	}, nil
+}
+
+func (asc AzureStorageConnection) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubscription, error) {
+	asc.logger.Info("subscribing to Azure storage queue", zap.String("queue", trigger.Spec.Topic))
+
+	if trigger.Spec.FunctionReference.Type != types.FunctionReferenceTypeFunctionName {
+		return nil, fmt.Errorf("unsupported function reference type (%v) for trigger %q", trigger.Spec.FunctionReference.Type, trigger.Metadata.Name)
+	}
+
+	subscription := &AzureQueueSubscription{
+		queue:           asc.service.GetQueue(trigger.Spec.Topic),
+		queueName:       trigger.Spec.Topic,
+		outputQueueName: trigger.Spec.ResponseTopic,
+		// with the addition of multi-tenancy, the users can create functions in any namespace. however,
+		// the triggers can only be created in the same namespace as the function.
+		// so essentially, function namespace = trigger namespace.
+		functionURL: asc.routerURL + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/"),
+		contentType: trigger.Spec.ContentType,
+		unsubscribe: make(chan bool),
+		done:        make(chan bool),
+	}
+
+	go runAzureQueueSubscription(asc, subscription)
+	return subscription, nil
+}
+
+func (asc AzureStorageConnection) unsubscribe(subscription messageQueueSubscription) error {
+	sub := subscription.(*AzureQueueSubscription)
+
+	asc.logger.Info("unsubscribing from Azure storage queue", zap.String("queue", sub.queueName))
+
+	// Let the worker know we've unsubscribed
+	sub.unsubscribe <- true
+
+	// Wait until the subscription is done
+	<-sub.done
+	return nil
+}
+
+func runAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubscription) {
+	var wg sync.WaitGroup
+
+	// Process the queue before waiting
+	pollAzureQueueSubscription(conn, sub, &wg)
+
+	timer := time.NewTimer(AzureQueuePollingInterval)
+
+	for {
+		conn.logger.Info("waiting before polling Azure storage queue", zap.Duration("interval_length", AzureQueuePollingInterval), zap.String("queue", sub.queueName))
+		select {
+		case <-sub.unsubscribe:
+			timer.Stop()
+			wg.Wait()
+			sub.done <- true
+			return
+		case <-timer.C:
+			pollAzureQueueSubscription(conn, sub, &wg)
+			timer.Reset(AzureQueuePollingInterval)
+			continue
+		}
+	}
+}
+
+func pollAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubscription, wg *sync.WaitGroup) {
+	conn.logger.Info("polling for messages from Azure storage queue", zap.String("queue", sub.queueName))
+
+	err := sub.queue.Create(nil)
+	if err != nil {
+		conn.logger.Error("failed to create message queue", zap.Error(err), zap.String("queue", sub.queueName))
+		return
+	}
+
+	for {
+		err := sub.queue.Create(nil)
+		if err != nil {
+			conn.logger.Error("failed to create message queue", zap.Error(err), zap.String("queue", sub.queueName))
+			return
+		}
+
+		messages, err := sub.queue.GetMessages(&storage.GetMessagesOptions{
+			NumOfMessages:     AzureMessageFetchCount,
+			VisibilityTimeout: int(AzureMessageVisibilityTimeout / time.Second),
+		})
+		if err != nil {
+			conn.logger.Error("failed to retrieve messages from Azure storage queue", zap.Error(err), zap.String("queue", sub.queueName))
+			break
+		}
+		if len(messages) == 0 {
+			break
+		}
+
+		wg.Add(len(messages))
+		for _, msg := range messages {
+			go func(conn AzureStorageConnection, sub *AzureQueueSubscription, msg AzureMessage) {
+				defer wg.Done()
+				invokeTriggeredFunction(conn, sub, msg)
+			}(conn, sub, msg)
+		}
+	}
+}
+
+func invokeTriggeredFunction(conn AzureStorageConnection, sub *AzureQueueSubscription, message AzureMessage) {
+	defer message.Delete(nil)
+
+	conn.logger.Info("making HTTP request to invoke function", zap.String("function_url", sub.functionURL))
+
+	for i := 0; i <= AzureQueueRetryLimit; i++ {
+		if i > 0 {
+			conn.logger.Info("retrying function invocation", zap.Int("retry", i), zap.String("function_url", sub.functionURL))
+		}
+		request, err := http.NewRequest("POST", sub.functionURL, bytes.NewReader(message.Bytes()))
+		if err != nil {
+			conn.logger.Error("failed to create HTTP request to invoke function", zap.Error(err), zap.String("function_url", sub.functionURL))
+			continue
+		}
+
+		request.Header.Set("X-Fission-MQTrigger-Topic", sub.queueName)
+		if len(sub.outputQueueName) > 0 {
+			request.Header.Set("X-Fission-MQTrigger-RespTopic", sub.outputQueueName)
+		}
+		if i > 0 {
+			request.Header.Set("X-Fission-MQTrigger-RetryCount", strconv.Itoa(i))
+		}
+		request.Header.Set("Content-Type", sub.contentType)
+
+		response, err := conn.httpClient.Do(request)
+		if err != nil {
+			conn.logger.Error("sending function invocation request failed", zap.Error(err), zap.String("function_url", sub.functionURL))
+			continue
+		}
+		defer response.Body.Close()
+
+		body, err := ioutil.ReadAll(response.Body)
+		if err != nil {
+			conn.logger.Error("failed to read response body from function invocation", zap.Error(err), zap.String("function_url", sub.functionURL))
+			continue
+		}
+
+		if response.StatusCode < 200 || response.StatusCode >= 300 {
+			conn.logger.Error("function invocation request returned a failure status code",
+				zap.String("function_url", sub.functionURL),
+				zap.String("body", string(body)),
+				zap.Int("status_code", response.StatusCode))
+			continue
+		}
+
+		if len(sub.outputQueueName) > 0 {
+			outputQueue := conn.service.GetQueue(sub.outputQueueName)
+			err = outputQueue.Create(nil)
+			if err != nil {
+				conn.logger.Error("failed to create output queue",
+					zap.Error(err),
+					zap.String("output_queue", sub.outputQueueName),
+					zap.String("function_url", sub.functionURL))
+				return
+			}
+
+			outputMessage := outputQueue.NewMessage(string(body))
+			err = outputMessage.Put(nil)
+			if err != nil {
+				conn.logger.Error("failed to post response body from function invocation to output queue",
+					zap.String("output_queue", sub.outputQueueName),
+					zap.String("function_url", sub.functionURL))
+				return
+			}
+		}
+
+		// Function invocation was successful
+		return
+	}
+
+	conn.logger.Error("function invocation retired too many times - moving message to poison queue",
+		zap.Int("retry_limit", AzureQueueRetryLimit),
+		zap.String("function_url", sub.functionURL))
+
+	poisonQueueName := sub.queueName + AzurePoisonQueueSuffix
+	poisonQueue := conn.service.GetQueue(poisonQueueName)
+	err := poisonQueue.Create(nil)
+	if err != nil {
+		conn.logger.Error("failed to create poison queue",
+			zap.Error(err),
+			zap.String("poison_queue_name", poisonQueueName),
+			zap.String("function_url", sub.functionURL))
+		return
+	}
+
+	poisonMessage := poisonQueue.NewMessage(string(message.Bytes()))
+	err = poisonMessage.Put(nil)
+	if err != nil {
+		conn.logger.Error("failed to post response body from function invocation failure poison queue",
+			zap.Error(err),
+			zap.String("poison_queue_name", poisonQueueName),
+			zap.String("function_url", sub.functionURL))
+		return
+	}
+}

pkg/mqtrigger/messageQueue/asq_test.go

@@ -0,0 +1,475 @@
+/*
+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 messageQueue
+
+import (
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/storage"
+	"github.com/fission/fission/pkg/types"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+)
+
+const (
+	DummyRouterURL = "http://localhost"
+)
+
+func panicIf(err error) {
+	if err != nil {
+		log.Panicf("Error: %v", err)
+	}
+}
+
+type azureQueueServiceMock struct {
+	mock.Mock
+}
+
+func (m *azureQueueServiceMock) GetQueue(name string) AzureQueue {
+	args := m.Called(name)
+	return args.Get(0).(AzureQueue)
+}
+
+type azureQueueMock struct {
+	mock.Mock
+}
+
+func (m *azureQueueMock) Create(options *storage.QueueServiceOptions) error {
+	args := m.Called(options)
+	return args.Error(0)
+}
+
+func (m *azureQueueMock) NewMessage(text string) AzureMessage {
+	args := m.Called(text)
+	return args.Get(0).(AzureMessage)
+}
+
+func (m *azureQueueMock) GetMessages(options *storage.GetMessagesOptions) ([]AzureMessage, error) {
+	args := m.Called(options)
+	return args.Get(0).([]AzureMessage), args.Error(1)
+}
+
+type azureMessageMock struct {
+	mock.Mock
+}
+
+func (m *azureMessageMock) Bytes() []byte {
+	args := m.Called()
+	return args.Get(0).([]byte)
+}
+
+func (m *azureMessageMock) Put(options *storage.PutMessageOptions) error {
+	args := m.Called(options)
+	return args.Error(0)
+}
+
+func (m *azureMessageMock) Delete(options *storage.QueueServiceOptions) error {
+	args := m.Called(options)
+	return args.Error(0)
+}
+
+type azureHTTPClientMock struct {
+	mock.Mock
+	bodyHandler func(res *http.Response)
+}
+
+func (m *azureHTTPClientMock) Do(req *http.Request) (*http.Response, error) {
+	args := m.Called(req)
+
+	res := args.Get(0).(*http.Response)
+	err := args.Error(1)
+
+	if res != nil && m.bodyHandler != nil {
+		m.bodyHandler(res)
+	}
+	return res, err
+}
+
+func TestNewStorageConnectionMissingAccountName(t *testing.T) {
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	connection, err := newAzureStorageConnection(logger, DummyRouterURL, MessageQueueConfig{
+		MQType: types.MessageQueueTypeASQ,
+		Url:    "",
+	})
+	require.Nil(t, connection)
+	require.Error(t, err, "Required environment variable 'AZURE_STORAGE_ACCOUNT_NAME' is not set")
+}
+
+func TestNewStorageConnectionMissingAccessKey(t *testing.T) {
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	_ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname")
+	connection, err := newAzureStorageConnection(logger, DummyRouterURL, MessageQueueConfig{
+		MQType: types.MessageQueueTypeASQ,
+		Url:    "",
+	})
+	_ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_NAME")
+	require.Nil(t, connection)
+	require.Error(t, err, "Required environment variable 'AZURE_STORAGE_ACCOUNT_KEY' is not set")
+}
+
+func TestNewStorageConnection(t *testing.T) {
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	_ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname")
+	_ = os.Setenv("AZURE_STORAGE_ACCOUNT_KEY", "bm90IGEga2V5")
+	connection, err := newAzureStorageConnection(logger, DummyRouterURL, MessageQueueConfig{
+		MQType: "azure-storage-queue",
+		Url:    "",
+	})
+	_ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_NAME")
+	_ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_KEY")
+	require.NoError(t, err)
+	require.IsType(t, &AzureStorageConnection{}, connection)
+
+	p := connection.(*AzureStorageConnection)
+	require.Equal(t, DummyRouterURL, p.routerURL)
+	require.NotNil(t, p.service)
+}
+
+func TestAzureStorageQueueSingleMessage(t *testing.T) {
+	runAzureStorageQueueTest(t, 1, false)
+}
+
+func TestAzureStorageQueueMultipleMessages(t *testing.T) {
+	runAzureStorageQueueTest(t, 10, false)
+}
+
+func TestAzureStorageQueueSingleOutputMessage(t *testing.T) {
+	runAzureStorageQueueTest(t, 1, true)
+}
+
+func TestAzureStorageQueueMultipleOutputMessages(t *testing.T) {
+	runAzureStorageQueueTest(t, 10, true)
+}
+
+func TestAzureStorageQueuePoisonMessage(t *testing.T) {
+	const (
+		TriggerName  = "queuetrigger"
+		QueueName    = "inputqueue"
+		MessageBody  = "input"
+		FunctionName = "badfunc"
+		ContentType  = "text/plain"
+	)
+
+	// Mock a HTTP client that returns different failures
+	httpClient := new(azureHTTPClientMock)
+	httpClient.On(
+		"Do",
+		mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "", ContentType, FunctionName, MessageBody)),
+	).Return(
+		&http.Response{
+			StatusCode: http.StatusInternalServerError,
+			Body:       ioutil.NopCloser(strings.NewReader("server error")),
+		},
+		nil,
+	).Once()
+	httpClient.On(
+		"Do",
+		mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "1", ContentType, FunctionName, MessageBody)),
+	).Return(
+		&http.Response{
+			StatusCode: http.StatusNotFound,
+			Body:       ioutil.NopCloser(strings.NewReader("not found")),
+		},
+		nil,
+	).Once()
+	httpClient.On(
+		"Do",
+		mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "2", ContentType, FunctionName, MessageBody)),
+	).Return(
+		&http.Response{
+			StatusCode: http.StatusBadRequest,
+			Body:       ioutil.NopCloser(strings.NewReader("bad request")),
+		},
+		nil,
+	).Once()
+	httpClient.On(
+		"Do",
+		mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "3", ContentType, FunctionName, MessageBody)),
+	).Return(
+		&http.Response{
+			StatusCode: http.StatusForbidden,
+			Body:       ioutil.NopCloser(strings.NewReader("not authorized")),
+		},
+		nil,
+	).Once()
+
+	// Mock a queue message with "input" as the message body
+	message := new(azureMessageMock)
+	message.On("Bytes").Return([]byte(MessageBody))
+	message.On(
+		"Delete",
+		mock.MatchedBy(
+			func(options *storage.QueueServiceOptions) bool {
+				return options == nil
+			},
+		),
+	).Return(nil)
+
+	// Mock a queue that performs a no-op create, returns a "poison" message, and then returns no more messages
+	queue := new(azureQueueMock)
+	queue.On(
+		"Create",
+		mock.MatchedBy(
+			func(options *storage.QueueServiceOptions) bool {
+				return options == nil
+			},
+		),
+	).Return(nil)
+	queue.On(
+		"GetMessages",
+		mock.MatchedBy(
+			func(options *storage.GetMessagesOptions) bool {
+				return options.NumOfMessages == AzureMessageFetchCount &&
+					options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second)
+			},
+		),
+	).Return([]AzureMessage{message}, nil).Once()
+	queue.On(
+		"GetMessages",
+		mock.MatchedBy(
+			func(options *storage.GetMessagesOptions) bool {
+				return options.NumOfMessages == AzureMessageFetchCount &&
+					options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second)
+			},
+		),
+	).Return([]AzureMessage{}, nil)
+
+	// Mock a poison queue message that performs a no-op Put
+	poisonMessage := new(azureMessageMock)
+	poisonMessage.On(
+		"Put",
+		mock.MatchedBy(
+			func(options *storage.PutMessageOptions) bool {
+				return options == nil
+			},
+		),
+	).Return(nil)
+
+	// Mock a poison queue that performs a no-op create and creates a new message
+	poisonQueue := new(azureQueueMock)
+	poisonQueue.On(
+		"Create",
+		mock.MatchedBy(
+			func(options *storage.QueueServiceOptions) bool {
+				return options == nil
+			},
+		),
+	).Return(nil)
+	poisonQueue.On("NewMessage", MessageBody).Return(poisonMessage).Once()
+
+	// Mock the queue service to return the input queue
+	service := new(azureQueueServiceMock)
+	service.On("GetQueue", QueueName).Return(queue).Once()
+	service.On("GetQueue", QueueName+AzurePoisonQueueSuffix).Return(poisonQueue).Once()
+
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	// Create the storage connection and subscribe to the trigger
+	connection := AzureStorageConnection{
+		logger:     logger,
+		routerURL:  DummyRouterURL,
+		service:    service,
+		httpClient: httpClient,
+	}
+	subscription, err := connection.subscribe(&fv1.MessageQueueTrigger{
+		Metadata: metav1.ObjectMeta{
+			Name:      TriggerName,
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: fv1.MessageQueueTriggerSpec{
+			FunctionReference: fv1.FunctionReference{
+				Type: types.FunctionReferenceTypeFunctionName,
+				Name: FunctionName,
+			},
+			MessageQueueType: types.MessageQueueTypeASQ,
+			Topic:            QueueName,
+			ContentType:      ContentType,
+		},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, subscription)
+
+	connection.unsubscribe(subscription)
+
+	mock.AssertExpectationsForObjects(t, httpClient, message, poisonMessage, queue, poisonQueue, service)
+}
+
+func httpRequestMatcher(t *testing.T, queue string, responseQueue string, retry string, contentType string, functionName string, body string) func(*http.Request) bool {
+	expectedURL := fmt.Sprintf("%s/fission-function/%s", DummyRouterURL, functionName)
+	return func(req *http.Request) bool {
+		requestBody, err := ioutil.ReadAll(req.Body)
+		require.NoError(t, err)
+
+		req.Body = ioutil.NopCloser(strings.NewReader(string(requestBody)))
+
+		return queue == req.Header.Get("X-Fission-MQTrigger-Topic") &&
+			responseQueue == req.Header.Get("X-Fission-MQTrigger-RespTopic") &&
+			retry == req.Header.Get("X-Fission-MQTrigger-RetryCount") &&
+			contentType == req.Header.Get("Content-Type") &&
+			req.URL.String() == expectedURL &&
+			string(requestBody) == body
+	}
+}
+
+func runAzureStorageQueueTest(t *testing.T, count int, output bool) {
+	const (
+		TriggerName      = "queuetrigger"
+		QueueName        = "inputqueue"
+		OutputQueueName  = "outputqueue"
+		MessageBody      = "input"
+		FunctionName     = "testfunc"
+		FunctionResponse = "output"
+		ContentType      = "text/plain"
+	)
+
+	responseTopic := ""
+	if output {
+		responseTopic = OutputQueueName
+	}
+
+	// Mock a HTTP client that returns http.StatusOK with "output" for the body
+	httpClient := new(azureHTTPClientMock)
+	httpClient.bodyHandler = func(res *http.Response) {
+		res.Body = ioutil.NopCloser(strings.NewReader(FunctionResponse))
+	}
+	httpClient.On(
+		"Do",
+		mock.MatchedBy(httpRequestMatcher(t, QueueName, responseTopic, "", ContentType, FunctionName, MessageBody)),
+	).Return(&http.Response{StatusCode: http.StatusOK}, nil).Times(count)
+
+	// Mock a queue message with "input" as the message body
+	message := new(azureMessageMock)
+	message.On("Bytes").Return([]byte(MessageBody)).Times(count)
+	message.On(
+		"Delete",
+		mock.MatchedBy(
+			func(options *storage.QueueServiceOptions) bool {
+				return options == nil
+			},
+		),
+	).Return(nil).Times(count)
+
+	// Mock a queue that performs a no-op create, returns a message the specified number of times, and then returns no more messages
+	queue := new(azureQueueMock)
+	queue.On(
+		"Create",
+		mock.MatchedBy(
+			func(options *storage.QueueServiceOptions) bool {
+				return options == nil
+			},
+		),
+	).Return(nil)
+	queue.On(
+		"GetMessages",
+		mock.MatchedBy(
+			func(options *storage.GetMessagesOptions) bool {
+				return options.NumOfMessages == AzureMessageFetchCount &&
+					options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second)
+			},
+		),
+	).Return([]AzureMessage{message}, nil).Times(count)
+	queue.On(
+		"GetMessages",
+		mock.MatchedBy(
+			func(options *storage.GetMessagesOptions) bool {
+				return options.NumOfMessages == AzureMessageFetchCount &&
+					options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second)
+			},
+		),
+	).Return([]AzureMessage{}, nil)
+
+	// Mock the output queue if needed
+	outputMessage := new(azureMessageMock)
+	outputQueue := new(azureQueueMock)
+	if output {
+		outputMessage.On(
+			"Put",
+			mock.MatchedBy(
+				func(options *storage.PutMessageOptions) bool {
+					return options == nil
+				},
+			),
+		).Return(nil).Times(count)
+
+		outputQueue.On(
+			"Create",
+			mock.MatchedBy(
+				func(options *storage.QueueServiceOptions) bool {
+					return options == nil
+				},
+			),
+		).Return(nil).Times(count)
+		outputQueue.On("NewMessage", FunctionResponse).Return(outputMessage).Times(count)
+	}
+
+	// Mock the queue service to return the input queue
+	service := new(azureQueueServiceMock)
+	service.On("GetQueue", QueueName).Return(queue).Once()
+	if output {
+		service.On("GetQueue", OutputQueueName).Return(outputQueue).Times(count)
+	}
+
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	// Create the storage connection and subscribe to the trigger
+	connection := AzureStorageConnection{
+		logger:     logger,
+		routerURL:  DummyRouterURL,
+		service:    service,
+		httpClient: httpClient,
+	}
+	subscription, err := connection.subscribe(&fv1.MessageQueueTrigger{
+		Metadata: metav1.ObjectMeta{
+			Name:      TriggerName,
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: fv1.MessageQueueTriggerSpec{
+			FunctionReference: fv1.FunctionReference{
+				Type: types.FunctionReferenceTypeFunctionName,
+				Name: FunctionName,
+			},
+			MessageQueueType: types.MessageQueueTypeASQ,
+			Topic:            QueueName,
+			ResponseTopic:    responseTopic,
+			ContentType:      ContentType,
+		},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, subscription)
+
+	connection.unsubscribe(subscription)
+
+	mock.AssertExpectationsForObjects(t, httpClient, message, outputMessage, queue, outputQueue, service)
+}

pkg/mqtrigger/messageQueue/kafka.go

@@ -0,0 +1,264 @@
+/*
+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"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"strings"
+
+	sarama "github.com/Shopify/sarama"
+	cluster "github.com/bsm/sarama-cluster"
+	"github.com/fission/fission/pkg/types"
+	"github.com/fission/fission/pkg/utils"
+	"go.uber.org/zap"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+)
+
+type (
+	Kafka struct {
+		logger    *zap.Logger
+		routerUrl string
+		brokers   []string
+		version   sarama.KafkaVersion
+	}
+)
+
+func makeKafkaMessageQueue(logger *zap.Logger, routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
+	if len(routerUrl) == 0 || len(mqCfg.Url) == 0 {
+		return nil, errors.New("the router URL or MQ URL is empty")
+	}
+	mqKafkaVersion := os.Getenv("MESSAGE_QUEUE_KAFKA_VERSION")
+
+	// Parse version string
+	kafkaVersion, err := sarama.ParseKafkaVersion(mqKafkaVersion)
+	if err != nil {
+		logger.Warn("error parsing kafka version string - falling back to default",
+			zap.Error(err),
+			zap.String("failed_version", mqKafkaVersion),
+			zap.Any("default_version", kafkaVersion))
+	}
+
+	kafka := Kafka{
+		logger:    logger.Named("kafka"),
+		routerUrl: routerUrl,
+		brokers:   strings.Split(mqCfg.Url, ","),
+		version:   kafkaVersion,
+	}
+	logger.Info("created kafka queue", zap.Any("kafka", kafka))
+	return kafka, nil
+}
+
+func isTopicValidForKafka(topic string) bool {
+	return true
+}
+
+func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubscription, error) {
+	kafka.logger.Info("inside kakfa subscribe", zap.Any("trigger", trigger))
+	kafka.logger.Info("brokers set", zap.Strings("brokers", kafka.brokers))
+
+	// Create new consumer
+	consumerConfig := cluster.NewConfig()
+	consumerConfig.Consumer.Return.Errors = true
+	consumerConfig.Group.Return.Notifications = true
+	consumerConfig.Config.Version = kafka.version
+	consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.Metadata.UID), []string{trigger.Spec.Topic}, consumerConfig)
+	kafka.logger.Info("created a new consumer", zap.Any("consumer", consumer))
+	if err != nil {
+		panic(err)
+	}
+
+	// Create new producer
+	producerConfig := sarama.NewConfig()
+	producerConfig.Producer.RequiredAcks = sarama.WaitForAll
+	producerConfig.Producer.Retry.Max = 10
+	producerConfig.Producer.Return.Successes = true
+	producerConfig.Version = kafka.version
+	producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig)
+	kafka.logger.Info("created a new producer", zap.Any("consumer", producer))
+	if err != nil {
+		panic(err)
+	}
+
+	// consume errors
+	go func() {
+		for err := range consumer.Errors() {
+			kafka.logger.Error("consumer error", zap.Error(err))
+		}
+	}()
+
+	// consume notifications
+	go func() {
+		for ntf := range consumer.Notifications() {
+			kafka.logger.Info("consumer notification", zap.Any("notification", ntf))
+		}
+	}()
+
+	// consume messages
+	go func() {
+		for msg := range consumer.Messages() {
+			kafka.logger.Info("calling message handler", zap.String("message", string(msg.Value[:])))
+			if kafkaMsgHandler(&kafka, producer, trigger, msg) {
+				consumer.MarkOffset(msg, "") // mark message as processed
+			}
+		}
+	}()
+
+	return consumer, nil
+}
+
+func (kafka Kafka) unsubscribe(subscription messageQueueSubscription) error {
+	return subscription.(*cluster.Consumer).Close()
+}
+
+func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *fv1.MessageQueueTrigger, msg *sarama.ConsumerMessage) bool {
+	var value string = string(msg.Value[:])
+	// Support other function ref types
+	if trigger.Spec.FunctionReference.Type != types.FunctionReferenceTypeFunctionName {
+		kafka.logger.Fatal("unsupported function reference type for trigger",
+			zap.Any("function_reference_type", trigger.Spec.FunctionReference.Type),
+			zap.String("trigger", trigger.Metadata.Name))
+	}
+
+	url := kafka.routerUrl + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/")
+	kafka.logger.Info("making HTTP request", zap.String("url", url))
+
+	// Generate the Headers
+	fissionHeaders := map[string]string{
+		"X-Fission-MQTrigger-Topic":      trigger.Spec.Topic,
+		"X-Fission-MQTrigger-RespTopic":  trigger.Spec.ResponseTopic,
+		"X-Fission-MQTrigger-ErrorTopic": trigger.Spec.ErrorTopic,
+		"Content-Type":                   trigger.Spec.ContentType,
+	}
+
+	// Create request
+	req, err := http.NewRequest("POST", url, strings.NewReader(value))
+	if err != nil {
+		kafka.logger.Error("failed to create HTTP request to invoke function",
+			zap.Error(err),
+			zap.String("function_url", url))
+		return false
+	}
+
+	// Set the headers came from Kafka record
+	// Using Header.Add() as msg.Headers may have keys with more than one value
+	if kafka.version.IsAtLeast(sarama.V0_11_0_0) {
+		for _, h := range msg.Headers {
+			req.Header.Add(string(h.Key), string(h.Value))
+		}
+	} else {
+		kafka.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
+			zap.Any("current_version", kafka.version))
+	}
+
+	for k, v := range fissionHeaders {
+		req.Header.Set(k, v)
+	}
+
+	// Make the request
+	var resp *http.Response
+	for attempt := 0; attempt <= trigger.Spec.MaxRetries; attempt++ {
+		// Make the request
+		resp, err = http.DefaultClient.Do(req)
+		if err != nil {
+			kafka.logger.Error("sending function invocation request failed",
+				zap.Error(err),
+				zap.String("function_url", url),
+				zap.String("trigger", trigger.Metadata.Name))
+			continue
+		}
+		if resp == nil {
+			continue
+		}
+		if err == nil && resp.StatusCode == http.StatusOK {
+			// Success, quit retrying
+			break
+		}
+	}
+
+	if resp == nil {
+		kafka.logger.Warn("every function invocation retry failed; final retry gave empty response",
+			zap.String("function_url", url),
+			zap.String("trigger", trigger.Metadata.Name))
+		return false
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	kafka.logger.Info("got response from function invocation",
+		zap.String("function_url", url),
+		zap.String("trigger", trigger.Metadata.Name),
+		zap.String("body", string(body)))
+	if err != nil {
+		errorHandler(kafka.logger, trigger, producer, fmt.Sprintf("request body error: %v", string(body)))
+		return false
+	}
+	if resp.StatusCode != 200 {
+		errorHandler(kafka.logger, trigger, producer, fmt.Sprintf("request returned failure: %v", resp.StatusCode))
+		return false
+	}
+	if len(trigger.Spec.ResponseTopic) > 0 {
+		// Generate Kafka record headers
+		var kafkaRecordHeaders []sarama.RecordHeader
+		if kafka.version.IsAtLeast(sarama.V0_11_0_0) {
+			for k, v := range resp.Header {
+				// One key may have multiple values
+				for _, v := range v {
+					kafkaRecordHeaders = append(kafkaRecordHeaders, sarama.RecordHeader{Key: []byte(k), Value: []byte(v)})
+				}
+			}
+		} else {
+			kafka.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
+				zap.Any("current_version", kafka.version))
+		}
+
+		_, _, err := producer.SendMessage(&sarama.ProducerMessage{
+			Topic:   trigger.Spec.ResponseTopic,
+			Value:   sarama.StringEncoder(body),
+			Headers: kafkaRecordHeaders,
+		})
+		if err != nil {
+			kafka.logger.Warn("failed to publish response body from function invocation to topic",
+				zap.Error(err),
+				zap.String("topic", trigger.Spec.Topic),
+				zap.String("function_url", url))
+			return false
+		}
+	}
+	return true
+}
+
+func errorHandler(logger *zap.Logger, trigger *fv1.MessageQueueTrigger, producer sarama.SyncProducer, body string) {
+	if len(trigger.Spec.ErrorTopic) > 0 {
+		_, _, err := producer.SendMessage(&sarama.ProducerMessage{
+			Topic: trigger.Spec.ErrorTopic,
+			Value: sarama.StringEncoder(body),
+		})
+		if err != nil {
+			logger.Warn("failed to publish message to error topic",
+				zap.Error(err),
+				zap.String("topic", trigger.Spec.Topic))
+			return
+		}
+	} else {
+		logger.Error("message received to publish to error topic, but no error topic was set",
+			zap.String("message", body))
+	}
+}

pkg/mqtrigger/messageQueue/messageQueue.go

@@ -0,0 +1,240 @@
+/*
+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"
+	"fmt"
+	"time"
+
+	"github.com/fission/fission/pkg/types"
+	"github.com/fission/fission/pkg/utils"
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+	"github.com/fission/fission/pkg/crd"
+)
+
+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 *fv1.MessageQueueTrigger) (messageQueueSubscription, error)
+		unsubscribe(triggerSub messageQueueSubscription) error
+	}
+
+	MessageQueueTriggerManager struct {
+		logger        *zap.Logger
+		reqChan       chan request
+		mqCfg         MessageQueueConfig
+		triggers      map[string]*triggerSubscription
+		fissionClient *crd.FissionClient
+		messageQueue  MessageQueue
+	}
+
+	triggerSubscription struct {
+		trigger      fv1.MessageQueueTrigger
+		subscription messageQueueSubscription
+	}
+
+	request struct {
+		requestType
+		triggerSub *triggerSubscription
+		respChan   chan response
+	}
+	response struct {
+		err      error
+		triggers *map[string]*triggerSubscription
+	}
+)
+
+func MakeMessageQueueTriggerManager(logger *zap.Logger, fissionClient *crd.FissionClient, routerUrl string, mqConfig MessageQueueConfig) *MessageQueueTriggerManager {
+	var messageQueue MessageQueue
+	var err error
+
+	mqTriggerMgr := MessageQueueTriggerManager{
+		logger:        logger.Named("message_queue_trigger_manager"),
+		reqChan:       make(chan request),
+		triggers:      make(map[string]*triggerSubscription),
+		fissionClient: fissionClient,
+	}
+	switch mqConfig.MQType {
+	case types.MessageQueueTypeNats:
+		messageQueue, err = makeNatsMessageQueue(logger, routerUrl, mqConfig)
+	case types.MessageQueueTypeASQ:
+		messageQueue, err = newAzureStorageConnection(logger, routerUrl, mqConfig)
+	case types.MessageQueueTypeKafka:
+		messageQueue, err = makeKafkaMessageQueue(logger, routerUrl, mqConfig)
+	default:
+		err = fmt.Errorf("no supported message queue type found for %q", mqConfig.MQType)
+	}
+	if err != nil {
+		logger.Fatal("failed to connect to remote message queue server", zap.Error(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
+			k := crd.CacheKey(&req.triggerSub.trigger.Metadata)
+			if _, ok := mqt.triggers[k]; ok {
+				err = errors.New("trigger already exists")
+			} else {
+				mqt.triggers[k] = req.triggerSub
+			}
+			req.respChan <- response{err: err}
+		case GET_ALL_TRIGGERS:
+			copyTriggers := make(map[string]*triggerSubscription)
+			for key, val := range mqt.triggers {
+				copyTriggers[key] = val
+			}
+			req.respChan <- response{triggers: &copyTriggers}
+		case DELETE_TRIGGER:
+			delete(mqt.triggers, crd.CacheKey(&req.triggerSub.trigger.Metadata))
+		}
+	}
+}
+
+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]*triggerSubscription {
+	respChan := make(chan response)
+	mqt.reqChan <- request{
+		requestType: GET_ALL_TRIGGERS,
+		respChan:    respChan,
+	}
+	r := <-respChan
+	return r.triggers
+}
+
+func (mqt *MessageQueueTriggerManager) delTrigger(m *metav1.ObjectMeta) {
+	mqt.reqChan <- request{
+		requestType: DELETE_TRIGGER,
+		triggerSub: &triggerSubscription{
+			trigger: fv1.MessageQueueTrigger{
+				Metadata: *m,
+			},
+		},
+	}
+}
+
+func (mqt *MessageQueueTriggerManager) syncTriggers() {
+	for {
+		// get new set of triggers
+		newTriggers, err := mqt.fissionClient.MessageQueueTriggers(metav1.NamespaceAll).List(metav1.ListOptions{})
+		if err != nil {
+			if utils.IsNetworkError(err) {
+				mqt.logger.Info("encountered network error, will retry", zap.Error(err))
+				time.Sleep(5 * time.Second)
+				continue
+			}
+			mqt.logger.Fatal("failed to read message queue trigger list", zap.Error(err))
+		}
+		newTriggerMap := make(map[string]*fv1.MessageQueueTrigger)
+		for index := range newTriggers.Items {
+			newTrigger := &newTriggers.Items[index]
+			newTriggerMap[crd.CacheKey(&newTrigger.Metadata)] = newTrigger
+		}
+
+		// get current set of triggers
+		currentTriggers := mqt.getAllTriggers()
+
+		// register new triggers
+		for key, trigger := range newTriggerMap {
+			if _, ok := (*currentTriggers)[key]; ok {
+				continue
+			}
+
+			// actually subscribe using the message queue client impl
+			sub, err := mqt.messageQueue.subscribe(trigger)
+			if err != nil {
+				mqt.logger.Warn("failed to subscribe to message queue trigger", zap.Error(err), zap.String("trigger_name", trigger.Metadata.Name))
+				continue
+			}
+
+			triggerSub := triggerSubscription{
+				trigger:      *trigger,
+				subscription: sub,
+			}
+
+			// add to our list
+			err = mqt.addTrigger(&triggerSub)
+			if err != nil {
+				mqt.logger.Fatal("adding message queue trigger failed", zap.Error(err), zap.String("trigger_name", trigger.Metadata.Name))
+			}
+
+			mqt.logger.Info("message queue trigger created", zap.String("trigger_name", trigger.Metadata.Name))
+		}
+
+		// remove old triggers
+		for key, triggerSub := range *currentTriggers {
+			if _, ok := newTriggerMap[key]; ok {
+				continue
+			}
+			err := mqt.messageQueue.unsubscribe(triggerSub.subscription)
+			if err != nil {
+				mqt.logger.Warn("failed to unsubscribe from message queue trigger", zap.Error(err), zap.String("trigger_name", triggerSub.trigger.Metadata.Name))
+				continue
+			}
+			mqt.delTrigger(&triggerSub.trigger.Metadata)
+			mqt.logger.Info("message queue trigger deleted", zap.String("trigger_name", triggerSub.trigger.Metadata.Name))
+		}
+
+		// TODO replace with a watch
+		time.Sleep(3 * time.Second)
+	}
+}
+
+func IsTopicValid(mqType string, topic string) bool {
+	switch mqType {
+	case fv1.MessageQueueTypeNats:
+		return isTopicValidForNats(topic)
+	case fv1.MessageQueueTypeKafka:
+		return isTopicValidForKafka(topic)
+	}
+	return false
+}

pkg/mqtrigger/messageQueue/nats.go

@@ -0,0 +1,217 @@
+/*
+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"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
+	ns "github.com/nats-io/go-nats-streaming"
+	nsUtil "github.com/nats-io/nats-streaming-server/util"
+	"go.uber.org/zap"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+	"github.com/fission/fission/pkg/types"
+	"github.com/fission/fission/pkg/utils"
+)
+
+const (
+	natsClusterID  = "fissionMQTrigger"
+	natsProtocol   = "nats://"
+	natsClientID   = "fission"
+	natsQueueGroup = "fission-messageQueueNatsTrigger"
+)
+
+type (
+	Nats struct {
+		logger    *zap.Logger
+		nsConn    ns.Conn
+		routerUrl string
+	}
+)
+
+func makeNatsMessageQueue(logger *zap.Logger, routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
+	conn, err := ns.Connect(natsClusterID, natsClientID, ns.NatsURL(mqCfg.Url),
+		ns.SetConnectionLostHandler(func(conn ns.Conn, reason error) {
+			// TODO: Better way to handle connection lost problem.
+			// Currently, MessageQueue has no such interface to expose the status of underlying
+			// messaging service, hence MessageQueueTriggerManager has no way to detect and handle
+			// such situation properly. It takes some time to redesign interface of MessageQueue.
+			// For now, we simply fatal here.
+			logger.Fatal("Connection lost", zap.Error(reason))
+		}),
+	)
+	if err != nil {
+		return nil, err
+	}
+	nats := Nats{
+		logger:    logger.Named("nats"),
+		nsConn:    conn,
+		routerUrl: routerUrl,
+	}
+	return nats, nil
+}
+
+func (nats Nats) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubscription, error) {
+	subj := trigger.Spec.Topic
+
+	if !isTopicValidForNats(subj) {
+		return nil, fmt.Errorf("not a valid topic: %q", trigger.Spec.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(string(trigger.Metadata.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.IsChannelNameValid(topic, false)
+}
+
+func msgHandler(nats *Nats, trigger *fv1.MessageQueueTrigger) func(*ns.Msg) {
+	return func(msg *ns.Msg) {
+
+		// Support other function ref types
+		if trigger.Spec.FunctionReference.Type != types.FunctionReferenceTypeFunctionName {
+			nats.logger.Fatal("unsupported function reference type for trigger",
+				zap.Any("function_reference_type", trigger.Spec.FunctionReference.Type),
+				zap.String("trigger", trigger.Metadata.Name))
+		}
+
+		// with the addition of multi-tenancy, the users can create functions in any namespace. however,
+		// the triggers can only be created in the same namespace as the function.
+		// so essentially, function namespace = trigger namespace.
+		url := nats.routerUrl + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/")
+		nats.logger.Info("making HTTP request", zap.String("url", url))
+
+		headers := map[string]string{
+			"X-Fission-MQTrigger-Topic":      trigger.Spec.Topic,
+			"X-Fission-MQTrigger-RespTopic":  trigger.Spec.ResponseTopic,
+			"X-Fission-MQTrigger-ErrorTopic": trigger.Spec.ErrorTopic,
+			"Content-Type":                   trigger.Spec.ContentType,
+		}
+
+		// Create request
+		req, err := http.NewRequest("POST", url, bytes.NewReader(msg.Data))
+
+		if err != nil {
+			nats.logger.Error("failed to create HTTP request to invoke function",
+				zap.Error(err),
+				zap.String("function_url", url))
+			return
+		}
+
+		for k, v := range headers {
+			req.Header.Set(k, v)
+		}
+
+		var resp *http.Response
+		for attempt := 0; attempt <= trigger.Spec.MaxRetries; attempt++ {
+			// Make the request
+			resp, err = http.DefaultClient.Do(req)
+			if err != nil {
+				nats.logger.Error("sending function invocation request failed",
+					zap.Error(err),
+					zap.String("function_url", url),
+					zap.String("trigger", trigger.Metadata.Name))
+				continue
+			}
+			if resp == nil {
+				continue
+			}
+			if err == nil && resp.StatusCode == http.StatusOK {
+				// Success, quit retrying
+				break
+			}
+		}
+
+		if resp == nil {
+			nats.logger.Warn("every function invocation retry failed; final retry gave empty response",
+				zap.String("function_url", url),
+				zap.String("trigger", trigger.Metadata.Name))
+			return
+		}
+
+		defer resp.Body.Close()
+
+		body, bodyErr := ioutil.ReadAll(resp.Body)
+		if bodyErr != nil {
+			nats.logger.Error("error reading function invocation response",
+				zap.Error(err),
+				zap.String("function_url", url),
+				zap.String("trigger", trigger.Metadata.Name))
+			return
+		}
+
+		// Only the latest error response will be published to error topic
+		if err != nil || resp.StatusCode != 200 {
+			if len(trigger.Spec.ErrorTopic) > 0 && len(body) > 0 {
+				publishErr := nats.nsConn.Publish(trigger.Spec.ErrorTopic, body)
+				if publishErr != nil {
+					nats.logger.Error("failed to publish function invocation error to error topic",
+						zap.Error(publishErr),
+						zap.String("topic", trigger.Spec.ErrorTopic),
+						zap.String("function_url", url),
+						zap.String("trigger", trigger.Metadata.Name))
+					// TODO: We will ack this message after max retries to prevent re-processing but
+					// this may cause message loss
+				}
+			}
+			return
+		}
+
+		// Trigger acks message only if a request was processed successfully
+		err = msg.Ack()
+		if err != nil {
+			nats.logger.Error("failed to ack message after successful function invocation from trigger",
+				zap.Error(err),
+				zap.String("function_url", url),
+				zap.String("trigger", trigger.Metadata.Name))
+		}
+
+		if len(trigger.Spec.ResponseTopic) > 0 {
+			err = nats.nsConn.Publish(trigger.Spec.ResponseTopic, body)
+			if err != nil {
+				nats.logger.Error("failed to publish message with function invocation response to topic",
+					zap.Error(err),
+					zap.String("topic", trigger.Spec.ResponseTopic),
+					zap.String("trigger", trigger.Metadata.Name))
+			}
+		}
+	}
+
+}