conn

Documentation

Index

• type AuthConfig
• type Conn
  • func NewTCPConn(addr string, timeout time.Duration) (*Conn, error)
  • func NewTLSConn(addr string, tlsCfg *tls.Config, timeout time.Duration) (*Conn, error)
  • func (c *Conn) Close() error
  • func (c *Conn) Closed() <-chan struct{}
  • func (c *Conn) Read(frameHandler func(f frame.Frame)) error
  • func (c *Conn) SendPayloadCmd(cmd api.BaseCommand, metadata api.MessageMetadata, payload []byte) error
  • func (c *Conn) SendSimpleCmd(cmd api.BaseCommand) error
• type Connector
  • func NewConnector(s frame.CmdSender, dispatcher *frame.Dispatcher, ac AuthConfig) *Connector
  • func (c *Connector) Connect(ctx context.Context, authMethod, proxyBrokerURL string) (*api.CommandConnected, error)
• type MockPulsarServer
  • func NewMockPulsarServer(ctx context.Context) (*MockPulsarServer, error)

Types

type AuthConfig

type AuthConfig struct {
	AuthMethod string
	AuthData   []byte
}

type Conn

type Conn struct {
	Rc io.ReadCloser

	Wmu sync.Mutex // protects w to ensure frames aren't interleaved
	W   io.Writer

	Cmu      sync.Mutex // protects following
	IsClosed bool
	Closedc  chan struct{}
}

Conn is responsible for writing and reading Frames to and from the underlying connection (r and w).

func NewTCPConn

func NewTCPConn(addr string, timeout time.Duration) (*Conn, error)

NewTCPConn creates a core using a TCPv4 connection to the given (pulsar server) address.

func NewTLSConn

func NewTLSConn(addr string, tlsCfg *tls.Config, timeout time.Duration) (*Conn, error)

NewTLSConn creates a core using a TCPv4+TLS connection to the given (pulsar server) address.

func (*Conn) Close

func (c *Conn) Close() error

Close closes the underlaying connection. This will cause read() to unblock and return an error. It will also cause the closed channel to unblock.

func (*Conn) Closed

func (c *Conn) Closed() <-chan struct{}

Closed returns a channel that will unblock when the connection has been closed and is no longer usable.

func (*Conn) Read

func (c *Conn) Read(frameHandler func(f frame.Frame)) error

Read blocks while it reads from r until an error occurs. It passes all frames to the provided handler, sequentially and from the same goroutine as called with. Any error encountered will close the connection. Also if close() is called, read() will unblock. Once read returns, the core should be considered unusable.

func (*Conn) SendPayloadCmd

func (c *Conn) SendPayloadCmd(cmd api.BaseCommand, metadata api.MessageMetadata, payload []byte) error

SendPayloadCmd writes a "payload" frame to the wire. It is safe to use concurrently.

func (*Conn) SendSimpleCmd

func (c *Conn) SendSimpleCmd(cmd api.BaseCommand) error

SendSimpleCmd writes a "simple" frame to the wire. It is safe to use concurrently.

type Connector

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

connector encapsulates the logic for the CONNECT <-> (CONNECTED|ERROR) request-response cycle.

https://pulsar.incubator.apache.org/docs/latest/project/BinaryProtocol/#Connectionestablishment-ly8l2n

func NewConnector

func NewConnector(s frame.CmdSender, dispatcher *frame.Dispatcher, ac AuthConfig) *Connector

NewConnector returns a ready-to-use connector.

func (*Connector) Connect

func (c *Connector) Connect(ctx context.Context, authMethod, proxyBrokerURL string) (*api.CommandConnected, error)

Connect initiates the client's session. After sending, the client should wait for a `Connected` or `Error` response from the server.

The provided context should have a timeout associated with it.

It's required to have completed Connect/Connected before using the client.

type MockPulsarServer

type MockPulsarServer struct {
	Addr  string
	Errs  chan error
	Conns chan *Conn
}

MockPulsarServer emulates a Pulsar server

func NewMockPulsarServer

func NewMockPulsarServer(ctx context.Context) (*MockPulsarServer, error)