firestore

Documentation

Index

• func DefaultGenerateSubscriptionName(topicName string) string
• type DefaultMarshaler
  • func (d DefaultMarshaler) Marshal(msg *message.Message) (interface{}, error)
  • func (d DefaultMarshaler) Unmarshal(doc *firestore.DocumentSnapshot) (*message.Message, error)
• type GenerateSubscriptionNameFn
• type Marshaler
• type Message
• type Publisher
  • func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(topic string, messages ...*message.Message) error
  • func (p *Publisher) PublishInTransaction(topic string, t *firestore.Transaction, messages ...*message.Message) error
• type PublisherConfig
• type Subscriber
  • func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) QueueLength(topic string) (int, error)
  • func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
  • func (s *Subscriber) SubscribeInitialize(topic string) error
• type SubscriberConfig
• type TransactionalPublisher
  • func NewTransactionalPublisher(config PublisherConfig, tx *firestore.Transaction, ...) (*TransactionalPublisher, error)
  • func (p *TransactionalPublisher) Close() error
  • func (p *TransactionalPublisher) Publish(topic string, messages ...*message.Message) error

Functions

func DefaultGenerateSubscriptionName

func DefaultGenerateSubscriptionName(topicName string) string

Types

type DefaultMarshaler

type DefaultMarshaler struct{}

func (DefaultMarshaler) Marshal

func (d DefaultMarshaler) Marshal(msg *message.Message) (interface{}, error)

func (DefaultMarshaler) Unmarshal

func (d DefaultMarshaler) Unmarshal(doc *firestore.DocumentSnapshot) (*message.Message, error)

type GenerateSubscriptionNameFn

type GenerateSubscriptionNameFn func(topicName string) string

type Marshaler

type Marshaler interface {
	// Marshal should return a data type which is supported by Firestore.
	// See the docstring on cloud.google.com/go/firestore.DocumentRef.Create.
	Marshal(msg *message.Message) (interface{}, error)

	// Unmarshal should set the following message fields: UUID, Metadata,
	// Payload.
	Unmarshal(doc *firestore.DocumentSnapshot) (*message.Message, error)
}

Marshaler marshals and unmarshals Watermill messages for storage in Firestore.

type Message

type Message struct {
	UUID     string            `firestore:"uuid"`
	Metadata map[string]string `firestore:"metadata"`
	Payload  string            `firestore:"payload"`
}

type Publisher

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

func NewPublisher

func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, messages ...*message.Message) error

func (*Publisher) PublishInTransaction

func (p *Publisher) PublishInTransaction(topic string, t *firestore.Transaction, messages ...*message.Message) error

type PublisherConfig

type PublisherConfig struct {
	// ProjectID is an ID of a Google Cloud project with Firestore database.
	ProjectID string

	// PubSubRootCollection is a name of a collection which will be used as a root collection for the PubSub.
	// It defaults to "pubsub".
	PubSubRootCollection string

	// MessagePublishTimeout is a timeout used for a single `Publish` call.
	// It defaults to 1 minute.
	MessagePublishTimeout time.Duration

	// SubscriptionsCacheValidityDuration is used for internal subscriptions cache
	// in order to reduce fetch calls to Firestore on each `Publish` method call.
	//
	// If you prefer to not cache subscriptions and fetch them each time `Publish`
	// is called, please set `DontCacheSubscriptions` to true.
	//
	// It defaults to 500 milliseconds.
	SubscriptionsCacheValidityDuration time.Duration

	// DontCacheSubscriptions should be set to true when you don't want
	// Publisher to keep an internal cache of subscribers.
	DontCacheSubscriptions bool

	// GoogleClientOpts are options passed directly to firestore client.
	GoogleClientOpts []option.ClientOption

	// Marshaler marshals message from Watermill to Firestore format and vice versa.
	Marshaler Marshaler

	// CustomFirestoreClient can be used to override a default client.
	CustomFirestoreClient client
}

type Subscriber

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

func NewSubscriber

func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) QueueLength

func (s *Subscriber) QueueLength(topic string) (int, error)

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

func (*Subscriber) SubscribeInitialize

func (s *Subscriber) SubscribeInitialize(topic string) error

type SubscriberConfig

type SubscriberConfig struct {
	// ProjectID is an ID of a Google Cloud project with Firestore database.
	ProjectID string

	// GenerateSubscriptionName should accept topic name and construct a subscription name basing on it.
	//
	// It defaults to topic -> topic + "_sub".
	GenerateSubscriptionName GenerateSubscriptionNameFn

	// PubSubRootCollection is a name of a collection which will be used as a root collection for the PubSub.
	//
	// It defaults to "pubsub".
	PubSubRootCollection string

	// Timeout is used for single Firestore operations.
	//
	// It defaults to 30 seconds.
	Timeout time.Duration

	// GoogleClientOpts are options passed directly to firestore client.
	GoogleClientOpts []option.ClientOption

	// Marshaler marshals message from Watermill to Firestore format and vice versa.
	Marshaler Marshaler

	// CustomFirestoreClient can be used to override a default client.
	CustomFirestoreClient client

	// ReadAllPeriod is a period of time between two read-all operations of a subscriber.
	// Read-all operation means that a subscription collection is read and all messages are consumed.
	// It's needed as a workaround for Firestore sometimes ignoring collection changes.
	// Thanks to that we're sure that all messages are consumed with at most ReadAllPeriod delay.
	//
	// It defaults to 10 seconds.
	ReadAllPeriod time.Duration
}

type TransactionalPublisher

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

func NewTransactionalPublisher

func NewTransactionalPublisher(config PublisherConfig, tx *firestore.Transaction, logger watermill.LoggerAdapter) (*TransactionalPublisher, error)

func (*TransactionalPublisher) Close

func (p *TransactionalPublisher) Close() error

func (*TransactionalPublisher) Publish

func (p *TransactionalPublisher) Publish(topic string, messages ...*message.Message) error