pubsub_channels

README

Overview

pubsub-channels is a tiny library which aims at hiding the usage of GCP's Pub/Sub behind a channel interface

Given a list of subscriptions, pubsub-channels exposes 2 channels:

• a subscription channel from which to read. It contains messages from the list of subscriptions, with the name of their source subscription.
• a publication channel to which to write. It expects messages with their target topic.

Usage

A basic example can be found in the example file

Documentation

Index

• type Options
• type PubMessage
• type PubSubChannels
  • func NewPubSubChannels(ctx context.Context, projectID string, subscriptionIDs []string, ...) (*PubSubChannels, error)
  • func (c *PubSubChannels) NewPubChan() chan<- PubMessage
  • func (c *PubSubChannels) NewSubChan(ctx context.Context) (<-chan SubMessage, <-chan error)
• type SubMessage

Types

type Options

type Options struct {
	// ClientOptions is the options that are passed to the underlying GCP client on creation
	ClientOptions []option.ClientOption
	// ReceiveSettings are the settings that are passed to the pubsub.Subscription on creation
	ReceiveSettings pubsub.ReceiveSettings
	// PublishSettings are the settings that are passed to the pubsub.Topic on creation
	PublishSettings pubsub.PublishSettings
}

Options contains creation options for the gcp client, the topics and the subscriptions

type PubMessage

type PubMessage struct {
	TopicID         string
	Message         *pubsub.Message
	ErrCallback     func(msgID string, err error)
	SuccessCallback func(msgID string)
}

PubMessage bundles a GCP message with its target topic id

type PubSubChannels

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

PubSubChannels wraps a pubsub.Client behind a channel interface.

func NewPubSubChannels

func NewPubSubChannels(ctx context.Context, projectID string, subscriptionIDs []string, options Options) (*PubSubChannels, error)

NewPubSubChannels creates a new *PubSubChannels The ctx is only used to instantiate the GCP client It returns an error if the GCP instantiation fails

func (*PubSubChannels) NewPubChan

func (c *PubSubChannels) NewPubChan() chan<- PubMessage

func (*PubSubChannels) NewSubChan

func (c *PubSubChannels) NewSubChan(ctx context.Context) (<-chan SubMessage, <-chan error)

type SubMessage

type SubMessage struct {
	SubscriptionID string
	Message        *pubsub.Message
}

SubMessage bundles a GCP message with its originating subscription id