kgox

Documentation

Index

• type DefaultMarshaler
  • func (m *DefaultMarshaler) Marshal(msg *messagex.Message, topic string) (*kgo.Record, error)
  • func (m *DefaultMarshaler) Unmarshal(r *kgo.Record) (*messagex.Message, error)
• type KgoxAdminClient
  • func NewPubSubAdminClient(cl *kadm.Client, defaultCreateTopicConfigEntries map[string]*string) *KgoxAdminClient
  • func (p *KgoxAdminClient) CreateTopic(ctx context.Context, partitions int32, replicationFactor int16, topic string, ...) (kadm.CreateTopicResponse, error)
  • func (p *KgoxAdminClient) HealthCheck(ctx context.Context) error
• type Marshaler
• 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) 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(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
	// contains filtered or unexported fields
}

func NewPubSubAdminClient

func NewPubSubAdminClient(cl *kadm.Client, 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) 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(m *messagex.Message, topic string) (*kgo.Record, 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) 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)
}