Simple Message Queue (formerly MNS):SDK for Go

Queue-based messaging model

The following example creates a queue, sends messages, receives messages, deletes messages, and deletes a queue.

package main

import (
	"fmt"
	"log"
	"net/http"
	_ "net/http/pprof"
	"time"

	"github.com/aliyun/aliyun-mns-go-sdk"
	"github.com/gogap/logs"
)

func main() {
	go func() {
		log.Println(http.ListenAndServe("localhost:8080", nil))
	}()

	// Replace with your own endpoint.
	endpoint := "http://xxx.mns.cn-hangzhou.aliyuncs.com"
	client := ali_mns.NewClient(endpoint)
	msg := ali_mns.MessageSendRequest{
		MessageBody:  "hello <\"aliyun-mns-go-sdk\">",
		DelaySeconds: 0,
		Priority:     8}

	queueManager := ali_mns.NewMNSQueueManager(client)
	queueName := "test-queue"
	err := queueManager.CreateQueue(queueName, 0, 65536, 345600, 30, 0, 3)
	time.Sleep(time.Duration(2) * time.Second)
	if err != nil && !ali_mns.ERR_MNS_QUEUE_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err) {
		fmt.Println(err)
		return
	}

	queue := ali_mns.NewMNSQueue(queueName, client)
	for i := 1; i < 10000; i++ {
		ret, err := queue.SendMessage(msg)
		go func() {
			fmt.Println(queue.QPSMonitor().QPS())
		}()

		if err != nil {
			fmt.Println(err)
		} else {
			logs.Pretty("response: ", ret)
		}

		endChan := make(chan int)
		respChan := make(chan ali_mns.MessageReceiveResponse)
		errChan := make(chan error)
		go func() {
			select {
			case resp := <-respChan:
				{
					logs.Pretty("response: ", resp)
					logs.Debug("change the visibility: ", resp.ReceiptHandle)
					if ret, e := queue.ChangeMessageVisibility(resp.ReceiptHandle, 5); e != nil {
						fmt.Println(e)
					} else {
						logs.Pretty("visibility changed", ret)
						logs.Debug("delete it now: ", ret.ReceiptHandle)
						if e := queue.DeleteMessage(ret.ReceiptHandle); e != nil {
							fmt.Println(e)
						}
						endChan <- 1
					}
				}
			case err := <-errChan:
				{
					fmt.Println(err)
					endChan <- 1
				}
			}
		}()

		queue.ReceiveMessage(respChan, errChan, 30)
		<-endChan
	}
}

This example demonstrates the following operations:

1. Create a client -- ali_mns.NewClient(endpoint) initializes the MNS client with your endpoint.

2. Create a queue -- CreateQueue accepts the queue name, delay seconds (0), maximum message size (65536 bytes), message retention period (345600 seconds), visibility timeout period (30 seconds), polling wait seconds (0), and slices (3).

3. Send a message -- SendMessage sends a MessageSendRequest with the message content, delay, and priority.

4. Receive a message -- ReceiveMessage uses channels for asynchronous message reception with a 30-second wait.

5. Change visibility timeout period -- ChangeMessageVisibility extends the visibility timeout period using the receipt handle and returns a new receipt handle.

6. Delete a message -- DeleteMessage removes a message from the queue using its receipt handle.

Topic-based messaging model

The following example creates a queue, creates a topic, subscribes to a topic, and publishes messages.

package main

import (
	"fmt"
	"time"

	"github.com/aliyun/aliyun-mns-go-sdk"
	"github.com/gogap/logs"
)

