google

Documentation

Index

• type Config
• type Directive
• type Marshaler
• type PubSubFactoryGenerator
  • func PubSubFactory(publishers PublisherHandler, subscribers SubscriberHandler) PubSubFactoryGenerator
• type PubSubMarshaler
  • func (ps PubSubMarshaler) Marshal(msg pubsubs.Message) (pubsub.Message, error)
• type PubSubUnmarshaler
  • func (ps *PubSubUnmarshaler) Unmarshal(msg *pubsub.Message) (pubsubs.Message, error)
• type Publisher
  • func NewPublisher(ctx context.Context, topic string, sink *pubsub.Topic, config *Config) *Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(msg actorkit.Envelope) error
  • func (p *Publisher) Wait()
• type PublisherHandler
• type PublisherSubscriberFactory
  • func NewPublisherSubscriberFactory(ctx context.Context, config Config) (*PublisherSubscriberFactory, error)
  • func (pf *PublisherSubscriberFactory) Close() error
  • func (pf *PublisherSubscriberFactory) Publisher(topic string, setting *pubsub.PublishSettings) (*Publisher, error)
  • func (pf *PublisherSubscriberFactory) Subscribe(topic string, id string, config *pubsub.SubscriptionConfig, ...) (pubsubs.Subscription, error)
  • func (pf *PublisherSubscriberFactory) Wait()
• type SubscriberHandler
• type Subscription
  • func (s *Subscription) Group() string
  • func (s *Subscription) ID() string
  • func (s *Subscription) Stop() error
  • func (s *Subscription) Topic() string
  • func (s *Subscription) Wait()
• type Unmarshaler

Types

type Config

type Config struct {
	ProjectID   string
	Log         actorkit.Logs
	Marshaler   Marshaler
	Unmarshaler Unmarshaler

	// MessageDeliveryTimeout is the timeout to wait before response
	// from the underline message broker before timeout.
	MessageDeliveryTimeout time.Duration

	// CreateMissingTopic flags dictates if we will create a topic if
	// it does not already exists in the google cloud.
	CreateMissingTopic bool

	// PublishSettings provided customized publishing settings for google pubsub
	// publisher.
	PublishSettings *pubsub.PublishSettings

	// ClientOptions provide options to be applied to create topic subscribers.
	ClientOptions []option.ClientOption

	// ConsumersCount sets the default consumer count to be used during a subscribers
	// internal operations.
	ConsumersCount int

	// MaxOutStandingMessage defines the maximum allowed message awaiting confirmation
	// for subscriptions.
	MaxOutStandingMessages int

	// MaxOutStandingBytes defines the maximum allowed bytes size awaiting confirmation
	// for subscriptions.
	MaxOutStandingBytes int

	// MaxExtension sets the maximum duration to be provided for message delivery extension.
	MaxExtension time.Duration

	// DefaultSubscriptionConfig sets the default configuration to be used in creating subscriptions.
	// This allows setting default values to be used apart from custom set options during instantiation
	// of a subscription.
	DefaultSubscriptionConfig *pubsub.SubscriptionConfig
}

Config provides a config struct for instantiating a Publisher type.

type Directive

type Directive int

Directive defines a int type for representing a giving action to be performed due to an error.

const (
	Ack Directive = iota
	Nack
)

set of possible directives.

type Marshaler

type Marshaler interface {
	Marshal(message pubsubs.Message) (pubsub.Message, error)
}

Marshaler defines a interface exposing method to transform a pubsubs.Message into a kafka message.

type PubSubFactoryGenerator

type PubSubFactoryGenerator func(pub *PublisherSubscriberFactory) pubsubs.PubSubFactory

PubSubFactoryGenerator returns a function which taken a PublisherSubscriberFactory returning a factory for generating publishers and subscribers.

func PubSubFactory

func PubSubFactory(publishers PublisherHandler, subscribers SubscriberHandler) PubSubFactoryGenerator

PubSubFactory provides a partial function for the generation of a pubsubs.PubSubFactory using the PubSubFactorGenerator function.

type PubSubMarshaler

type PubSubMarshaler struct {
	Now       func() time.Time
	Marshaler pubsubs.Marshaler
}

PubSubMarshaler implements the Marshaler interface.

func (PubSubMarshaler) Marshal

