gcppubsub

Documentation

Index

• func DelSubscription(project, name string) error
• func Do(ctx context.Context, args DoArgs) error
• func GetPublisher(project, id string) (zpubsub.MultiPublisher, *gpubsub.Topic, error)
• func GetSubscription(project, id string, topic *gpubsub.Topic, ackdeadline ...time.Duration) (*gpubsub.Subscription, error)
• func GetTopic(project, id string) (*gpubsub.Topic, error)
• func PublishRaw(ctx context.Context, topic *gpubsub.Topic, msg []byte) (string, error)
• type Callback
• type DoArgs
• type LengthySubscriber
  • func NewLengthySubscriber(ctx interface{}, project, subscription string, callback Callback, o ...Option) *LengthySubscriber
  • func (l *LengthySubscriber) Start(ctx context.Context, done ...chan error) error
• type Option
  • func WithClient(v *pubsubv1.SubscriberClient) Option
  • func WithDeadline(v int) Option
  • func WithLogger(v *log.Logger) Option
  • func WithMaxMessages(v int) Option
  • func WithNoExtend(v bool) Option
• type PubsubPublisher
  • func NewPubsubPublisher(projectId string, topicname string) (*PubsubPublisher, error)
  • func (p *PubsubPublisher) Publish(key string, data interface{}) error
  • func (p *PubsubPublisher) RawTopic() *gpubsub.Topic
  • func (p *PubsubPublisher) Test() error

Functions

func DelSubscription

func DelSubscription(project, name string) error

DelSubscription converts the client into an utter introvert.

func Do

func Do(ctx context.Context, args DoArgs) error

Do is our helper function to setup an async PubSub subscription. So far, this works as intended but the frequency of multiple redelivery of messages is surprisingly high than LengthySubscriber. Still under observation while being used in non-critical workflows.

Since this is a long subscription library, the default use case is that every message's processing time will most likely be beyond the subscription's ack time and will be processing one message at a time, although you can set 'MaxOutstandingMessages' to > 1 in which case some form of concurrent processing can still be done.

The function will block until ctx is cancelled or if setup returns an error.

func GetPublisher

func GetPublisher(project, id string) (zpubsub.MultiPublisher, *gpubsub.Topic, error)

GetPublisher is a simple wrapper to create a PubSub publisher using gizmo's Publisher interface.

Deprecated: The gizmo package used in this function is now unmaintained.

func GetSubscription

func GetSubscription(project, id string, topic *gpubsub.Topic, ackdeadline ...time.Duration) (*gpubsub.Subscription, error)

GetSubscription retrieves a PubSub subscription. It creates the subscription if it doesn't exist, using the provided topic object. The default Ack deadline, if not provided, is one minute.

func GetTopic

func GetTopic(project, id string) (*gpubsub.Topic, error)

GetTopic retrieves a PubSub topic. It creates the topic if it doesn't exist.

func PublishRaw

func PublishRaw(ctx context.Context, topic *gpubsub.Topic, msg []byte) (string, error)

PublishRaw is a convenience function for publishing raw data to a topic.

Types

type Callback

type Callback func(ctx interface{}, data []byte) error

type DoArgs

type DoArgs struct {
	ProjectId      string // required
	TopicId        string // required
	SubscriptionId string // required

	// Required. The callback function to process the message asynchronously.
	// The callback is responsible for ack'ing the message or not.
	ReceiveCallback func(context.Context, *pubsub.Message)

	MaxOutstandingMessages int // optional, defaults to 1

	// Optional. Defaults to true when MaxOutstandingMessages is 1, else it
	// defaults to false (StreamingPull/async).
	Synchronous bool

	// Optional. Used only when creating the subscription (if not exists).
	// Defaults to 1 minute.
	AckDeadline time.Duration
}

type LengthySubscriber

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

func NewLengthySubscriber

func NewLengthySubscriber(ctx interface{}, project, subscription string, callback Callback, o ...Option) *LengthySubscriber

NewLengthySubscriber creates a lengthy subscriber object for PubSub.

func (*LengthySubscriber) Start

func (l *LengthySubscriber) Start(ctx context.Context, done ...chan error) error

Start starts the main goroutine handler. Terminates via 'ctx'. If 'done' is provided, will send a message there to signal caller that it's done with processing.

type Option

type Option interface {
	Apply(*LengthySubscriber)
}

func WithClient

func WithClient(v *pubsubv1.SubscriberClient) Option

WithClient sets the PubSub client. If not provided, an internal client is used using the environment's credentials.

func WithDeadline

func WithDeadline(v int) Option

WithDeadline sets the deadline option.

func WithLogger

func WithLogger(v *log.Logger) Option

WithLogger sets the logger option. Can be silenced by setting v to:

log.New(ioutil.Discard, "", 0)

func WithMaxMessages

func WithMaxMessages(v int) Option

WithMaxMessages sets the maximum messages retrieved during a pull. Default = 1.

func WithNoExtend

func WithNoExtend(v bool) Option

WithNoExtend sets the flag to not extend the ack deadline.

type PubsubPublisher

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

func NewPubsubPublisher

func NewPubsubPublisher(projectId string, topicname string) (*PubsubPublisher, error)

Deprecated: The gizmo package used in this function is now unmaintained.

func (*PubsubPublisher) Publish

func (p *PubsubPublisher) Publish(key string, data interface{}) error

func (*PubsubPublisher) RawTopic

func (p *PubsubPublisher) RawTopic() *gpubsub.Topic

func (*PubsubPublisher) Test

func (p *PubsubPublisher) Test() error