pulsarutils

Documentation

Index

• func CompactAndPublishSequences(ctx *armadacontext.Context, sequences []*armadaevents.EventSequence, ...) error
• func NewMessageId(id int) pulsar.MessageID
• func NewPulsarClient(config *configuration.PulsarConfig) (pulsar.Client, error)
• func PublishSequences(ctx *armadacontext.Context, producer pulsar.Producer, ...) error
• type MockMessageId
• type MockPulsarMessage
  • func NewPulsarMessage(id int, publishTime time.Time, payload []byte) MockPulsarMessage
  • func (m MockPulsarMessage) ID() pulsar.MessageID
  • func (m MockPulsarMessage) Payload() []byte
  • func (m MockPulsarMessage) Properties() map[string]string
  • func (m MockPulsarMessage) PublishTime() time.Time
• type Publisher
• type PulsarPublisher
  • func NewPulsarPublisher(pulsarClient pulsar.Client, producerOptions pulsar.ProducerOptions, ...) (*PulsarPublisher, error)
  • func (p *PulsarPublisher) Close()
  • func (p *PulsarPublisher) PublishMessages(ctx *armadacontext.Context, es *armadaevents.EventSequence) error

Functions

func CompactAndPublishSequences

func CompactAndPublishSequences(ctx *armadacontext.Context, sequences []*armadaevents.EventSequence, producer pulsar.Producer, maxMessageSizeInBytes uint, scheduler schedulers.Scheduler) error

CompactAndPublishSequences reduces the number of sequences to the smallest possible, while respecting per-job set ordering and max Pulsar message size, and then publishes to Pulsar.

func NewMessageId

func NewMessageId(id int) pulsar.MessageID

func NewPulsarClient

func NewPulsarClient(config *configuration.PulsarConfig) (pulsar.Client, error)

func PublishSequences

func PublishSequences(ctx *armadacontext.Context, producer pulsar.Producer, sequences []*armadaevents.EventSequence, scheduler schedulers.Scheduler) error

PublishSequences publishes several event sequences to Pulsar. For efficiency, all sequences are queued for publishing and then flushed. Returns once all sequences have been received by Pulsar.

To reduce the number of separate sequences sent and ensure limit message size, call eventutil.CompactEventSequences(sequences) and eventutil.LimitSequencesByteSize(sequences, int(srv.MaxAllowedMessageSize)) before passing to this function.

Types

type MockMessageId

type MockMessageId struct {
	pulsar.MessageID
	// contains filtered or unexported fields
}

type MockPulsarMessage

type MockPulsarMessage struct {
	pulsar.Message
	// contains filtered or unexported fields
}

func NewPulsarMessage

func NewPulsarMessage(id int, publishTime time.Time, payload []byte) MockPulsarMessage

func (MockPulsarMessage) ID

func (m MockPulsarMessage) ID() pulsar.MessageID

func (MockPulsarMessage) Payload

func (m MockPulsarMessage) Payload() []byte

func (MockPulsarMessage) Properties

func (m MockPulsarMessage) Properties() map[string]string

func (MockPulsarMessage) PublishTime

func (m MockPulsarMessage) PublishTime() time.Time

type Publisher

type Publisher interface {
	PublishMessages(ctx *armadacontext.Context, es *armadaevents.EventSequence) error
	Close()
}

Publisher is an interface to be implemented by structs that handle publishing messages to pulsar

type PulsarPublisher

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

PulsarPublisher is the default implementation of Publisher

func NewPulsarPublisher

func NewPulsarPublisher(
	pulsarClient pulsar.Client,
	producerOptions pulsar.ProducerOptions,
	maxAllowedMessageSize uint,
) (*PulsarPublisher, error)

func (*PulsarPublisher) Close

func (p *PulsarPublisher) Close()

func (*PulsarPublisher) PublishMessages

func (p *PulsarPublisher) PublishMessages(ctx *armadacontext.Context, es *armadaevents.EventSequence) error

PublishMessages publishes all event sequences to pulsar. Event sequences for a given jobset will be combined into single event sequences up to maxMessageBatchSize.