Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

internal

package

Go to main page

Versions in this module

v0

v0.7.4

Dec 25, 2021

v0.7.3

Dec 25, 2021

v0.7.2

Dec 25, 2021

v0.7.1

Dec 25, 2021

Changes in this version

+ const All

+ const BinaryPort

+ const BinaryService

+ const BinaryTLSPort

+ const ClientVersionString

+ const HTTPAdminServiceV1Format

+ const HTTPAdminServiceV2Format

+ const HTTPLookupServiceBasePathV1

+ const HTTPLookupServiceBasePathV2

+ const HTTPPort

+ const HTTPSPort

+ const HTTPSService

+ const HTTPService

+ const HTTPTopicUnderNamespaceV1

+ const HTTPTopicUnderNamespaceV2

+ const MaxFrameSize

+ const MaxMessageSize

+ const MessageFramePadding

+ const NonPersistent

+ const Persistent

+ const PulsarProtocolVersion

+ const PulsarVersion

+ const SSLService

+ var ErrConnectionClosed = errors.New("connection closed")

+ var ErrCorruptedMessage = errors.New("corrupted message")

+ var ErrEOM = errors.New("EOF")

+ var ErrRequestTimeOut = errors.New("request timed out")

+ func CloseIdleConnections(c *http.Client)

+ func ConvertFromStringMap(m map[string]string) []*pb.KeyValue

+ func ConvertToStringMap(pbb []*pb.KeyValue) map[string]string

+ func Crc32cCheckSum(data []byte) uint32

+ func GetAndAdd(n *uint64, diff uint64) uint64

+ func GetTopicRestPath(tn *TopicName) string

+ func IsV2Namespace(namespace string) bool

+ func IsV2TopicName(tn *TopicName) bool

+ func JavaStringHash(s string) uint32

+ func Murmur3_32Hash(s string) uint32

+ func ParseRelativeTimeInSeconds(relativeTime string) (time.Duration, error)

+ func TimestampMillis(t time.Time) uint64

+ func TopicNameWithoutPartitionPart(tn *TopicName) string

+ type Backoff struct

+ func (b *Backoff) Next() time.Duration

+ type BatchBuilder interface

+ Add func(metadata *pb.SingleMessageMetadata, sequenceIDGenerator *uint64, ...) bool

+ Close func() error

+ Flush func() (batchData Buffer, sequenceID uint64, callbacks []interface{}, err error)

+ FlushBatches func() (batchData []Buffer, sequenceID []uint64, callbacks [][]interface{}, ...)

+ IsFull func() bool

+ IsMultiBatches func() bool

+ func NewBatchBuilder(maxMessages uint, maxBatchSize uint, producerName string, producerID uint64, ...) (BatchBuilder, error)

+ func NewKeyBasedBatchBuilder(maxMessages uint, maxBatchSize uint, producerName string, producerID uint64, ...) (BatchBuilder, error)

+ type BatcherBuilderProvider func(maxMessages uint, maxBatchSize uint, producerName string, producerID uint64, ...) (BatchBuilder, error)

+ type BlockingQueue interface

+ CompareAndPoll func(compare func(item interface{}) bool) interface{}

+ Peek func() interface{}

+ PeekLast func() interface{}

+ Poll func() interface{}

+ Put func(item interface{})

+ ReadableSlice func() []interface{}

+ Size func() int

+ Take func() interface{}

+ func NewBlockingQueue(maxSize int) BlockingQueue

+ type Buffer interface

+ Capacity func() uint32

+ Clear func()

+ Get func(readerIndex uint32, size uint32) []byte

+ IsWritable func() bool

+ MoveToFront func()

+ Put func(writerIdx uint32, s []byte)

+ PutUint32 func(n uint32, writerIdx uint32)

+ Read func(size uint32) []byte

+ ReadUint16 func() uint16

+ ReadUint32 func() uint32

+ ReadableBytes func() uint32

+ ReadableSlice func() []byte

+ ReaderIndex func() uint32

+ Resize func(newSize uint32)

+ ResizeIfNeeded func(spaceNeeded uint32)

+ WritableBytes func() uint32

+ WritableSlice func() []byte

+ Write func(s []byte)

+ WriteUint16 func(n uint16)

+ WriteUint32 func(n uint32)

+ WriterIndex func() uint32

+ WrittenBytes func(size uint32)

+ func NewBuffer(size int) Buffer

+ func NewBufferWrapper(buf []byte) Buffer

+ type BuffersPool interface

+ GetBuffer func() Buffer

+ type CheckSum struct

+ func (cs *CheckSum) Write(p []byte) (int, error)

+ type ClientHandlers struct

+ func NewClientHandlers() ClientHandlers

+ func (h *ClientHandlers) Add(c Closable)

+ func (h *ClientHandlers) Close()

+ func (h *ClientHandlers) Del(c Closable)

+ func (h *ClientHandlers) Val(c Closable) bool

+ type Closable interface

+ Close func()

+ type Connection interface

+ AddConsumeHandler func(id uint64, handler ConsumerHandler)

+ Close func()

+ DeleteConsumeHandler func(id uint64)

+ GetMaxMessageSize func() int32

+ ID func() string

+ RegisterListener func(id uint64, listener ConnectionListener)

+ SendRequest func(requestID uint64, req *pb.BaseCommand, callback func(*pb.BaseCommand, error))

+ SendRequestNoWait func(req *pb.BaseCommand) error

+ UnregisterListener func(id uint64)

+ WriteData func(data Buffer)

+ type ConnectionListener interface

+ ConnectionClosed func()

+ ReceivedSendReceipt func(response *pb.CommandSendReceipt)

+ type ConnectionPool interface

+ Close func()

