azuresb

Documentation

Overview

Package azuresb provides an implementation of pubsub using Azure Service Bus Topic and Subscription. See https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview for an overview.

URLs

For pubsub.OpenTopic and pubsub.OpenSubscription, azuresb registers for the scheme "azuresb". The default URL opener will use a Service Bus Connection String based on the environment variable "SERVICEBUS_CONNECTION_STRING". To customize the URL opener, or for more details on the URL format, see URLOpener. See https://gocloud.dev/concepts/urls/ for background information.

Message Delivery Semantics

Azure ServiceBus supports at-least-once semantics in the default Peek-Lock mode; messages will be redelivered if they are not Acked, or if they are explicitly Nacked.

ServiceBus also supports a Receive-Delete mode, which essentially auto-acks a message when it is delivered, resulting in at-most-once semantics. Set SubscriberOptions.ReceiveAndDelete to true to tell azuresb.Subscription that you've enabled Receive-Delete mode. When enabled, pubsub.Message.Ack is a no-op, pubsub.Message.Nackable will return false, and pubsub.Message.Nack will panic.

See https://godoc.org/gocloud.dev/pubsub#hdr-At_most_once_and_At_least_once_Delivery for more background.

As

azuresb exposes the following types for As:

• Topic: *servicebus.Topic
• Subscription: *servicebus.Subscription
• Message.BeforeSend: *servicebus.Message
• Message.AfterSend: None
• Message: *servicebus.Message
• Error: common.Retryable, *amqp.Error, *amqp.DetachError

Example (OpenSubscriptionFromURL)

package main

import (
	"context"
	"log"

	"gocloud.dev/pubsub"
)

func main() {
	// PRAGMA: This example is used on gocloud.dev; PRAGMA comments adjust how it is shown and can be ignored.
	// PRAGMA: On gocloud.dev, add a blank import: _ "gocloud.dev/pubsub/azuresb"
	// PRAGMA: On gocloud.dev, hide lines until the next blank line.
	ctx := context.Background()

	// pubsub.OpenSubscription creates a *pubsub.Subscription from a URL.
	// This URL will open the subscription "mysubscription" for the topic
	// "mytopic" using a connection string from the environment variable
	// SERVICEBUS_CONNECTION_STRING.
	subscription, err := pubsub.OpenSubscription(ctx,
		"azuresb://mytopic?subscription=mysubscription")
	if err != nil {
		log.Fatal(err)
	}
	defer subscription.Shutdown(ctx)
}

Example (OpenTopicFromURL)

package main

import (
	"context"
	"log"

	"gocloud.dev/pubsub"
)

func main() {
	// PRAGMA: This example is used on gocloud.dev; PRAGMA comments adjust how it is shown and can be ignored.
	// PRAGMA: On gocloud.dev, add a blank import: _ "gocloud.dev/pubsub/azuresb"
	// PRAGMA: On gocloud.dev, hide lines until the next blank line.
	ctx := context.Background()

	// pubsub.OpenTopic creates a *pubsub.Topic from a URL.
	// This URL will open the topic "mytopic" using a connection string
	// from the environment variable SERVICEBUS_CONNECTION_STRING.
	topic, err := pubsub.OpenTopic(ctx, "azuresb://mytopic")
	if err != nil {
		log.Fatal(err)
	}
	defer topic.Shutdown(ctx)
}

Index

• Constants
• func NewNamespaceFromConnectionString(connectionString string) (*servicebus.Namespace, error)
• func NewSubscription(parentTopic *servicebus.Topic, subscriptionName string, ...) (*servicebus.Subscription, error)
• func NewTopic(ns *servicebus.Namespace, topicName string, opts []servicebus.TopicOption) (*servicebus.Topic, error)
• func OpenSubscription(ctx context.Context, parentNamespace *servicebus.Namespace, ...) (*pubsub.Subscription, error)
• func OpenTopic(ctx context.Context, sbTopic *servicebus.Topic, opts *TopicOptions) (*pubsub.Topic, error)
• type SubscriptionOptions
• type TopicOptions
• type URLOpener
  • func (o *URLOpener) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsub.Subscription, error)
  • func (o *URLOpener) OpenTopicURL(ctx context.Context, u *url.URL) (*pubsub.Topic, error)

Examples

• Package (OpenSubscriptionFromURL)
• Package (OpenTopicFromURL)
• OpenSubscription
• OpenTopic

Constants

const Scheme = "azuresb"

Scheme is the URL scheme azuresb registers its URLOpeners under on pubsub.DefaultMux.

Functions

func NewNamespaceFromConnectionString

func NewNamespaceFromConnectionString(connectionString string) (*servicebus.Namespace, error)

NewNamespaceFromConnectionString returns a *servicebus.Namespace from a Service Bus connection string. https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-get-started-with-queues

func NewSubscription

func NewSubscription(parentTopic *servicebus.Topic, subscriptionName string, opts []servicebus.SubscriptionOption) (*servicebus.Subscription, error)

NewSubscription returns a *servicebus.Subscription associated with a Service Bus Topic.

func NewTopic

func NewTopic(ns *servicebus.Namespace, topicName string, opts []servicebus.TopicOption) (*servicebus.Topic, error)

NewTopic returns a *servicebus.Topic associated with a Service Bus Namespace.

func OpenSubscription

func OpenSubscription(ctx context.Context, parentNamespace *servicebus.Namespace, parentTopic *servicebus.Topic, sbSubscription *servicebus.Subscription, opts *SubscriptionOptions) (*pubsub.Subscription, error)

OpenSubscription initializes a pubsub Subscription on a given Service Bus Subscription and its parent Service Bus Topic.

Example

package main

import (
	"context"
	"log"
	"os"

	"gocloud.dev/pubsub/azuresb"
)

func main() {
	// PRAGMA: This example is used on gocloud.dev; PRAGMA comments adjust how it is shown and can be ignored.
	// PRAGMA: On gocloud.dev, hide lines until the next blank line.
	ctx := context.Background()

	// Change these as needed for your application.
	serviceBusConnString := os.Getenv("SERVICEBUS_CONNECTION_STRING")
	const topicName = "test-topic"
	const subscriptionName = "test-subscription"

	// Connect to Azure Service Bus for the given subscription.
	busNamespace, err := azuresb.NewNamespaceFromConnectionString(serviceBusConnString)
	if err != nil {
		log.Fatal(err)
	}
	busTopic, err := azuresb.NewTopic(busNamespace, topicName, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer busTopic.Close(ctx)
	busSub, err := azuresb.NewSubscription(busTopic, subscriptionName, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer busSub.Close(ctx)

	// Construct a *pubsub.Subscription.
	subscription, err := azuresb.OpenSubscription(ctx,
		busNamespace, busTopic, busSub, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer subscription.Shutdown(ctx)
}

func OpenTopic

func OpenTopic(ctx context.Context, sbTopic *servicebus.Topic, opts *TopicOptions) (*pubsub.Topic, error)

OpenTopic initializes a pubsub Topic on a given Service Bus Topic.

Example

package main

import (
	"context"
	"log"
	"os"

	"gocloud.dev/pubsub/azuresb"
)

func main() {
	// PRAGMA: This example is used on gocloud.dev; PRAGMA comments adjust how it is shown and can be ignored.
	// PRAGMA: On gocloud.dev, hide lines until the next blank line.
	ctx := context.Background()

	// Change these as needed for your application.
	connString := os.Getenv("SERVICEBUS_CONNECTION_STRING")
	topicName := "test-topic"

	if connString == "" {
		log.Fatal("Service Bus ConnectionString is not set")
	}

	// Connect to Azure Service Bus for the given topic.
	busNamespace, err := azuresb.NewNamespaceFromConnectionString(connString)
	if err != nil {
		log.Fatal(err)
	}
	busTopic, err := azuresb.NewTopic(busNamespace, topicName, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer busTopic.Close(ctx)

	// Construct a *pubsub.Topic.
	topic, err := azuresb.OpenTopic(ctx, busTopic, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer topic.Shutdown(ctx)
}

Types

type SubscriptionOptions

type SubscriptionOptions struct {
	// If false, the serviceBus.Subscription MUST be in the default Peek-Lock mode.
	// If true, the serviceBus.Subscription MUST be in Receive-and-Delete mode.
	// When true: pubsub.Message.Ack will be a no-op, pubsub.Message.Nackable
	// will return true, and pubsub.Message.Nack will panic.
	ReceiveAndDelete bool
}

SubscriptionOptions will contain configuration for subscriptions.

type TopicOptions

type TopicOptions struct{}

TopicOptions provides configuration options for an Azure SB Topic.

type URLOpener

type URLOpener struct {
	// ConnectionString is the Service Bus connection string (required).
	// https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-get-started-with-queues
	ConnectionString string

	// Options passed when creating the ServiceBus Topic/Subscription.
	ServiceBusTopicOptions        []servicebus.TopicOption
	ServiceBusSubscriptionOptions []servicebus.SubscriptionOption

	// TopicOptions specifies the options to pass to OpenTopic.
	TopicOptions TopicOptions
	// SubscriptionOptions specifies the options to pass to OpenSubscription.
	SubscriptionOptions SubscriptionOptions
}

URLOpener opens Azure Service Bus URLs like "azuresb://mytopic" for topics or "azuresb://mytopic?subscription=mysubscription" for subscriptions.

• The URL's host+path is used as the topic name.
• For subscriptions, the subscription name must be provided in the "subscription" query parameter.

No other query parameters are supported.

func (*URLOpener) OpenSubscriptionURL

func (o *URLOpener) OpenSubscriptionURL(ctx context.Context, u *url.URL) (*pubsub.Subscription, error)

OpenSubscriptionURL opens a pubsub.Subscription based on u.

func (*URLOpener) OpenTopicURL

func (o *URLOpener) OpenTopicURL(ctx context.Context, u *url.URL) (*pubsub.Topic, error)

OpenTopicURL opens a pubsub.Topic based on u.