pubsub

Documentation

Index

• Constants
• func DefaultProjectId() string
• type Client
  • func DefaultClient(promNamespace string) (Client, error)
  • func MustDefaultClient() Client
  • func NewClient(ctx context.Context, projectID string, promNamespace string, ...) (Client, error)
• type ClientCommon
• type FanoutMessageHandler
  • func RouteBy(classifier func(Message) string, handlers map[string]FanoutMessageHandler) FanoutMessageHandler
• type Message
• type MessageCommon
• type MessageHandler
  • func Fanout(handlers ...FanoutMessageHandler) MessageHandler
• type PublishResult
• type PublishSettings
• type PushMessage
• type PushRequest
• type RealMessage
• type RetriableError
  • func AsRetriable(err error, retriable bool) RetriableError
• type RetryPolicy
• type Subscription
• type SubscriptionCommon
• type SubscriptionConfig
• type SubscriptionConfigToUpdate
• type SubscriptionIterator
• type Topic
• type TopicCommon
• type TopicConfig
• type TopicConfigToUpdate
• type TopicIterator

Constants

const (
	DefaultProjectIdEnvVar = "PUBSUB_GCLOUD_PROJECT_ID"
	DefaultEmulatorHost    = "localhost:8802"
	EmulatorHostEnvVar     = "PUBSUB_EMULATOR_HOST"
)

Functions

func DefaultProjectId

func DefaultProjectId() string

Types

type Client

type Client interface {
	ClientCommon
	IsEmulator() bool
	CreateTopic(ctx context.Context, topicID string) (Topic, error)
	Subscription(id string) Subscription
	Subscriptions(context.Context) SubscriptionIterator
	Topic(id string) Topic
	Topics(context.Context) TopicIterator
}

func DefaultClient

func DefaultClient(promNamespace string) (Client, error)

func MustDefaultClient

func MustDefaultClient() Client

func NewClient

func NewClient(
	ctx context.Context,
	projectID string,
	promNamespace string,
	promLabels prometheus.Labels,
	opts ...option.ClientOption,
) (Client, error)

type ClientCommon

type ClientCommon interface {
	Close() error
}

type FanoutMessageHandler

type FanoutMessageHandler func(context.Context, Message) RetriableError

func RouteBy

func RouteBy(classifier func(Message) string, handlers map[string]FanoutMessageHandler) FanoutMessageHandler

RouteBy uses a classifier function to chose exactly one handler to process a message. If no handler matches the classifier return, routing handler will return nil (no error).

type Message

type Message interface {
	MessageCommon
	RealMessage() *RealMessage
}

type MessageCommon

type MessageCommon interface {
	Ack()
	Nack()
}

type MessageHandler

type MessageHandler func(context.Context, Message)

func Fanout

func Fanout(handlers ...FanoutMessageHandler) MessageHandler

Fanout creates a composite handler that will call each of the given handlers concurrently, and ack the message if all of them either succeed or return non-retriable errors, or else nack the message if any returns a retriable error.

Fanout will panic if no handlers are given.

Fanout will call all the handlers concurrently via an errgroup. Thus the context passed to each will be canceled if any other handler fails. It is up to the inner handlers to decide if they should also cancel in this case or not.

The inner handlers _MUST NOT_ call `Ack()` or `Nack()` on the message!

If an inner handler panics, it may cause messages to get stuck indefinitely.

type PublishResult

type PublishResult = pubsub.PublishResult

type PublishSettings

type PublishSettings = pubsub.PublishSettings

type PushMessage

type PushMessage = pubsubv1.PubsubMessage

type PushRequest

type PushRequest struct {
	Message      PushMessage `json:"message"`
	Subscription string      `json:"subscription"`
}

surprisingly, google doesn't export a type for this

type RealMessage

type RealMessage = pubsub.Message

for imports to have access to the real one they need for sending

type RetriableError

type RetriableError interface {
	error
	IsRetriable() bool
}

func AsRetriable

func AsRetriable(err error, retriable bool) RetriableError

type RetryPolicy

type RetryPolicy = pubsub.RetryPolicy

type Subscription

type Subscription interface {
	SubscriptionCommon
	Receive(ctx context.Context, msg MessageHandler) error
	ReceiveSettings() *pubsub.ReceiveSettings
	EnsureDefaultConfig(context.Context, ...func(*SubscriptionConfigToUpdate)) (SubscriptionConfig, error)
}

type SubscriptionCommon

type SubscriptionCommon interface {
	fmt.Stringer
	ID() string
	Config(context.Context) (pubsub.SubscriptionConfig, error)
	Delete(context.Context) error
	Exists(context.Context) (bool, error)
	Update(context.Context, pubsub.SubscriptionConfigToUpdate) (pubsub.SubscriptionConfig, error)
	SeekToTime(ctx context.Context, t time.Time) error
}

type SubscriptionConfig

type SubscriptionConfig = pubsub.SubscriptionConfig

type SubscriptionConfigToUpdate

type SubscriptionConfigToUpdate = pubsub.SubscriptionConfigToUpdate

type SubscriptionIterator

type SubscriptionIterator interface {
	Next() (Subscription, error)
}

type Topic

type Topic interface {
	TopicCommon
	PublishSettings() *PublishSettings
	Subscriptions(context.Context) SubscriptionIterator

	// CreateSubscription lives on Client in the real API, but we move it here to
	// avoid worrying about the Topic member on the SubscriptionConfig struct
	CreateSubscription(ctx context.Context, id string, cfg SubscriptionConfig) (Subscription, error)
}

type TopicCommon

type TopicCommon interface {
	fmt.Stringer
	Config(context.Context) (TopicConfig, error)
	Delete(context.Context) error
	Exists(context.Context) (bool, error)
	ID() string
	Publish(context.Context, *RealMessage) *PublishResult
	Stop()
	Update(context.Context, TopicConfigToUpdate) (pubsub.TopicConfig, error)
}

type TopicConfig

type TopicConfig = pubsub.TopicConfig

re-export for ease of use

type TopicConfigToUpdate

type TopicConfigToUpdate = pubsub.TopicConfigToUpdate

type TopicIterator

type TopicIterator interface {
	Next() (Topic, error)
}