kafka

package module

v2.0.7 Latest Latest Go to latest Published: Jan 5, 2024 License: MIT Imports: 13 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/devexps/go-micro

Links

Open Source Insights

Documentation ¶

Index ¶

Constants
func RegisterSubscriber[T any](srv *Server, ctx context.Context, topic, queue string, disableAutoAck bool, ...) error
type Server
- func NewServer(opts ...ServerOption) *Server
type ServerOption
type SubscribeOption
type SubscribeOptionMap
type SubscriberMap
type Transport

Constants ¶

View Source

const (
	KindKafka transport.Kind = "kafka"
)

Variables ¶

This section is empty.

Functions ¶

func RegisterSubscriber ¶ added in v2.0.5

func RegisterSubscriber[T any](srv *Server, ctx context.Context, topic, queue string, disableAutoAck bool, handler func(context.Context, string, broker.Headers, *T) error, opts ...broker.SubscribeOption) error

Types ¶

type Server ¶

type Server struct {
	broker.Broker

	sync.RWMutex
	// contains filtered or unexported fields
}

func NewServer ¶

func NewServer(opts ...ServerOption) *Server

func (*Server) Endpoint ¶

func (s *Server) Endpoint() (*url.URL, error)

func (*Server) Name ¶ added in v2.0.5

func (s *Server) Name() string

func (*Server) RegisterSubscriber ¶

func (s *Server) RegisterSubscriber(ctx context.Context, topic, queue string, disableAutoAck bool, handler broker.Handler, binder broker.Binder, opts ...broker.SubscribeOption) error

RegisterSubscriber registers a subscriber @param ctx is the context @param topic is Subscribe topics @param queue is Subscribe group @param handler is Subscriber handler

func (*Server) Start ¶

func (s *Server) Start(ctx context.Context) error

func (*Server) Stop ¶

func (s *Server) Stop(_ context.Context) error

type ServerOption ¶

type ServerOption func(o *Server)

func WithAddress ¶

func WithAddress(addrs []string) ServerOption

WithAddress .

func WithBrokerOptions ¶

func WithBrokerOptions(opts ...broker.Option) ServerOption

WithBrokerOptions .

func WithCodec ¶

func WithCodec(c string) ServerOption

WithCodec .

func WithEnableKeepAlive ¶ added in v2.0.5

func WithEnableKeepAlive(enable bool) ServerOption

WithEnableKeepAlive enable keep alive

func WithGlobalPropagator ¶

func WithGlobalPropagator() ServerOption

WithGlobalPropagator .

func WithGlobalTracerProvider ¶

func WithGlobalTracerProvider() ServerOption

WithGlobalTracerProvider .

func WithPlainMechanism ¶

func WithPlainMechanism(username, password string) ServerOption

WithPlainMechanism .

func WithPropagator ¶

func WithPropagator(propagators propagation.TextMapPropagator) ServerOption

WithPropagator .

func WithTracerProvider ¶

func WithTracerProvider(provider trace.TracerProvider, tracerName string) ServerOption

WithTracerProvider .

type SubscribeOption ¶

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

type SubscribeOptionMap ¶

type SubscribeOptionMap map[string]*SubscribeOption

type SubscriberMap ¶

type SubscriberMap map[string]broker.Subscriber

type Transport ¶

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

Transport is a Kafka transport.

func (*Transport) Endpoint ¶

func (tr *Transport) Endpoint() string

Endpoint returns the transport endpoint.

func (*Transport) Kind ¶

func (tr *Transport) Kind() transport.Kind

Kind returns the transport kind.

func (*Transport) NodeFilters ¶

func (tr *Transport) NodeFilters() []selector.NodeFilter

NodeFilters returns the client select filters.

func (*Transport) Operation ¶

func (tr *Transport) Operation() string

Operation returns the transport operation.

func (*Transport) ReplyHeader ¶

func (tr *Transport) ReplyHeader() transport.Header

ReplyHeader returns the reply header.

func (*Transport) RequestHeader ¶

func (tr *Transport) RequestHeader() transport.Header

RequestHeader returns the request header.

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL