mempubsub

Documentation

Overview

Package mempubsub provides an in-memory pubsub implementation. Use NewTopic to construct a *pubsub.Topic, and/or NewSubscription to construct a *pubsub.Subscription.

mempubsub should not be used for production: it is intended for local development and testing.

URLs

For pubsub.OpenTopic and pubsub.OpenSubscription, mempubsub registers for the scheme "mem". 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

mempubsub supports at-least-once semantics; applications must call Message.Ack after processing a message, or it will be redelivered. See https://godoc.org/gocloud.dev/pubsub#hdr-At_most_once_and_At_least_once_Delivery for more background.

As

mempubsub does not support any types for As.

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/mempubsub"
	// PRAGMA: On gocloud.dev, hide lines until the next blank line.
	ctx := context.Background()

	// Create a topic.
	topic, err := pubsub.OpenTopic(ctx, "mem://topicA")
	if err != nil {
		log.Fatal(err)
	}
	defer topic.Shutdown(ctx)

	// Create a subscription connected to that topic.
	subscription, err := pubsub.OpenSubscription(ctx, "mem://topicA")
	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/mempubsub"
	// PRAGMA: On gocloud.dev, hide lines until the next blank line.
	ctx := context.Background()

	topic, err := pubsub.OpenTopic(ctx, "mem://topicA")
	if err != nil {
		log.Fatal(err)
	}
	defer topic.Shutdown(ctx)
}

Index

• Constants
• func NewSubscription(pstopic *pubsub.Topic, ackDeadline time.Duration) *pubsub.Subscription
• func NewSubscriptionWithOptions(pstopic *pubsub.Topic, ackDeadline time.Duration, opts *SubscriptionOptions) *pubsub.Subscription
• func NewTopic() *pubsub.Topic
• func NewTopicWithOptions(opts *TopicOptions) *pubsub.Topic
• 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)
• NewSubscription
• NewTopic

Constants

const Scheme = "mem"

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

Functions

func NewSubscription

func NewSubscription(pstopic *pubsub.Topic, ackDeadline time.Duration) *pubsub.Subscription

NewSubscription creates a new subscription for the given topic. It panics if the given topic did not come from mempubsub. If a message is not acked within in the given ack deadline from when it is received, then it will be redelivered.

Example

package main

import (
	"context"
	"time"

	"gocloud.dev/pubsub/mempubsub"
)

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()

	// Construct a *pubsub.Topic.
	topic := mempubsub.NewTopic()
	defer topic.Shutdown(ctx)

	// Construct a *pubsub.Subscription for the topic.
	subscription := mempubsub.NewSubscription(topic, 1*time.Minute /* ack deadline */)
	defer subscription.Shutdown(ctx)
}

func NewSubscriptionWithOptions

func NewSubscriptionWithOptions(pstopic *pubsub.Topic, ackDeadline time.Duration, opts *SubscriptionOptions) *pubsub.Subscription

NewSubscriptionWithOptions is similar to NewSubscription, but supports SubscriptionOptions.

func NewTopic

func NewTopic() *pubsub.Topic

NewTopic creates a new in-memory topic.

Example

package main

import (
	"context"

	"gocloud.dev/pubsub/mempubsub"
)

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()

	topic := mempubsub.NewTopic()
	defer topic.Shutdown(ctx)
}

func NewTopicWithOptions

func NewTopicWithOptions(opts *TopicOptions) *pubsub.Topic

NewTopicWithOptions is similar to NewTopic, but supports TopicOptions.

Types

type SubscriptionOptions

type SubscriptionOptions struct {
	// ReceiveBatcherOptions adds constraints to the default batching done for receives.
	ReceiveBatcherOptions batcher.Options

	// AckBatcherOptions adds constraints to the default batching done for acks.
	AckBatcherOptions batcher.Options
}

SubscriptionOptions will contain configuration for subscriptions.

type TopicOptions

type TopicOptions struct {
	// BatcherOptions adds constraints to the default batching done for sends.
	BatcherOptions batcher.Options
}

TopicOptions contains configuration options for topics.

type URLOpener

type URLOpener struct {
	// contains filtered or unexported fields
}

URLOpener opens mempubsub URLs like "mem://topic".

The URL's host+path is used as the topic to create or subscribe to.

Query parameters:

• ackdeadline: The ack deadline for OpenSubscription, in time.ParseDuration formats. Defaults to 1m.

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.