func main() {
	// Replace with your own endpoint.
	endpoint := "http://xxx.mns.cn-hangzhou.aliyuncs.com"
	queueName := "test-queue"
	topicName := "test-topic"
	queueSubName := "test-sub-queue"
	httpSubName := "test-sub-http"
	client := ali_mns.NewClient(endpoint)

	// 1. create a queue for receiving pushed messages
	queueManager := ali_mns.NewMNSQueueManager(client)
	err := queueManager.CreateSimpleQueue(queueName)
	if err != nil && !ali_mns.ERR_MNS_QUEUE_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err) {
		fmt.Println(err)
		return
	}

	// 2. create the topic
	topicManager := ali_mns.NewMNSTopicManager(client)
	// topicManager.DeleteTopic("testTopic")
	err = topicManager.CreateSimpleTopic(topicName)
	if err != nil && !ali_mns.ERR_MNS_TOPIC_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err) {
		fmt.Println(err)
		return
	}

	topic := ali_mns.NewMNSTopic(topicName, client)
	// 3. subscribe to topic, the endpoint is queue
	queueSub := ali_mns.MessageSubsribeRequest{
		Endpoint:            topic.GenerateQueueEndpoint(queueName),
		NotifyContentFormat: ali_mns.SIMPLIFIED,
	}

	// 4. subscribe to topic, the endpoint is HTTP(S)
	httpSub := ali_mns.MessageSubsribeRequest{
		Endpoint:            "http://www.baidu.com",
		NotifyContentFormat: ali_mns.SIMPLIFIED,
	}

	err = topic.Subscribe(queueSubName, queueSub)
	if err != nil && !ali_mns.ERR_MNS_SUBSCRIPTION_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err) {
		fmt.Println(err)
		return
	}

	err = topic.Subscribe(httpSubName, httpSub)
	if err != nil && !ali_mns.ERR_MNS_SUBSCRIPTION_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err) {
		fmt.Println(err)
		return
	}

	/*


			sub = ali_mns.MessageSubsribeRequest{
	        Endpoint:  topic.GenerateMailEndpoint("a@b.com"),
	        NotifyContentFormat: ali_mns.SIMPLIFIED,
	    }
	    err = topic.Subscribe("SubscriptionNameB", sub)
	    if (err != nil && !ali_mns.ERR_MNS_SUBSCRIPTION_ALREADY_EXIST_AND_HAVE_SAME_ATTR.IsEqual(err)) {
	        fmt.Println(err)
	        return
	    }
	*/

	time.Sleep(time.Duration(2) * time.Second)

	// 5. now publish message
	msg := ali_mns.MessagePublishRequest{
		MessageBody: "hello topic <\"aliyun-mns-go-sdk\">",
		MessageAttributes: &ali_mns.MessageAttributes{
			MailAttributes: &ali_mns.MailAttributes{
				Subject:     "AAA Chinese characters",
				AccountName: "BBB",
			},
		},
	}
	_, err = topic.PublishMessage(msg)
	if err != nil {
		fmt.Println(err)
		return
	}

	// 6. receive the message from queue
	queue := ali_mns.NewMNSQueue(queueName, client)
	endChan := make(chan int)
	respChan := make(chan ali_mns.MessageReceiveResponse)
	errChan := make(chan error)
	go func() {
		select {
		case resp := <-respChan:
			{
				logs.Pretty("response: ", resp)
				fmt.Println("change the visibility: ", resp.ReceiptHandle)
				if ret, e := queue.ChangeMessageVisibility(resp.ReceiptHandle, 5); e != nil {
					fmt.Println(e)
				} else {
					logs.Pretty("visibility changed", ret)
					fmt.Println("delete it now: ", ret.ReceiptHandle)
					if e := queue.DeleteMessage(ret.ReceiptHandle); e != nil {
						fmt.Println(e)
					}
					endChan <- 1
				}
			}
		case err := <-errChan:
			{
				fmt.Println(err)
				endChan <- 1
			}
		}
	}()

	queue.ReceiveMessage(respChan, errChan, 30)
	<-endChan
}

This example demonstrates the following operations:

1. Create a queue -- CreateSimpleQueue creates a queue to receive messages pushed from topic subscriptions.

2. Create a topic -- CreateSimpleTopic creates a topic for publishing messages.

3. Subscribe with a queue endpoint -- topic.GenerateQueueEndpoint(queueName) generates the queue endpoint, and topic.Subscribe registers the subscription with SIMPLIFIED notification content format.

4. Subscribe with an HTTP(S) endpoint -- Subscriptions can also push messages to an HTTP(S) URL.

5. Publish a message -- PublishMessage publishes a MessagePublishRequest to the topic. The request can include MessageAttributes such as MailAttributes with Subject and AccountName.

6. Receive the message -- Messages published to the topic are pushed to the subscription queue. Use ReceiveMessage to consume them.

The commented-out code block shows how to create a mail endpoint subscription using topic.GenerateMailEndpoint("a@b.com").