pub

Documentation

Index

• Variables
• type Producer
  • func NewProducer(s frame.CmdSender, dispatcher *frame.Dispatcher, reqID *msg.MonotonicID, ...) *Producer
  • func (p *Producer) AddTraceHook(th TraceHook)
  • func (p *Producer) Close(ctx context.Context) error
  • func (p *Producer) Closed() <-chan struct{}
  • func (p *Producer) ConnClosed() <-chan struct{}
  • func (p *Producer) HandleCloseProducer(f frame.Frame) error
  • func (p *Producer) Send(ctx context.Context, payload []byte) (*api.CommandSendReceipt, error)
• type TraceHook

Variables

var ErrClosedProducer = errors.New("producer is closed")

ErrClosedProducer is returned when attempting to send from a closed Producer.

Types

type Producer

type Producer struct {
	S frame.CmdSender

	ProducerID   uint64
	ProducerName string

	ReqID *msg.MonotonicID
	SeqID *msg.MonotonicID

	Dispatcher *frame.Dispatcher // handles request/response state

	Mu       sync.RWMutex // protects following
	IsClosed bool
	Closedc  chan struct{}
	// contains filtered or unexported fields
}

Producer is responsible for creating a subscription producer and managing its state.

func NewProducer

func NewProducer(s frame.CmdSender, dispatcher *frame.Dispatcher, reqID *msg.MonotonicID, producerID uint64) *Producer

NewProducer returns a ready-to-use producer. A producer sends messages (type MESSAGE) to Pulsar.

func (*Producer) AddTraceHook

func (p *Producer) AddTraceHook(th TraceHook)

只在初始化Producer的时候执行一次AddTraceHook

func (*Producer) Close

func (p *Producer) Close(ctx context.Context) error

Close closes the producer. When receiving a CloseProducer command, the broker will stop accepting any more messages for the producer, wait until all pending messages are persisted and then reply Success to the client. https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#command-closeproducer

func (*Producer) Closed

func (p *Producer) Closed() <-chan struct{}

Closed returns a channel that will block _unless_ the producer has been closed, in which case the channel will have been closed. TODO: Rename Done

func (*Producer) ConnClosed

func (p *Producer) ConnClosed() <-chan struct{}

ConnClosed unblocks when the producer's connection has been closed. Once that happens, it's necessary to first recreate the client and then the producer.

func (*Producer) HandleCloseProducer

func (p *Producer) HandleCloseProducer(f frame.Frame) error

HandleCloseProducer should be called when a CLOSE_PRODUCER message is received associated with this producer. The broker can send a CloseProducer command to client when it’s performing a graceful failover (eg: broker is being restarted, or the topic is being unloaded by load balancer to be transferred to a different broker).

When receiving the CloseProducer, the client is expected to go through the service discovery lookup again and recreate the producer again. The TCP connection is not being affected. https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#command-closeproducer

func (*Producer) Send

func (p *Producer) Send(ctx context.Context, payload []byte) (*api.CommandSendReceipt, error)

Send sends a message and waits for a SendReceipt.

type TraceHook

type TraceHook interface {
	OnSend(ctx context.Context, msg *api.MessageMetadata, payload []byte)
}