publish

Documentation

Index

• Constants
• type MessageQueueInitPublisher
  • func NewMessageQueueInitPublisher(publishOptions *MessageQueuePublisherOptions, ...) (*MessageQueueInitPublisher, error)
  • func (mqip *MessageQueueInitPublisher) AddStorageOps(storageOps MessageQueueInitPublisherOperations)
  • func (mqip *MessageQueueInitPublisher) CloseMessageQueues() error
  • func (mqip *MessageQueueInitPublisher) GetMaxBatchSize() int
  • func (mqip *MessageQueueInitPublisher) HandleInitQueueMessages(ctx context.Context) error
• type MessageQueueInitPublisherOperations
• type MessageQueuePublisherOptions
  • func NewMessageQueuePublisherOptions() *MessageQueuePublisherOptions
  • func (mqpo *MessageQueuePublisherOptions) AddFlags(flagSet *pflag.FlagSet)
• type MessageQueueUpdatePublisher
  • func NewMessageQueueUpdatePublisher(messageQueueConfig *messagequeue.MessageQueueConfig, topic string, ...) (*MessageQueueUpdatePublisher, error)
  • func (mqup *MessageQueueUpdatePublisher) CloseMessageQueue() error
  • func (mqup *MessageQueueUpdatePublisher) GetMaxBatchSize() int
  • func (mqup *MessageQueueUpdatePublisher) WriteQueueMessages(messages []kafka.Message) error

Constants

const (
	CompressionCodecNone   = "none"
	CompressionCodecGzip   = "gzip"
	CompressionCodecSnappy = "snappy"
	CompressionCodecLz4    = "lz4"
	CompressionCodecZstd   = "zstd"
)

Types

type MessageQueueInitPublisher

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

func NewMessageQueueInitPublisher

func NewMessageQueueInitPublisher(
	publishOptions *MessageQueuePublisherOptions,
	dataClient coreclientset.Interface, kubeClient kubernetes.Interface,
	messageQueueSpec *lifetimesapi.MessageQueueSpec,
	topic *messagequeue.MessageQueueTopic) (
	*MessageQueueInitPublisher, error)

func (*MessageQueueInitPublisher) AddStorageOps

func (mqip *MessageQueueInitPublisher) AddStorageOps(
	storageOps MessageQueueInitPublisherOperations)

func (*MessageQueueInitPublisher) CloseMessageQueues

func (mqip *MessageQueueInitPublisher) CloseMessageQueues() error

func (*MessageQueueInitPublisher) GetMaxBatchSize

func (mqip *MessageQueueInitPublisher) GetMaxBatchSize() int

func (*MessageQueueInitPublisher) HandleInitQueueMessages

func (mqip *MessageQueueInitPublisher) HandleInitQueueMessages(
	ctx context.Context) error

type MessageQueueInitPublisherOperations

type MessageQueueInitPublisherOperations interface {
	HasInitialData() (bool, error)
	CreateInitialDataMessages(ctx context.Context) ([]kafka.Message, error)
}

type MessageQueuePublisherOptions

type MessageQueuePublisherOptions struct {
	CompressionCodec string
}

func NewMessageQueuePublisherOptions

func NewMessageQueuePublisherOptions() *MessageQueuePublisherOptions

func (*MessageQueuePublisherOptions) AddFlags

func (mqpo *MessageQueuePublisherOptions) AddFlags(flagSet *pflag.FlagSet)

type MessageQueueUpdatePublisher

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

func NewMessageQueueUpdatePublisher

func NewMessageQueueUpdatePublisher(
	messageQueueConfig *messagequeue.MessageQueueConfig, topic string,
	compressionCodecOption string, maxBatchBytes int) (
	*MessageQueueUpdatePublisher, error)

func (*MessageQueueUpdatePublisher) CloseMessageQueue

func (mqup *MessageQueueUpdatePublisher) CloseMessageQueue() error

func (*MessageQueueUpdatePublisher) GetMaxBatchSize

func (mqup *MessageQueueUpdatePublisher) GetMaxBatchSize() int

func (*MessageQueueUpdatePublisher) WriteQueueMessages

func (mqup *MessageQueueUpdatePublisher) WriteQueueMessages(
	messages []kafka.Message) error