Code:play 

package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
)

// autoRenewMessageLock runs a loop calling [azservicebus.Receiver.RenewLock] until the message has been completed, the lock
// has expired or the passed in context has been cancelled.
//
// This function returns nil if the message's lock has expired.
//
// - receiver - the receiver that received 'msg'
// - msg - the received message
// - configuredLockDuration - the lock duration, configured at the queue or subscription level.
func autoRenewMessageLock(ctx context.Context, receiver *azservicebus.Receiver, msg *azservicebus.ReceivedMessage, configuredLockDuration time.Duration) error {
	for {
		log.Printf("Renewing lock for %s", msg.MessageID)
		if err := receiver.RenewMessageLock(ctx, msg, nil); err != nil {
			// You get this error if your lock has expired _or_ if you've settled the message.
			// We can ignore it here because the error will also be returned by the settlement
			// method (CompleteMessage, AbandonMessage, etc..)
			if sbErr := (*azservicebus.Error)(nil); errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeLockLost {
				log.Printf("Lock was lost for message with ID %s", msg.MessageID)
				return nil
			}

			return err
		}
		log.Printf("Renewed lock for %s", msg.MessageID)

		select {
		case <-ctx.Done():
			log.Printf("Cancelling lock renewal for %s", msg.MessageID)
			return ctx.Err()
		case <-time.After(configuredLockDuration / 2):
		}
	}
}

func main() {
	// This is configurable, on the service. Change this value to match since it's used
	// to figure out how often to renew a lock.
	const queueOrSubscriptionLockDuration = 30 * time.Second

	messages, err := receiver.ReceiveMessages(context.TODO(), 1, nil)

	if err != nil {
		//  TODO: Update the following line with your application specific error handling logic
		log.Printf("ERROR: %s", err)
		return
	}

	for _, msg := range messages {
		go func(msg *azservicebus.ReceivedMessage) {
			//
			// Each goroutine will _eventually_ stop once its message has been settled.
			//
			// If you want to be more proactive you can create a separate context for each message
			// and cancel that when you settle.
			//
			err := autoRenewMessageLock(context.TODO(), receiver, msg, queueOrSubscriptionLockDuration)

			if err != nil {
				//  TODO: Update the following line with your application specific error handling logic
				log.Printf("ERROR: %s", err)
				return
			}
		}(msg)
	}

	// TODO: Process the messages.
	//
	// Our lock renewal goroutines will stop once RenewLock returns that the lock has been lost. This happens
	// if the lock has expired or if we've settled the message.
}

Code:play 

package main

import (
	"fmt"

	azlog "github.com/Azure/azure-sdk-for-go/sdk/azcore/log"
	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
)

func main() {
	// Required import:
	//   import azlog "github.com/Azure/azure-sdk-for-go/sdk/azcore/log"

	// print log output to stdout
	azlog.SetListener(func(event azlog.Event, s string) {
		fmt.Printf("[%s] %s\n", event, s)
	})

	// pick the set of events to log
	azlog.SetEvents(
		// EventConn is used whenever we create a connection or any links (ie: receivers, senders).
		azservicebus.EventConn,
		// EventAuth is used when we're doing authentication/claims negotiation.
		azservicebus.EventAuth,
		// EventReceiver represents operations that happen on Receivers.
		azservicebus.EventReceiver,
		// EventAdmin is used for operations in the azservicebus/admin.Client
		azservicebus.EventAdmin,
	)
}

Code:

{
	// NOTE: If you'd like to authenticate using a Service Bus connection string
	// look at `NewClientFromConnectionString` instead.

	// `DefaultAzureCredential` tries several common credential types. For more credential types
	// see this link: https://pkg.go.dev/github.com/Azure/azure-sdk-for-go/sdk/azidentity#readme-credential-types.
	credential, err := azidentity.NewDefaultAzureCredential(nil)

	if err != nil {
		panic(err)
	}

	client, err = azservicebus.NewClient("<ex: myservicebus.servicebus.windows.net>", credential, nil)

	if err != nil {
		panic(err)
	}
}

Code:

