IoT Platform:Go SDK connection example

Prepare the development environment

This example uses Go 1.12.7.

Download the SDK

Use the following command to import the Go AMQP SDK.

import "pack.ag/amqp"

For more information about how to use the SDK, see package amqp.

Code example

package main
import (
	"os"
    "context"
    "crypto/hmac"
    "crypto/sha1"
    "encoding/base64"
    "fmt"
    "pack.ag/amqp"
    "time"
)
// For parameter descriptions, see the AMQP client connection guide.
const consumerGroupId = "${YourConsumerGroupId}"
const clientId = "${YourClientId}"
// iotInstanceId: The instance ID.
const iotInstanceId = "${YourIotInstanceId}"
// The endpoint. For more information, see the AMQP client connection guide.
const host = "${YourHost}"
func main() {
	// If you leak the project code, your AccessKey may be exposed. This compromises the security of all resources in your account. 
	// The following code provides an example on how to use an environment variable to obtain the AccessKey. This method is for reference only.
	accessKey := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
	accessSecret := os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    address := "amqps://" + host + ":5671"
    timestamp := time.Now().Nanosecond() / 1000000
    // For information about how to construct the user name, see the AMQP client connection guide.
    userName := fmt.Sprintf("%s|authMode=aksign,signMethod=Hmacsha1,consumerGroupId=%s,authId=%s,iotInstanceId=%s,timestamp=%d|", 
        clientId, consumerGroupId, accessKey, iotInstanceId, timestamp)
    stringToSign := fmt.Sprintf("authId=%s&timestamp=%d", accessKey, timestamp)
    hmacKey := hmac.New(sha1.New, []byte(accessSecret))
    hmacKey.Write([]byte(stringToSign))
    // Calculate the signature. For information about how to construct the password, see the AMQP client connection guide.
    password := base64.StdEncoding.EncodeToString(hmacKey.Sum(nil))
    amqpManager := &AmqpManager{
        address:address,
        userName:userName,
        password:password,
    }
    // To accept messages or cancel operations, derive the context from context.Background().
    ctx := context.Background()
    amqpManager.startReceiveMessage(ctx)
}
// Business function. This is a custom implementation. The function is executed asynchronously. Consider the resource consumption of your system.
func (am *AmqpManager) processMessage(message *amqp.Message) {
    fmt.Println("data received:", string(message.GetData()), " properties:", message.ApplicationProperties)
}
type AmqpManager struct {
    address     string
    userName     string
    password     string
    client       *amqp.Client
    session     *amqp.Session
    receiver     *amqp.Receiver
}
func (am *AmqpManager) startReceiveMessage(ctx context.Context)  {
    childCtx, _ := context.WithCancel(ctx)
    err := am.generateReceiverWithRetry(childCtx)
    if nil != err {
        return
    }
    defer func() {
        am.receiver.Close(childCtx)
        am.session.Close(childCtx)
        am.client.Close()
    }()
    for {
        // Blocks to receive messages. If the context is background, the process is not interrupted.
        message, err := am.receiver.Receive(ctx)
        if nil == err {
            go am.processMessage(message)
            message.Accept()
        } else {
            fmt.Println("amqp receive data error:", err)
            // If the operation is actively canceled, exit the program.
            select {
            case <- childCtx.Done(): return
            default:
            }
            // If the operation is not actively canceled, re-establish the connection.
            err := am.generateReceiverWithRetry(childCtx)
            if nil != err {
                return
            }
        }
    }
}
func (am *AmqpManager) generateReceiverWithRetry(ctx context.Context) error {
    // Back off and reconnect. Start with 10 ms and double the interval up to 20s.
    duration := 10 * time.Millisecond
    maxDuration := 20000 * time.Millisecond
    times := 1
    // In case of an exception, back off and reconnect.
    for {
        select {
        case <- ctx.Done(): return amqp.ErrConnClosed
        default:
        }
        err := am.generateReceiver()
        if nil != err {
            time.Sleep(duration)
            if duration < maxDuration {
                duration *= 2
            }
            fmt.Println("amqp connect retry,times:", times, ",duration:", duration)
            times ++
        } else {
            fmt.Println("amqp connect init success")
            return nil
        }
    }
}
// The connection and session states cannot be determined because the package is not visible. Restart the connection to retrieve the states.
func (am *AmqpManager) generateReceiver() error {
    if am.session != nil {
        receiver, err := am.session.NewReceiver(
            amqp.LinkSourceAddress("/queue-name"),
            amqp.LinkCredit(20),
        )
        // If a network disconnection occurs, the connection closes and session creation fails. 
        // If the connection remains open, the session is created successfully.
        if err == nil {
            am.receiver = receiver
            return nil
        }
    }
    // Clean up the previous connection.
    if am.client != nil {
        am.client.Close()
    }
    client, err := amqp.Dial(am.address, amqp.ConnSASLPlain(am.userName, am.password), )
    if err != nil {
        return err
    }
    am.client = client
    session, err := client.NewSession()
    if err != nil {
        return err
    }
    am.session = session
    receiver, err := am.session.NewReceiver(
        amqp.LinkSourceAddress("/queue-name"),
        amqp.LinkCredit(20),
    )
    if err != nil {
        return err
    }
    am.receiver = receiver
    return nil
}

Configure the parameters in the preceding code as described in the following table. For more information, see Connect an AMQP client to IoT Platform.

Sample run results

• Success: A log message similar to the following is returned. This indicates that the AMQP client has successfully connected to IoT Platform and received messages from the device.

  amqp connect init success
  data received: {"deviceType":"CustomCategory","iotId":"xxx","requestId":"1613726251726","checkFailedData":0,"productKey":"xxx","gmtCreate":"1613726121717,"deviceName":"xxx","items":{"Temperature":{"value":24,"time":1613726121715},"Humidity":{"value":19,"time":1613726121715}}}  properties: map[generateTime:1613726121721 messageId:xxx xxx s:1 topic: /xxx/thing/event/property/post]
  data received: {"deviceType":"CustomCategory","iotId":"xxx","requestId":"1613725651726","checkFailedData":0,"productKey":"xxx","gmtCreate":"1613725521715,"deviceName":"xxx","items":{"Temperature":{"value":28,"time":1613725521712},"Humidity":{"value":19,"time":1613725521712}}}  properties: map[generateTime:1613725521719 messageId:1362689721104473600 qos:1 topic: /xxx/thing/event/property/post]

• Failed: A log message similar to the following one is returned, indicating that the AMQP client failed to connect to IoT Platform.

  Use the error log to check your code and network settings. Resolve the issue and run the code again.

  amqp connect retry,times: 1 ,duration: 20ms
  amqp connect retry,times: 2 ,duration: 40ms
  amqp connect retry,times: 3 ,duration: 80ms
  amqp connect retry,times: 4 ,duration: 160ms
  amqp connect retry,times: 5 ,duration: 320ms
  amqp connect retry,times: 6 ,duration: 640ms
  amqp connect retry,times: 7 ,duration: 1.28s
  amqp connect retry,times: 8 ,duration: 2.56s
  amqp connect retry,times: 9 ,duration: 5.12s
  amqp connect retry,times: 10 ,duration: 10.24s
  amqp connect retry,times: 11 ,duration: 20.48s

References

For more information about error codes for server-side subscription messages, see Message-related error codes.