+ GetConnection func(logicalAddr *url.URL, physicalAddr *url.URL) (Connection, error)

+ func NewConnectionPool(tlsOptions *TLSOptions, auth auth.Provider, connectionTimeout time.Duration, ...) ConnectionPool

+ type ConsumerHandler interface

+ ConnectionClosed func()

+ MessageReceived func(response *pb.CommandMessage, headersAndPayload Buffer) error

+ type GetTopicsOfNamespaceMode string

+ type HTTPClient interface

+ Get func(endpoint string, obj interface{}, params map[string]string) error

+ func NewHTTPClient(serviceURL *url.URL, serviceNameResolver ServiceNameResolver, ...) (HTTPClient, error)

+ type LeveledMetrics struct

+ AcksCounter prometheus.Counter

+ BytesPending prometheus.Gauge

+ BytesPublished prometheus.Counter

+ BytesReceived prometheus.Counter

+ ConsumersClosed prometheus.Counter

+ ConsumersOpened prometheus.Counter

+ ConsumersPartitions prometheus.Gauge

+ DlqCounter prometheus.Counter

+ MessagesPending prometheus.Gauge

+ MessagesPublished prometheus.Counter

+ MessagesReceived prometheus.Counter

+ NacksCounter prometheus.Counter

+ PrefetchedBytes prometheus.Gauge

+ PrefetchedMessages prometheus.Gauge

+ ProcessingTime prometheus.Observer

+ ProducersClosed prometheus.Counter

+ ProducersOpened prometheus.Counter

+ ProducersPartitions prometheus.Gauge

+ PublishErrorsMsgTooLarge prometheus.Counter

+ PublishErrorsTimeout prometheus.Counter

+ PublishLatency prometheus.Observer

+ PublishRPCLatency prometheus.Observer

+ ReadersClosed prometheus.Counter

+ ReadersOpened prometheus.Counter

+ type LookupResult struct

+ LogicalAddr *url.URL

+ PhysicalAddr *url.URL

+ type LookupService interface

+ GetPartitionedTopicMetadata func(topic string) (*PartitionedTopicMetadata, error)

+ GetTopicsOfNamespace func(namespace string, mode GetTopicsOfNamespaceMode) ([]string, error)

+ Lookup func(topic string) (*LookupResult, error)

+ func NewHTTPLookupService(httpClient HTTPClient, serviceURL *url.URL, ...) LookupService

+ func NewLookupService(rpcClient RPCClient, serviceURL *url.URL, ...) LookupService

+ type MessageReader struct

+ func NewMessageReader(headersAndPayload Buffer) *MessageReader

+ func NewMessageReaderFromArray(headersAndPayload []byte) *MessageReader

+ func (r *MessageReader) ReadMessage() (*pb.SingleMessageMetadata, []byte, error)

+ func (r *MessageReader) ReadMessageMetadata() (*pb.MessageMetadata, error)

+ func (r *MessageReader) ResetBuffer(buffer Buffer)

+ type Metrics struct

+ ConnectionsClosed prometheus.Counter

+ ConnectionsEstablishmentErrors prometheus.Counter

+ ConnectionsHandshakeErrors prometheus.Counter

+ ConnectionsOpened prometheus.Counter

+ LookupRequestsCount prometheus.Counter

+ PartitionedTopicMetadataRequestsCount prometheus.Counter

+ RPCRequestCount prometheus.Counter

+ func NewMetricsProvider(metricsCardinality int, userDefinedLabels map[string]string) *Metrics

+ func (mp *Metrics) GetLeveledMetrics(t string) *LeveledMetrics

+ type PartitionedTopicMetadata struct

+ Partitions int

+ type PulsarServiceURI struct

+ ServiceHosts []string

+ ServiceInfos []string

+ ServiceName string

+ URL *url.URL

+ func NewPulsarServiceURIFromURIString(uri string) (*PulsarServiceURI, error)

+ func NewPulsarServiceURIFromURL(url *url.URL) (*PulsarServiceURI, error)

+ type RPCClient interface

+ NewConsumerID func() uint64

+ NewProducerID func() uint64

+ NewRequestID func() uint64

+ Request func(logicalAddr *url.URL, physicalAddr *url.URL, requestID uint64, ...) (*RPCResult, error)

+ RequestOnCnx func(cnx Connection, requestID uint64, cmdType pb.BaseCommand_Type, ...) (*RPCResult, error)

+ RequestOnCnxNoWait func(cnx Connection, cmdType pb.BaseCommand_Type, message proto.Message) error

+ RequestToAnyBroker func(requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)

+ func NewRPCClient(serviceURL *url.URL, serviceNameResolver ServiceNameResolver, ...) RPCClient

+ type RPCResult struct

+ Cnx Connection

+ Response *pb.BaseCommand

+ type Semaphore interface

+ Acquire func(ctx context.Context) bool

+ Release func()

+ TryAcquire func() bool

+ func NewSemaphore(maxPermits int32) Semaphore

+ type ServiceNameResolver interface

+ GetAddressList func() []*url.URL

+ GetServiceURI func() *PulsarServiceURI

+ GetServiceURL func() *url.URL

+ ResolveHost func() (*url.URL, error)

+ ResolveHostURI func() (*PulsarServiceURI, error)

+ UpdateServiceURL func(url *url.URL) error

+ func NewPulsarServiceNameResolver(url *url.URL) ServiceNameResolver

+ type TLSOptions struct

+ AllowInsecureConnection bool

+ TrustCertsFilePath string

+ ValidateHostname bool

+ type TopicName struct

+ Domain string

+ Name string

+ Namespace string

+ Partition int

+ Tenant string

+ Topic string

+ func ParseTopicName(topic string) (*TopicName, error)