{
	// NOTE: If you'd like to authenticate using a Service Bus connection string
	// look at `NewClientFromConnectionString` instead.
	client, err = azservicebus.NewClient(endpoint, tokenCredential, &azservicebus.ClientOptions{
		// NOTE: you don't need to configure these explicitly if you like the defaults.
		// For more information see:
		//  https://pkg.go.dev/github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus#RetryOptions
		RetryOptions: azservicebus.RetryOptions{
			// MaxRetries specifies the maximum number of attempts a failed operation will be retried
			// before producing an error.
			MaxRetries: 3,
			// RetryDelay specifies the initial amount of delay to use before retrying an operation.
			// The delay increases exponentially with each retry up to the maximum specified by MaxRetryDelay.
			RetryDelay: 4 * time.Second,
			// MaxRetryDelay specifies the maximum delay allowed before retrying an operation.
			// Typically the value is greater than or equal to the value specified in RetryDelay.
			MaxRetryDelay: 120 * time.Second,
		},
	})

	if err != nil {
		panic(err)
	}
}

Code:

{
	// NOTE: If you'd like to authenticate using a Service Bus connection string
	// look at `NewClientFromConnectionString` instead.

	// `DefaultAzureCredential` tries several common credential types. For more credential types
	// see this link: https://pkg.go.dev/github.com/Azure/azure-sdk-for-go/sdk/azidentity#readme-credential-types.
	credential, err := azidentity.NewDefaultAzureCredential(nil)

	if err != nil {
		panic(err)
	}

	client, err = azservicebus.NewClient("<ex: myservicebus.servicebus.windows.net>", credential, &azservicebus.ClientOptions{
		// A custom endpoint can be used when you need to connect to a TCP proxy.
		CustomEndpoint: "<address/hostname of TCP proxy>",
	})

	if err != nil {
		panic(err)
	}
}

Code:

{
	// You can use an HTTP proxy, with websockets, by setting the appropriate HTTP(s)_PROXY
	// variable in your environment, as described in the https://pkg.go.dev/net/http#ProxyFromEnvironment
	// function.
	//
	// A proxy is NOT required to use websockets.
	newWebSocketConnFn := func(ctx context.Context, args azservicebus.NewWebSocketConnArgs) (net.Conn, error) {
		opts := &websocket.DialOptions{Subprotocols: []string{"amqp"}}
		wssConn, _, err := websocket.Dial(ctx, args.Host, opts)

		if err != nil {
			return nil, err
		}

		return websocket.NetConn(ctx, wssConn, websocket.MessageBinary), nil
	}

	// NOTE: If you'd like to authenticate using a Service Bus connection string
	// look at `NewClientFromConnectionString` instead.
	client, err = azservicebus.NewClient(endpoint, tokenCredential, &azservicebus.ClientOptions{
		NewWebSocketConn: newWebSocketConnFn,
	})

	if err != nil {
		panic(err)
	}

	// NOTE: For users of `nhooyr.io/websocket` there's an open discussion here:
	//   https://github.com/nhooyr/websocket/discussions/380
	//
	// An error ("failed to read frame header: EOF") can be returned when the
	// websocket connection is closed. This error will be returned from the
	// `Client.Close` function and can be ignored, as the websocket "close handshake"
	// has already completed.
	defer client.Close(context.TODO())
}

Code:

{
	// NOTE: If you'd like to authenticate via Azure Active Directory look at
	// the `NewClient` function instead.

	client, err = azservicebus.NewClientFromConnectionString(connectionString, nil)

	if err != nil {
		panic(err)
	}
}

Code:

{
	for {
		func() {
			sessionReceiver, err := client.AcceptNextSessionForQueue(context.TODO(), "exampleSessionQueue", nil)

			if err != nil {
				var sbErr *azservicebus.Error

				if errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeTimeout {
					// there are no sessions available. This isn't fatal - we can use the client and
					// try to AcceptNextSessionForQueue() again.
					fmt.Printf("No session available\n")
					return
				} else {
					panic(err)
				}
			}

			defer sessionReceiver.Close(context.TODO())

			fmt.Printf("Session receiver was assigned session ID \"%s\"", sessionReceiver.SessionID())

			// see ExampleClient_AcceptSessionForQueue() for usage of the SessionReceiver.
		}()
	}
}

Code:play 

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
)

