kgox

Documentation

Index

• type DefaultMarshaler
  • func (m *DefaultMarshaler) Marshal(ctx context.Context, msg *messagex.Message, topic string) (*kgo.Record, error)
  • func (m *DefaultMarshaler) Unmarshal(r *kgo.Record) (*messagex.Message, error)
• type KgoxAdminClient
  • func NewPubSubAdminClient(cl *kgo.Client, config *pubsubx.Config, ...) *KgoxAdminClient
  • func (p *KgoxAdminClient) CreateTopic(ctx context.Context, partitions int32, replicationFactor int16, topic string, ...) (kadm.CreateTopicResponse, error)
  • func (p *KgoxAdminClient) CreateTopics(ctx context.Context, partitions int32, replicationFactor int16, ...) (kadm.CreateTopicResponses, error)
  • func (p *KgoxAdminClient) DeleteGroup(ctx context.Context, group messagex.ConsumerGroup) (kadm.DeleteGroupResponse, error)
  • func (p *KgoxAdminClient) DeleteGroups(ctx context.Context, groups ...messagex.ConsumerGroup) (kadm.DeleteGroupResponses, error)
  • func (p *KgoxAdminClient) DeleteTopicWithRetryTopics(ctx context.Context, topic string) (kadm.DeleteTopicResponses, error)
  • func (p *KgoxAdminClient) DeleteTopicsWithRetryTopics(ctx context.Context, topics ...string) (kadm.DeleteTopicResponses, error)
  • func (p *KgoxAdminClient) HealthCheck(ctx context.Context) error
• type Marshaler
• type PoisonQueueHandler
• type PubSub
  • func NewPubSub(l *logrusx.Logger, config *pubsubx.Config, opts *pubsubx.PubSubOptions) (*PubSub, error)
  • func (p *PubSub) AdminClient() (pubsubx.PubSubAdminClient, error)
  • func (p *PubSub) Close() error
  • func (p *PubSub) PoisonQueueHandler() PoisonQueueHandler
  • func (p *PubSub) Publisher() pubsubx.Publisher
  • func (p *PubSub) Subscriber(group string, topics []messagex.Topic, opts ...pubsubx.SubscriberOption) (pubsubx.Subscriber, error)
• type Unmarshaler

Types

type DefaultMarshaler

type DefaultMarshaler struct{}

func (*DefaultMarshaler) Marshal

func (m *DefaultMarshaler) Marshal(ctx context.Context, msg *messagex.Message, topic string) (*kgo.Record, error)

func (*DefaultMarshaler) Unmarshal

func (m *DefaultMarshaler) Unmarshal(r *kgo.Record) (*messagex.Message, error)

Unmarshal implements Unmarshaler.

type KgoxAdminClient

type KgoxAdminClient struct {
	*kadm.Client
	*pubsubx.Config
	// contains filtered or unexported fields
}

func NewPubSubAdminClient

func NewPubSubAdminClient(cl *kgo.Client, config *pubsubx.Config, defaultCreateTopicConfigEntries map[string]*string) *KgoxAdminClient

func (*KgoxAdminClient) CreateTopic

func (p *KgoxAdminClient) CreateTopic(ctx context.Context, partitions int32, replicationFactor int16, topic string, configs ...map[string]*string) (kadm.CreateTopicResponse, error)

CreateTopic implements PubSubAdminClient. Subtle: this method shadows the method (*Client).CreateTopic of pubsubAdminClient.Client.

func (*KgoxAdminClient) CreateTopics

func (p *KgoxAdminClient) CreateTopics(ctx context.Context, partitions int32, replicationFactor int16, topics []string, configs ...map[string]*string) (kadm.CreateTopicResponses, error)

CreateTopics implements PubSubAdminClient. Subtle: this method shadows the method (*Client).CreateTopics of pubsubAdminClient.Client.

func (*KgoxAdminClient) DeleteGroup

func (p *KgoxAdminClient) DeleteGroup(ctx context.Context, group messagex.ConsumerGroup) (kadm.DeleteGroupResponse, error)

DeleteGroup implements PubSubAdminClient. Subtle: this method shadows the method (*Client).DeleteGroup of pubsubAdminClient.Client.

func (*KgoxAdminClient) DeleteGroups

func (p *KgoxAdminClient) DeleteGroups(ctx context.Context, groups ...messagex.ConsumerGroup) (kadm.DeleteGroupResponses, error)

DeleteGroups implements PubSubAdminClient. Subtle: this method shadows the method (*Client).DeleteGroups of pubsubAdminClient.Client.

func (*KgoxAdminClient) DeleteTopicWithRetryTopics

func (p *KgoxAdminClient) DeleteTopicWithRetryTopics(ctx context.Context, topic string) (kadm.DeleteTopicResponses, error)

func (*KgoxAdminClient) DeleteTopicsWithRetryTopics

func (p *KgoxAdminClient) DeleteTopicsWithRetryTopics(ctx context.Context, topics ...string) (kadm.DeleteTopicResponses, error)

func (*KgoxAdminClient) HealthCheck

func (p *KgoxAdminClient) HealthCheck(ctx context.Context) error

HealthCheck implements pubsubx.PubSubAdminClient.

type Marshaler

type Marshaler interface {
	// Marshal marshals a message into a Kafka record.
	Marshal(ctx context.Context, m *messagex.Message, topic string) (*kgo.Record, error)
}

type PoisonQueueHandler

type PoisonQueueHandler interface {
	PublishMessagesToPoisonQueue(ctx context.Context, topic string, consumerGroup messagex.ConsumerGroup, msgErrs []error, msgs []*messagex.Message) error
	PublishMessagesToPoisonQueueWithGenericError(ctx context.Context, topic string, consumerGroup messagex.ConsumerGroup, msgErr error, msgs ...*messagex.Message) error
	CanUsePoisonQueue() bool
	ConsumeQueue(ctx context.Context, handler pubsubx.Handler) ([]error, error)
}

type PubSub

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

func NewPubSub

func NewPubSub(l *logrusx.Logger, config *pubsubx.Config, opts *pubsubx.PubSubOptions) (*PubSub, error)

func (*PubSub) AdminClient

func (p *PubSub) AdminClient() (pubsubx.PubSubAdminClient, error)

AdminClient implements pubsubx.PubSub.

func (*PubSub) Close

func (p *PubSub) Close() error

Close implements pubsubx.PubSub.

func (*PubSub) PoisonQueueHandler

func (p *PubSub) PoisonQueueHandler() PoisonQueueHandler

PoisonQueueHandler implements pubsubx.PubSub.

func (*PubSub) Publisher

func (p *PubSub) Publisher() pubsubx.Publisher

Publisher implements pubsubx.PubSub.

func (*PubSub) Subscriber

func (p *PubSub) Subscriber(group string, topics []messagex.Topic, opts ...pubsubx.SubscriberOption) (pubsubx.Subscriber, error)

Subscriber implements pubsubx.PubSub.

type Unmarshaler

type Unmarshaler interface {
	// Unmarshal unmarshals a Kafka record into a message.
	Unmarshal(r *kgo.Record) (*messagex.Message, error)
}