func (ps PubSubMarshaler) Marshal(msg pubsubs.Message) (pubsub.Message, error)

Marshal marshals giving message into a pubsub message.

type PubSubUnmarshaler

type PubSubUnmarshaler struct {
	Unmarshaler pubsubs.Unmarshaler
}

PubSubUnmarshaler implements the Unmarshaler interface.

func (*PubSubUnmarshaler) Unmarshal

func (ps *PubSubUnmarshaler) Unmarshal(msg *pubsub.Message) (pubsubs.Message, error)

Unmarshal transforms giving pubsub.Message into a pubsubs.Message type.

type Publisher

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

Publisher implements the topic publishing provider for the google pubsub layer.

func NewPublisher

func NewPublisher(ctx context.Context, topic string, sink *pubsub.Topic, config *Config) *Publisher

NewPublisher returns a new instance of a Publisher.

func (*Publisher) Close

func (p *Publisher) Close() error

Close closes giving publisher and returns any encountered error.

func (*Publisher) Publish

func (p *Publisher) Publish(msg actorkit.Envelope) error

Publish attempts to publish giving message into provided topic publisher returning an error for failed attempt.

func (*Publisher) Wait

func (p *Publisher) Wait()

Wait blocks till publisher is closed.

type PublisherHandler

type PublisherHandler func(*PublisherSubscriberFactory, string) (pubsubs.Publisher, error)

PublisherHandler defines a function type which takes a giving PublisherSubscriberFactory and a given topic, returning a new publisher with all related underline specific details added and instantiated.

type PublisherSubscriberFactory

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

PublisherSubscriberFactory implements a Google pubsub Publisher and Subscriber/Consumer factory which handles creation of publishers and subscribers for topic publishing and consumption.

func NewPublisherSubscriberFactory

func NewPublisherSubscriberFactory(ctx context.Context, config Config) (*PublisherSubscriberFactory, error)

NewPublisherSubscriberFactory returns a new instance of publisher factory.

func (*PublisherSubscriberFactory) Close

func (pf *PublisherSubscriberFactory) Close() error

Close closes giving publisher factory and all previous created publishers.

func (*PublisherSubscriberFactory) Publisher

func (pf *PublisherSubscriberFactory) Publisher(topic string, setting *pubsub.PublishSettings) (*Publisher, error)

Publisher returns giving publisher for giving topic, if provided config allows the creation of publisher if not present then a new publisher is created for topic and returned, else an error is returned if not found or due to some other issues.

func (*PublisherSubscriberFactory) Subscribe

func (pf *PublisherSubscriberFactory) Subscribe(topic string, id string, config *pubsub.SubscriptionConfig, receiver pubsubs.Receiver) (pubsubs.Subscription, error)

Subscribe subscribes to a giving topic, if one exists then a new subscription with a ever incrementing id is assigned to new subscription.

func (*PublisherSubscriberFactory) Wait

func (pf *PublisherSubscriberFactory) Wait()

Wait blocks till all generated publishers close and have being reclaimed.

type SubscriberHandler

type SubscriberHandler func(*PublisherSubscriberFactory, string, string, pubsubs.Receiver) (pubsubs.Subscription, error)

SubscriberHandler defines a function type which takes a giving PublisherSubscriberFactory and a given topic, returning a new subscription with all related underline specific details added and instantiated.

type Subscription

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

Subscription implements a subscriber of a giving topic which is being subscribe to for. It implements the pubsubs.Subscription interface.

func (*Subscription) Group

func (s *Subscription) Group() string

Group returns the group or queue group name of giving subscription.

func (*Subscription) ID

func (s *Subscription) ID() string

ID returns the identification of giving subscription used for durability if supported.

func (*Subscription) Stop

func (s *Subscription) Stop() error

Stop ends giving subscription and it's operation in listening to given topic.

func (*Subscription) Topic

func (s *Subscription) Topic() string

Topic returns the topic name of giving subscription.

func (*Subscription) Wait

func (s *Subscription) Wait()

Wait blocks till a giving subscription is closed.

type Unmarshaler

type Unmarshaler interface {
	Unmarshal(*pubsub.Message) (pubsubs.Message, error)
}

Unmarshaler defines an interface who's implementer exposes said method to transform a kafka message into a pubsubs Message.