func main() {
	var sessionEnabledQueueName = "exampleSessionQueue"

	for {
		// You can have multiple active session receivers, provided they're each receiving
		// from different sessions.
		//
		// AcceptNextSessionForQueue (or AcceptNextSessionForSubscription) makes it simple to implement
		// this pattern, consuming multiple session receivers in parallel.
		sessionReceiver, err := client.AcceptNextSessionForQueue(context.TODO(), sessionEnabledQueueName, nil)

		if err != nil {
			var sbErr *azservicebus.Error

			if errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeTimeout {
				fmt.Printf("No sessions available\n")

				// NOTE: you could also continue here, which will block and wait again for a
				// session to become available.
				break
			}

			panic(err)
		}

		fmt.Printf("Got receiving for session '%s'\n", sessionReceiver.SessionID())

		// consume the session
		go func() {
			defer func() {
				fmt.Printf("Closing receiver for session '%s'\n", sessionReceiver.SessionID())
				err := sessionReceiver.Close(context.TODO())

				if err != nil {
					panic(err)
				}
			}()

			// we're only reading a few messages here, but you can also receive in a loop
			messages, err := sessionReceiver.ReceiveMessages(context.TODO(), 10, nil)

			if err != nil {
				panic(err)
			}

			fmt.Printf("Received %d messages from session '%s'\n", len(messages), sessionReceiver.SessionID())

			for _, m := range messages {
				if err := processMessageFromSession(context.TODO(), sessionReceiver, m); err != nil {
					panic(err)
				}
			}
		}()
	}
}

func processMessageFromSession(ctx context.Context, receiver *azservicebus.SessionReceiver, receivedMsg *azservicebus.ReceivedMessage) error {
	fmt.Printf("Received message for session %s, message ID %s\n", *receivedMsg.SessionID, receivedMsg.MessageID)
	return receiver.CompleteMessage(ctx, receivedMsg, nil)
}

Code:

{
	for {
		func() {
			sessionReceiver, err := client.AcceptNextSessionForSubscription(context.TODO(), "exampleTopicName", "exampleSubscriptionName", nil)

			if err != nil {
				var sbErr *azservicebus.Error

				if errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeTimeout {
					// there are no sessions available. This isn't fatal - we can use the client and
					// try to AcceptNextSessionForSubscription() again.
					fmt.Printf("No session available\n")
					return
				} else {
					panic(err)
				}
			}

			defer sessionReceiver.Close(context.TODO())

			fmt.Printf("Session receiver was assigned session ID \"%s\"", sessionReceiver.SessionID())

			// see AcceptSessionForSubscription() for some usage of the SessionReceiver itself.
		}()
	}
}

Code:

{
	sessionReceiver, err := client.AcceptSessionForQueue(context.TODO(), "exampleSessionQueue", "exampleSessionId", nil)

	if err != nil {
		panic(err)
	}

	defer sessionReceiver.Close(context.TODO())

	// session receivers work just like non-session receivers
	// with one difference - instead of a lock per message there is a lock
	// for the session itself.

	// Like a message lock, you'll want to periodically renew your session lock.
	err = sessionReceiver.RenewSessionLock(context.TODO(), nil)

	if err != nil {
		panic(err)
	}

	messages, err := sessionReceiver.ReceiveMessages(context.TODO(), 5, nil)

	if err != nil {
		panic(err)
	}

	for _, message := range messages {
		err = sessionReceiver.CompleteMessage(context.TODO(), message, nil)

		if err != nil {
			panic(err)
		}

		fmt.Printf("Received message from session ID \"%s\" and completed it", *message.SessionID)
	}
}

Code:

{
	sessionReceiver, err := client.AcceptSessionForSubscription(context.TODO(), "exampleTopic", "exampleSubscription", "exampleSessionId", nil)

	if err != nil {
		panic(err)
	}

	defer sessionReceiver.Close(context.TODO())

	// see ExampleClient_AcceptSessionForQueue() for usage of the SessionReceiver.
}

Code:

{
	receiver, err = client.NewReceiverForQueue(
		"exampleQueue",
		&azservicebus.ReceiverOptions{
			ReceiveMode: azservicebus.ReceiveModePeekLock,
		},
	)
	exitOnError("Failed to create Receiver", err)

	// close the receiver when it's no longer needed
	defer receiver.Close(context.TODO())
}

Code:

{
	receiver, err = client.NewReceiverForQueue(
		"exampleQueue",
		&azservicebus.ReceiverOptions{
			ReceiveMode: azservicebus.ReceiveModePeekLock,
			SubQueue:    azservicebus.SubQueueDeadLetter,
		},
	)
	exitOnError("Failed to create Receiver for DeadLetterQueue", err)

	// close the receiver when it's no longer needed
	defer receiver.Close(context.TODO())
}

Code:

{
	receiver, err = client.NewReceiverForSubscription(
		"exampleTopic",
		"exampleSubscription",
		&azservicebus.ReceiverOptions{
			ReceiveMode: azservicebus.ReceiveModePeekLock,
		},
	)
	exitOnError("Failed to create Receiver", err)

	// close the receiver when it's no longer needed
	defer receiver.Close(context.TODO())
}

