srv

Documentation

Index

• type Discoverer
  • func NewDiscoverer(s frame.CmdSender, dispatcher *frame.Dispatcher, reqID *msg.MonotonicID) *Discoverer
  • func (d *Discoverer) LookupTopic(ctx context.Context, topic string, authoritative bool) (*api.CommandLookupTopicResponse, error)
  • func (d *Discoverer) PartitionedMetadata(ctx context.Context, topic string) (*api.CommandPartitionedTopicMetadataResponse, error)
• type Pinger
  • func NewPinger(s frame.CmdSender, dispatcher *frame.Dispatcher) *Pinger
  • func (p *Pinger) HandlePing(msgType api.BaseCommand_Type, msg *api.CommandPing) error
  • func (p *Pinger) Ping(ctx context.Context) error
• type Server
  • func NewServer(ctx context.Context) (*Server, error)
  • func (m *Server) AssertReceived(ctx context.Context, frameTypes ...api.BaseCommand_Type) error
  • func (m *Server) Broadcast(f frame.Frame) error
  • func (m *Server) CloseAll() error
  • func (m *Server) SetIgnoreConnects(ignore bool)
  • func (m *Server) SetIgnorePings(ignore bool)
  • func (m *Server) SetTopicLookupResp(topic, serviceURL string, respType api.CommandLookupTopicResponse_LookupType, ...)
  • func (m *Server) TotalNumConns() int

Types

type Discoverer

type Discoverer struct {
	S          frame.CmdSender
	ReqID      *msg.MonotonicID
	Dispatcher *frame.Dispatcher
}

Discoverer is responsible for topic discovery and metadata lookups.

https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#Servicediscovery-40v5m

func NewDiscoverer

func NewDiscoverer(s frame.CmdSender, dispatcher *frame.Dispatcher, reqID *msg.MonotonicID) *Discoverer

NewDiscoverer returns a ready-to-use discoverer

func (*Discoverer) LookupTopic

func (d *Discoverer) LookupTopic(ctx context.Context, topic string, authoritative bool) (*api.CommandLookupTopicResponse, error)

LookupTopic performs a LOOKUP request for the given topic. The response will determine the proper broker to use for the topic, or indicate that another LOOKUP request is necessary.

https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#Topiclookup-dk72wp

func (*Discoverer) PartitionedMetadata

func (d *Discoverer) PartitionedMetadata(ctx context.Context, topic string) (*api.CommandPartitionedTopicMetadataResponse, error)

PartitionedMetadata performs a PARTITIONED_METADATA request for the given topic. The response can be used to determine how many, if any, partitions there are for the topic.

https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#Partitionedtopicsdiscovery-g14a9h

type Pinger

type Pinger struct {
	S          frame.CmdSender
	Dispatcher *frame.Dispatcher // used to manage the request/response state
}

Pinger is responsible for the PING <-> PONG (Keep Alive) interactions.

It responds to all PING requests with a PONG. It also enables PINGing the Pulsar server.

https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#KeepAlive-53utwq

func NewPinger

func NewPinger(s frame.CmdSender, dispatcher *frame.Dispatcher) *Pinger

NewPinger returns a ready-to-use pinger.

func (*Pinger) HandlePing

func (p *Pinger) HandlePing(msgType api.BaseCommand_Type, msg *api.CommandPing) error

HandlePing responds immediately with a PONG message.

A valid client implementation must respond to PINGs with PONGs, and may optionally send periodic pings.

func (*Pinger) Ping

func (p *Pinger) Ping(ctx context.Context) error

Ping sends a PING message to the Pulsar server, then waits for either a PONG response or the context to timeout.

type Server

type Server struct {
	Addr     string
	Received <-chan frame.Frame
	// contains filtered or unexported fields
}

Server emulates a Pulsar server

func NewServer

func NewServer(ctx context.Context) (*Server, error)

NewServer returns a ready-to-use Pulsar test server. The server will be closed when the context is canceled.

func (*Server) AssertReceived

func (m *Server) AssertReceived(ctx context.Context, frameTypes ...api.BaseCommand_Type) error

AssertReceived accepts a list of message types. It then compares them to the frames it has received, and returns an error if they don't match or if the context times out. Note: The m.Received channel is buffered, so AssertReceived can be called from the main goroutine.

func (*Server) Broadcast

func (m *Server) Broadcast(f frame.Frame) error

Broadcast sends the given frame to all connected clients.

func (*Server) CloseAll

func (m *Server) CloseAll() error

CloseAll closes all connected client connections.

func (*Server) SetIgnoreConnects

func (m *Server) SetIgnoreConnects(ignore bool)

SetIgnoreConnects instructs the server to NOT respond to CONNECT requests if true.

func (*Server) SetIgnorePings

func (m *Server) SetIgnorePings(ignore bool)

SetIgnorePings instructs the server to NOT respond to PING requests if true.

func (*Server) SetTopicLookupResp

func (m *Server) SetTopicLookupResp(topic, serviceURL string, respType api.CommandLookupTopicResponse_LookupType, proxyThroughServiceURL bool)

SetTopicLookupResp updates the BrokerServiceURL returned for the given topic from LOOKUP requests. If not set, the server's Addr is used by default. If connect if false, the response type is REDIRECT.

func (*Server) TotalNumConns

func (m *Server) TotalNumConns() int

TotalNumConns returns the total number of connections (active or inactive) received by the Server.