segments

Documentation

Index

• type Config
• type Marshaler
• type MarshalerWrapper
  • func (kc MarshalerWrapper) Marshal(message pubsubs.Message) (segment.Message, error)
• type PubSubFactoryGenerator
  • func PubSubFactory(publishers PublisherHandler, subscribers SubscriberHandler, ...) PubSubFactoryGenerator
• type 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) (*Publisher, error)
  • func (pf *PublisherSubscriberFactory) QueueSubscribe(topic string, grp string, id string, receiver pubsubs.Receiver) (*Subscription, error)
  • func (pf *PublisherSubscriberFactory) Subscribe(topic string, id string, receiver pubsubs.Receiver) (*Subscription, error)
  • func (pf *PublisherSubscriberFactory) Wait()
• type QueueGroupSubscriberHandler
• type SubscriberHandler
• type Subscription
  • func (s *Subscription) Error() error
  • func (s *Subscription) Group() string
  • func (s *Subscription) ID() string
  • func (s *Subscription) Stop() error
  • func (s *Subscription) Topic() string
• type Unmarshaler
• type UnmarshalerWrapper
  • func (kc UnmarshalerWrapper) Unmarshal(message segment.Message) (pubsubs.Message, error)

Types

type Config

type Config struct {
	Brokers                []string
	ProjectID              string
	MinMessageSize         uint64
	MaxMessageSize         uint64
	AutoCommit             bool
	MessageDeliveryTimeout time.Duration
	MaxAckInterval         time.Duration
	Marshaler              Marshaler
	Unmarshaler            Unmarshaler
	Log                    actorkit.Logs
	Dialer                 *segment.Dialer
	Balancer               segment.Balancer
	Compression            segment.CompressionCodec

	// WriterConfigOverride can be provided to set default
	// configuration values for which will be used for creating writers.
	WriterConfigOverride *segment.WriterConfig

	// ReaderConfigOverride can be provided to set default
	// configuration values for which will be used for creating readers.
	ReaderConfigOverride *segment.ReaderConfig
}

Config provides a config struct for instantiating a PublishSubscribeFactory type.

type Marshaler

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

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

type MarshalerWrapper

type MarshalerWrapper struct {
	Envelope pubsubs.Marshaler
}

MarshalerWrapper implements the Marshaler interface.

func (MarshalerWrapper) Marshal

func (kc MarshalerWrapper) Marshal(message pubsubs.Message) (segment.Message, error)

Marshal implements the Marshaler interface.

type PubSubFactoryGenerator

type PubSubFactoryGenerator func(factory *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, groupSubscribers QueueGroupSubscriberHandler) PubSubFactoryGenerator

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

type Publisher

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

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

func (*Publisher) Close

func (p *Publisher) Close() error

Close closes giving subscriber.

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 the publisher is closed.

type PublisherHandler

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

PublisherHandler defines a function type which takes a giving PublisherFactory 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 segment Publisher factory which handles creation of publishers for topic publishing and management.

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) (*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) QueueSubscribe

func (pf *PublisherSubscriberFactory) QueueSubscribe(topic string, grp string, id string, receiver pubsubs.Receiver) (*Subscription, error)

QueueSubscribe returns a new subscription for a giving topic in a given queue group which will be used for processing messages for giving topic from the nats streaming provider. If the topic already has a subscriber then a subscriber with a ever increasing _id is added and returned if a user defined group id is not set, the subscriber receives the giving id as it's queue group name for it's subscription.

Implementation hold's no respect for the id value, it is lost once a subscription is lost.

func (*PublisherSubscriberFactory) Subscribe

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

Subscribe returns a new subscription for a giving topic which will be used for processing messages for giving topic from the nats streaming provider. If the topic already has a subscriber then a subscriber with a ever increasing _id is added. The id value is used as a durable name value for the giving subscription. If one exists then that is returned.

Implementation hold's no respect for the id value, it is lost once a subscription is lost.

func (*PublisherSubscriberFactory) Wait

func (pf *PublisherSubscriberFactory) Wait()

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

type QueueGroupSubscriberHandler

type QueueGroupSubscriberHandler func(p *PublisherSubscriberFactory, topic string, group string, id string, r pubsubs.Receiver) (pubsubs.Subscription, error)

QueueGroupSubscriberHandler defines a function type which will return a subscription for a queue group.

type SubscriberHandler

type SubscriberHandler func(p *PublisherSubscriberFactory, topic string, id string, r pubsubs.Receiver) (pubsubs.Subscription, error)

SubscriberHandler defines a function type which takes a giving SubscriptionFactory 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) Error

func (s *Subscription) Error() error

Error returns any error which was the cause for the stopping of subscription, it will block till subscription ends to get error if not done, so use carefully.

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.

type Unmarshaler

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

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

type UnmarshalerWrapper

type UnmarshalerWrapper struct {
	Envelope pubsubs.Unmarshaler
}

UnmarshalerWrapper implements the Unmarshaler interface.

func (UnmarshalerWrapper) Unmarshal

func (kc UnmarshalerWrapper) Unmarshal(message segment.Message) (pubsubs.Message, error)

Unmarshal implements the Unmarshaler interface.