Code:

{
	receiver, err = client.NewReceiverForSubscription(
		"exampleTopic",
		"exampleSubscription",
		&azservicebus.ReceiverOptions{
			ReceiveMode: azservicebus.ReceiveModePeekLock,
			SubQueue:    azservicebus.SubQueueDeadLetter,
		},
	)
	exitOnError("Failed to create Receiver for DeadLetterQueue", err)

	// close the receiver when it's no longer needed
	defer receiver.Close(context.TODO())
}

Code:

{
	sender, err = client.NewSender("<queue or topic>", nil)

	if err != nil {
		panic(err)
	}

	// close the sender when it's no longer needed
	defer sender.Close(context.TODO())
}

Code:

{
	// Send a message to a queue
	sbMessage := &azservicebus.Message{
		Body: []byte("body of message"),
	}
	err = sender.SendMessage(context.TODO(), sbMessage, nil)
	if err != nil {
		panic(err)
	}
	// Create a receiver
	receiver, err := client.NewReceiverForQueue("myqueue", nil)
	if err != nil {
		panic(err)
	}
	defer receiver.Close(context.TODO())
	// Get the message from a queue
	messages, err := receiver.ReceiveMessages(context.TODO(), 1, nil)
	if err != nil {
		panic(err)
	}
	// Send a message to the dead letter queue
	for _, message := range messages {
		deadLetterOptions := &azservicebus.DeadLetterOptions{
			ErrorDescription: to.Ptr("exampleErrorDescription"),
			Reason:           to.Ptr("exampleReason"),
		}
		err := receiver.DeadLetterMessage(context.TODO(), message, deadLetterOptions)
		if err != nil {
			panic(err)
		}
	}
}

Code:

{
	// ReceiveMessages respects the passed in context, and will gracefully stop
	// receiving when 'ctx' is cancelled.
	ctx, cancel := context.WithTimeout(context.TODO(), 60*time.Second)
	defer cancel()

	messages, err = receiver.ReceiveMessages(ctx,
		// The number of messages to receive. Note this is merely an upper
		// bound. It is possible to get fewer message (or zero), depending
		// on the contents of the remote queue or subscription and network
		// conditions.
		1,
		nil,
	)

	if err != nil {
		panic(err)
	}

	for _, message := range messages {
		// The message body is a []byte. For this example we're just assuming that the body
		// was a string, converted to bytes but any []byte payload is valid.
		var body []byte = message.Body
		fmt.Printf("Message received with body: %s\n", string(body))

		// For more information about settling messages:
		// https://docs.microsoft.com/azure/service-bus-messaging/message-transfers-locks-settlement#settling-receive-operations
		err = receiver.CompleteMessage(context.TODO(), message, nil)

		if err != nil {
			var sbErr *azservicebus.Error

			if errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeLockLost {
				// The message lock has expired. This isn't fatal for the client, but it does mean
				// that this message can be received by another Receiver (or potentially this one!).
				fmt.Printf("Message lock expired\n")

				// You can extend the message lock by calling receiver.RenewMessageLock(msg) before the
				// message lock has expired.
				continue
			}

			panic(err)
		}

		fmt.Printf("Received and completed the message\n")
	}
}

Code:

{
	// AMQP is the underlying protocol for all interaction with Service Bus.
	// You can, if needed, send and receive messages that have a 1:1 correspondence
	// with an AMQP message. This gives you full control over details that are not
	// exposed via the azservicebus.ReceivedMessage type.

	messages, err := receiver.ReceiveMessages(context.TODO(), 1, nil)

	if err != nil {
		panic(err)
	}

	// NOTE: For this example we'll assume we received at least one message.

	// Every received message carries a RawAMQPMessage.
	var rawAMQPMessage *azservicebus.AMQPAnnotatedMessage = messages[0].RawAMQPMessage

	// All the various body encodings available for AMQP messages are exposed via Body
	_ = rawAMQPMessage.Body.Data
	_ = rawAMQPMessage.Body.Value
	_ = rawAMQPMessage.Body.Sequence

	// delivery and message annotations
	_ = rawAMQPMessage.DeliveryAnnotations
	_ = rawAMQPMessage.MessageAnnotations

	// headers and footers
	_ = rawAMQPMessage.Header
	_ = rawAMQPMessage.Footer

	// Settlement (if in azservicebus.ReceiveModePeekLockMode) stil works on the ReceivedMessage.
	err = receiver.CompleteMessage(context.TODO(), messages[0], nil)

	if err != nil {
		panic(err)
	}
}

Code:

{
	// Create a dead letter receiver
	deadLetterReceiver, err := client.NewReceiverForQueue(
		"myqueue",
		&azservicebus.ReceiverOptions{
			SubQueue: azservicebus.SubQueueDeadLetter,
		},
	)
	if err != nil {
		panic(err)
	}
	defer deadLetterReceiver.Close(context.TODO())
	// Get messages from the dead letter queue
	deadLetterMessages, err := deadLetterReceiver.ReceiveMessages(context.TODO(), 1, nil)
	if err != nil {
		panic(err)
	}
	// Make messages in the dead letter queue as complete
	for _, deadLetterMessage := range deadLetterMessages {
		fmt.Printf("DeadLetter Reason: %s\nDeadLetter Description: %s\n", *deadLetterMessage.DeadLetterReason, *deadLetterMessage.DeadLetterErrorDescription)
		err := deadLetterReceiver.CompleteMessage(context.TODO(), deadLetterMessage, nil)
		if err != nil {
			panic(err)
		}
	}
}

Code:

{
	// there are two ways of scheduling messages:
	// 1. Using the `Sender.ScheduleMessages()` function.
	// 2. Setting the `Message.ScheduledEnqueueTime` field on a message.

	// schedule the message to be delivered in an hour.
	sequenceNumbers, err := sender.ScheduleMessages(context.TODO(),
		[]*azservicebus.Message{
			{Body: []byte("hello world")},
		}, time.Now().Add(time.Hour), nil)
	exitOnError("Failed to schedule messages", err)

	err = sender.CancelScheduledMessages(context.TODO(), sequenceNumbers, nil)
	exitOnError("Failed to cancel scheduled messages", err)

	// or you can set the `ScheduledEnqueueTime` field on a message when you send it
	future := time.Now().Add(time.Hour)

	err = sender.SendMessage(context.TODO(),
		&azservicebus.Message{
			Body: []byte("hello world"),
			// schedule the message to be delivered in an hour.
			ScheduledEnqueueTime: &future,
		}, nil)
	exitOnError("Failed to schedule messages using SendMessage", err)
}

Code:

{
	// AMQP is the underlying protocol for all interaction with Service Bus.
	// You can, if needed, send and receive messages that have a 1:1 correspondence
	// with an AMQP message. This gives you full control over details that are not
	// exposed via the azservicebus.ReceivedMessage type.

	message := &azservicebus.AMQPAnnotatedMessage{
		Body: azservicebus.AMQPAnnotatedMessageBody{
			// there are three kinds of different body encodings
			// Data, Value and Sequence. See the azservicebus.AMQPMessageBody
			// documentation for more details.
			Value: "hello",
		},
		// full access to fields not normally exposed in azservicebus.Message, like
		// the delivery and message annotations.
		MessageAnnotations: map[any]any{
			"x-opt-message-test": "test-annotation-value",
		},
		DeliveryAnnotations: map[any]any{
			"x-opt-delivery-test": "test-annotation-value",
		},
	}

	err := sender.SendAMQPAnnotatedMessage(context.TODO(), message, nil)

	if err != nil {
		panic(err)
	}
}

Code:

{
	message := &azservicebus.Message{
		Body: []byte("hello, this is a message"),
	}

	err = sender.SendMessage(context.TODO(), message, nil)
	exitOnError("Failed to send message", err)
}

Code:

{
	batch, err := sender.NewMessageBatch(context.TODO(), nil)
	exitOnError("Failed to create message batch", err)

	// By calling AddMessage multiple times you can add multiple messages into a
	// batch. This can help with message throughput, as you can send multiple
	// messages in a single send.
	err = batch.AddMessage(&azservicebus.Message{Body: []byte("hello world")}, nil)

	// We also support adding AMQPMessages directly to a batch as well
	// batch.AddAMQPMessage(&azservicebus.AMQPMessage{})

	if err != nil {
		switch err {
		case azservicebus.ErrMessageTooLarge:
			// At this point you can do a few things:
			// 1. Ignore this message
			// 2. Send this batch (it's full) and create a new batch.
			//
			// The batch can still be used after this error if you have
			// smaller messages you'd still like to add in.
			fmt.Printf("Failed to add message to batch\n")
		default:
			exitOnError("Error while trying to add message to batch", err)
		}
	}

	// After you add all the messages to the batch you send it using
	// Sender.SendMessageBatch()
	err = sender.SendMessageBatch(context.TODO(), batch, nil)
	exitOnError("Failed to send message batch", err)
}