pulsar

Sep 23, 2024

Changes in this version

type ConsumerOptions

+ BackOffPolicyFunc func() backoff.Policy

type DLQPolicy

+ InitialSubscriptionName string

type ProducerOptions

+ BackOffPolicyFunc func() backoff.Policy

type ReaderOptions

+ BackoffPolicyFunc func() backoff.Policy

Jul 15, 2024

Changes in this version

+ var ErrInvalidAck = errors.New("invalid ack")

+ var ErrMaxConcurrentOpsReached = newError(MaxConcurrentOperationsReached, "Max concurrent operations reached")

+ var ErrTransactionCoordinatorNotEnabled = newError(TransactionCoordinatorNotEnabled, ...)

type AvroSchema

+ Codec avro.Schema

type Consumer

+ GetLastMessageIDs func() ([]TopicMessageID, error)

+ UnsubscribeForce func() error

type ConsumerOptions

+ EnableZeroQueueConsumer bool

type Result

+ const MaxConcurrentOperationsReached

+ const TransactionCoordinatorNotEnabled

+ type TopicMessageID interface

+ Topic func() string

Feb 29, 2024

Changes in this version

type ConsumerOptions

+ StartMessageIDInclusive bool

type Producer

+ FlushWithCtx func(ctx context.Context) error

Jan 22, 2024

Changes in this version

+ var ErrContextExpired = newError(TimeoutError, "message send context expired")

+ var ErrFailAddToBatch = newError(AddToBatchFailed, "message add to batch failed")

+ var ErrInvalidMessage = newError(InvalidMessage, "invalid message")

+ var ErrMemoryBufferIsFull = newError(ClientMemoryBufferIsFull, "client memory buffer is full")

+ var ErrMessageTooLarge = newError(MessageTooBig, "message size exceeds MaxMessageSize")

+ var ErrMetaTooLarge = newError(InvalidMessage, "message metadata size exceeds MaxMessageSize")

+ var ErrProducerBlockedQuotaExceeded = newError(ProducerBlockedQuotaExceededException, "producer blocked")

+ var ErrProducerClosed = newError(ProducerClosed, "producer already been closed")

+ var ErrProducerFenced = newError(ProducerFenced, "producer fenced")

+ var ErrSchema = newError(SchemaFailure, "schema error")

+ var ErrSendQueueIsFull = newError(ProducerQueueIsFull, "producer send queue is full")

+ var ErrSendTimeout = newError(TimeoutError, "message send timeout")

+ var ErrTopicNotfound = newError(TopicNotFound, "topic not found")

+ var ErrTopicTerminated = newError(TopicTerminated, "topic terminated")

+ var ErrTransaction = errors.New("transaction error")

type Reader

+ GetLastMessageID func() (MessageID, error)

type Result

+ const ProducerFenced

Jul 4, 2023

Changes in this version

+ func NewSinglePartitionRouter() func(*ProducerMessage, TopicMetadata) int

type Client

+ NewTransaction func(duration time.Duration) (Transaction, error)

type ClientOptions

+ TLSCipherSuites []uint16

+ TLSMaxVersion uint16

+ TLSMinVersion uint16

type Consumer

+ AckWithTxn func(Message, Transaction) error

type ConsumerOptions

+ SubscriptionMode SubscriptionMode

type ProducerMessage

+ Transaction Transaction

+ type ProtoNativeSchema struct

+ func NewProtoNativeSchemaWithMessage(message proto.Message, properties map[string]string) *ProtoNativeSchema

+ func (ps *ProtoNativeSchema) Decode(data []byte, v interface{}) error

+ func (ps *ProtoNativeSchema) Encode(data interface{}) ([]byte, error)

+ func (ps *ProtoNativeSchema) GetSchemaInfo() *SchemaInfo

+ func (ps *ProtoNativeSchema) Validate(message []byte) error

+ type ProtoNativeSchemaData struct

+ FileDescriptorSet []byte

+ RootFileDescriptorName string

+ RootMessageTypeName string

type SchemaType

+ const ProtoNative

+ type SubscriptionMode int

+ const Durable

+ const NonDurable

type TxnID

+ LeastSigBits uint64

+ MostSigBits uint64

Apr 3, 2023

Changes in this version

+ const PropertyOriginMessageID

+ const TransactionCoordinatorAssign

+ type AckGroupingOptions struct

+ MaxSize uint32

+ MaxTime time.Duration

type ClientOptions

+ ConnectionMaxIdleTime time.Duration

+ EnableTransaction bool

+ MemoryLimitBytes int64

+ TLSCertificateFile string

+ TLSKeyFilePath string

type Consumer

+ AckCumulative func(msg Message) error

+ AckIDCumulative func(msgID MessageID) error

+ ReconsumeLaterWithCustomProperties func(msg Message, customProperties map[string]string, delay time.Duration)

+ type ConsumerEventListener interface

+ BecameActive func(consumer Consumer, topicName string, partition int32)

+ BecameInactive func(consumer Consumer, topicName string, partition int32)

type ConsumerOptions

+ AckGroupingOptions *AckGroupingOptions

+ AutoAckIncompleteChunk bool

+ EnableAutoScaledReceiverQueueSize bool

+ EnableBatchIndexAcknowledgment bool

+ EventListener ConsumerEventListener

+ ExpireTimeOfIncompleteChunk time.Duration

+ MaxPendingChunkedMessage int

type MessageID

+ func NewMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32) MessageID

+ BatchSize func() int32

+ String func() string

+ type ProducerAccessMode int

+ const ProducerAccessModeExclusive

+ const ProducerAccessModeShared

+ const ProducerAccessModeWaitForExclusive

type ProducerOptions

+ ChunkMaxMessageSize uint

+ EnableChunking bool

type ReaderOptions

+ AutoAckIncompleteChunk bool

+ BackoffPolicy internal.BackoffPolicy

+ ExpireTimeOfIncompleteChunk time.Duration

+ MaxPendingChunkedMessage int

type Result

+ const ClientMemoryBufferIsFull

+ const InvalidStatus

+ const TransactionNoFoundError

+ type Transaction interface

+ Abort func(context.Context) error

+ Commit func(context.Context) error

+ GetState func() TxnState

+ GetTxnID func() TxnID

+ type TxnID struct

+ type TxnState int32

+ const TxnAborted

+ const TxnAborting

+ const TxnCommitted

+ const TxnCommitting

+ const TxnError

+ const TxnOpen

+ const TxnTimeout

Sep 29, 2022

Changes in this version

type Authentication

+ func NewAuthenticationBasic(username, password string) (Authentication, error)

type AvroSchema

+ func NewAvroSchemaWithValidation(avroSchemaDef string, properties map[string]string) (*AvroSchema, error)

type Client

+ CreateTableView func(TableViewOptions) (TableView, error)

type ClientOptions

+ KeepAliveInterval time.Duration

+ MetricsRegisterer prometheus.Registerer

type ConsumerOptions

+ AckWithResponse bool

+ BackoffPolicy internal.BackoffPolicy

type DLQPolicy

+ ProducerOptions ProducerOptions

type JSONSchema

+ func NewJSONSchemaWithValidation(jsonAvroSchemaDef string, properties map[string]string) (*JSONSchema, error)

type Message

+ BrokerPublishTime func() *time.Time

+ Index func() *uint64

+ SchemaVersion func() []byte

type ProducerMessage

+ Schema Schema

type ProducerOptions

+ BackoffPolicy internal.BackoffPolicy

+ DisableMultiSchema bool

type ProtoSchema

+ func NewProtoSchemaWithValidation(protoAvroSchemaDef string, properties map[string]string) (*ProtoSchema, error)

type ReaderOptions

+ Schema Schema

+ SubscriptionName string

type Result

+ const SchemaFailure

type Schema

+ func NewSchema(schemaType SchemaType, schemaData []byte, properties map[string]string) (schema Schema, err error)

+ type TableView interface

+ Close func()

+ ContainsKey func(key string) bool

+ Entries func() map[string]interface{}

+ ForEach func(func(string, interface{}) error) error

+ ForEachAndListen func(func(string, interface{}) error) error

+ Get func(key string) interface{}

+ IsEmpty func() bool

+ Keys func() []string

+ Size func() int

+ type TableViewImpl struct

+ func (tv *TableViewImpl) Close()

+ func (tv *TableViewImpl) ContainsKey(key string) bool

+ func (tv *TableViewImpl) Entries() map[string]interface{}

+ func (tv *TableViewImpl) ForEach(action func(string, interface{}) error) error

+ func (tv *TableViewImpl) ForEachAndListen(action func(string, interface{}) error) error

+ func (tv *TableViewImpl) Get(key string) interface{}

+ func (tv *TableViewImpl) IsEmpty() bool

+ func (tv *TableViewImpl) Keys() []string

+ func (tv *TableViewImpl) Size() int

+ type TableViewOptions struct

+ AutoUpdatePartitionsInterval time.Duration

+ Logger log.Logger

+ Schema Schema

+ SchemaValueType reflect.Type

+ Topic string

Feb 10, 2022

Changes in this version

type ConsumerOptions

+ EnableDefaultNackBackoffPolicy bool

+ NackBackoffPolicy NackBackoffPolicy

+ SubscriptionProperties map[string]string

+ type NackBackoffPolicy interface

+ Next func(redeliveryCount uint32) int64

Oct 27, 2021

Changes in this version

type ClientOptions

+ MetricsCardinality MetricsCardinality

type ConsumerOptions

+ Decryption *MessageDecryptionInfo

+ type EncryptionContext struct

+ Algorithm string

+ BatchSize int

+ CompressionType CompressionType

+ Keys map[string]EncryptionKey

+ Param []byte

+ UncompressedSize int

+ type EncryptionKey struct

+ KeyValue []byte

+ Metadata map[string]string

type Message

+ GetEncryptionContext func() *EncryptionContext

+ type MessageDecryptionInfo struct

+ ConsumerCryptoFailureAction int

+ KeyReader crypto.KeyReader

+ MessageCrypto crypto.MessageCrypto

+ type MetricsCardinality int

+ const MetricsCardinalityNamespace

+ const MetricsCardinalityNone

+ const MetricsCardinalityTenant

+ const MetricsCardinalityTopic

+ type ProducerEncryptionInfo struct

+ KeyReader crypto.KeyReader

+ Keys []string

+ MessageCrypto crypto.MessageCrypto

+ ProducerCryptoFailureAction int

type ProducerOptions

+ Encryption *ProducerEncryptionInfo

type ReaderOptions

+ Decryption *MessageDecryptionInfo

type Result

+ const ProducerClosed

Jul 19, 2021

Changes in this version

type MessageID

+ BatchIdx func() int32

+ EntryID func() int64

+ LedgerID func() int64

+ PartitionIdx func() int32

type ProducerMessage

+ DisableReplication bool

type ProducerOptions

+ PartitionsAutoDiscoveryInterval time.Duration

May 11, 2021

Changes in this version

type ClientOptions

+ ListenerName string

type Result

+ const AddToBatchFailed

+ const AlreadyClosedError

+ const AuthenticationError

+ const AuthorizationError

+ const BrokerMetadataError

+ const BrokerPersistenceError

+ const ChecksumError

+ const ConnectError

+ const ConsumerClosed

+ const ConsumerNotFound

+ const ConsumerNotInitialized

+ const CryptoError

+ const ErrorGettingAuthenticationData

+ const InvalidBatchBuilderType

+ const InvalidConfiguration

+ const InvalidMessage

+ const InvalidTopicName

+ const InvalidURL

+ const LookupError

+ const MessageTooBig

+ const NotConnectedError

+ const Ok

+ const OperationNotSupported

+ const ProducerBlockedQuotaExceededError

+ const ProducerBlockedQuotaExceededException

+ const ProducerNotInitialized

+ const ProducerQueueIsFull

+ const ReadError

+ const SeekFailed

+ const ServiceUnitNotReady

+ const TimeoutError

+ const TooManyLookupRequestException

+ const TopicTerminated

+ const UnknownError

+ const UnsupportedVersionError

Feb 9, 2021

Changes in this version

+ func GetBatcherBuilderProvider(typ BatcherBuilderType) (internal.BatcherBuilderProvider, error)

+ type BatcherBuilderType int

+ const DefaultBatchBuilder

+ const KeyBasedBatchBuilder

type ClientOptions

+ CustomMetricsLabels map[string]string

type Message

+ OrderingKey func() string

type ProducerMessage

+ OrderingKey string

type ProducerOptions

+ DisableBlockIfQueueFull bool

+ SendTimeout time.Duration

Nov 16, 2020

Changes in this version

+ const DlqTopicSuffix

+ const IoMaxSize

+ const MaxReconsumeTimes

+ const RetryTopicSuffix

+ const SysPropertyDelayTime

+ const SysPropertyOriginMessageID

+ const SysPropertyRealTopic

+ const SysPropertyReconsumeTimes

+ const SysPropertyRetryTopic

+ func NewDefaultRouter(hashFunc func(string) uint32, maxBatchingMessages uint, maxBatchingSize uint, ...) func(*ProducerMessage, uint32) int

+ func ReadElements(r io.Reader, elements ...interface{}) error

+ func WriteElements(w io.Writer, elements ...interface{}) error

+ type AvroCodec struct

+ Codec *goavro.Codec

+ func NewSchemaDefinition(schema *goavro.Codec) *AvroCodec

+ type AvroSchema struct

+ func NewAvroSchema(avroSchemaDef string, properties map[string]string) *AvroSchema

+ func (as *AvroSchema) Decode(data []byte, v interface{}) error

+ func (as *AvroSchema) Encode(data interface{}) ([]byte, error)

+ func (as *AvroSchema) GetSchemaInfo() *SchemaInfo

+ func (as *AvroSchema) Validate(message []byte) error

+ type BinaryFreeList chan []byte

+ var BinarySerializer BinaryFreeList = make(chan []byte, IoMaxSize)

+ func (b BinaryFreeList) Borrow() (buf []byte)

+ func (b BinaryFreeList) Float32(buf []byte) (float32, error)

+ func (b BinaryFreeList) Float64(buf []byte) (float64, error)

+ func (b BinaryFreeList) PutDouble(datum interface{}) ([]byte, error)

+ func (b BinaryFreeList) PutFloat(datum interface{}) ([]byte, error)

+ func (b BinaryFreeList) PutUint16(w io.Writer, byteOrder binary.ByteOrder, val uint16) error

+ func (b BinaryFreeList) PutUint32(w io.Writer, byteOrder binary.ByteOrder, val uint32) error

+ func (b BinaryFreeList) PutUint64(w io.Writer, byteOrder binary.ByteOrder, val uint64) error

+ func (b BinaryFreeList) PutUint8(w io.Writer, val uint8) error

+ func (b BinaryFreeList) Return(buf []byte)

+ func (b BinaryFreeList) Uint16(r io.Reader, byteOrder binary.ByteOrder) (uint16, error)

+ func (b BinaryFreeList) Uint32(r io.Reader, byteOrder binary.ByteOrder) (uint32, error)

+ func (b BinaryFreeList) Uint64(r io.Reader, byteOrder binary.ByteOrder) (uint64, error)

+ func (b BinaryFreeList) Uint8(r io.Reader) (uint8, error)

+ type BytesSchema struct

+ func NewBytesSchema(properties map[string]string) *BytesSchema

+ func (bs *BytesSchema) Decode(data []byte, v interface{}) error

+ func (bs *BytesSchema) Encode(data interface{}) ([]byte, error)

+ func (bs *BytesSchema) GetSchemaInfo() *SchemaInfo

+ func (bs *BytesSchema) Validate(message []byte) error

type ClientOptions

+ Logger log.Logger

type Consumer

+ ReconsumeLater func(msg Message, delay time.Duration)

type ConsumerOptions

+ KeySharedPolicy *KeySharedPolicy

+ MaxReconnectToBroker *uint

+ RetryEnable bool

+ Schema Schema

type DLQPolicy

+ DeadLetterTopic string

+ RetryLetterTopic string

+ type DoubleSchema struct

+ func NewDoubleSchema(properties map[string]string) *DoubleSchema

+ func (ds *DoubleSchema) Decode(data []byte, v interface{}) error

+ func (ds *DoubleSchema) Encode(value interface{}) ([]byte, error)

+ func (ds *DoubleSchema) GetSchemaInfo() *SchemaInfo

+ func (ds *DoubleSchema) Validate(message []byte) error

+ type FloatSchema struct

+ func NewFloatSchema(properties map[string]string) *FloatSchema

+ func (fs *FloatSchema) Decode(data []byte, v interface{}) error

+ func (fs *FloatSchema) Encode(value interface{}) ([]byte, error)

+ func (fs *FloatSchema) GetSchemaInfo() *SchemaInfo

+ func (fs *FloatSchema) Validate(message []byte) error

+ type Int16Schema struct

+ func NewInt16Schema(properties map[string]string) *Int16Schema

+ func (is16 *Int16Schema) Decode(data []byte, v interface{}) error

+ func (is16 *Int16Schema) Encode(value interface{}) ([]byte, error)

+ func (is16 *Int16Schema) GetSchemaInfo() *SchemaInfo

+ func (is16 *Int16Schema) Validate(message []byte) error

+ type Int32Schema struct

+ func NewInt32Schema(properties map[string]string) *Int32Schema

+ func (is32 *Int32Schema) Decode(data []byte, v interface{}) error

+ func (is32 *Int32Schema) Encode(value interface{}) ([]byte, error)

+ func (is32 *Int32Schema) GetSchemaInfo() *SchemaInfo

+ func (is32 *Int32Schema) Validate(message []byte) error

+ type Int64Schema struct

+ func NewInt64Schema(properties map[string]string) *Int64Schema

+ func (is64 *Int64Schema) Decode(data []byte, v interface{}) error

+ func (is64 *Int64Schema) Encode(value interface{}) ([]byte, error)

+ func (is64 *Int64Schema) GetSchemaInfo() *SchemaInfo

+ func (is64 *Int64Schema) Validate(message []byte) error

+ type Int8Schema struct

+ func NewInt8Schema(properties map[string]string) *Int8Schema

+ func (is8 *Int8Schema) Decode(data []byte, v interface{}) error

+ func (is8 *Int8Schema) Encode(value interface{}) ([]byte, error)

+ func (is8 *Int8Schema) GetSchemaInfo() *SchemaInfo

+ func (is8 *Int8Schema) Validate(message []byte) error

+ type JSONSchema struct

+ func NewJSONSchema(jsonAvroSchemaDef string, properties map[string]string) *JSONSchema

+ func (js *JSONSchema) Decode(data []byte, v interface{}) error

+ func (js *JSONSchema) Encode(data interface{}) ([]byte, error)

+ func (js *JSONSchema) GetSchemaInfo() *SchemaInfo

+ func (js *JSONSchema) Validate(message []byte) error

+ type KeySharedPolicy struct

+ AllowOutOfOrderDelivery bool

+ HashRanges []int

+ Mode KeySharedPolicyMode

+ func NewKeySharedPolicySticky(hashRanges []int) (*KeySharedPolicy, error)

+ type KeySharedPolicyMode int

+ const KeySharedPolicyModeAutoSplit

+ const KeySharedPolicyModeSticky

type Message

+ GetSchemaValue func(v interface{}) error

type ProducerMessage

+ Value interface{}

type ProducerOptions

+ MaxReconnectToBroker *uint

+ Schema Schema

+ type ProtoSchema struct

+ func NewProtoSchema(protoAvroSchemaDef string, properties map[string]string) *ProtoSchema

+ func (ps *ProtoSchema) Decode(data []byte, v interface{}) error

+ func (ps *ProtoSchema) Encode(data interface{}) ([]byte, error)

+ func (ps *ProtoSchema) GetSchemaInfo() *SchemaInfo

+ func (ps *ProtoSchema) Validate(message []byte) error

+ type RetryMessage struct

+ type Schema interface

+ Decode func(data []byte, v interface{}) error

+ Encode func(v interface{}) ([]byte, error)

+ GetSchemaInfo func() *SchemaInfo

+ Validate func(message []byte) error

+ type SchemaInfo struct

+ Name string

+ Properties map[string]string

+ Schema string

+ Type SchemaType

+ type SchemaType int

+ const AUTO

+ const AVRO

+ const AutoConsume

+ const AutoPublish

+ const BOOLEAN

+ const BYTES

+ const DOUBLE

+ const FLOAT

+ const INT16

+ const INT32

+ const INT64

+ const INT8

+ const JSON

+ const KeyValue

+ const NONE

+ const PROTOBUF

+ const STRING

+ type StringSchema struct

+ func NewStringSchema(properties map[string]string) *StringSchema

+ func (ss *StringSchema) Decode(data []byte, v interface{}) error

+ func (ss *StringSchema) Encode(v interface{}) ([]byte, error)

+ func (ss *StringSchema) GetSchemaInfo() *SchemaInfo

+ func (ss *StringSchema) Validate(message []byte) error

Aug 24, 2020

Changes in this version

type Authentication

+ func NewAuthenticationFromTLSCertSupplier(tlsCertSupplier func() (*tls.Certificate, error)) Authentication

+ func NewAuthenticationOAuth2(authParams map[string]string) Authentication

+ type CompressionLevel int

+ const Better

+ const Default

+ const Faster

type Consumer

+ Name func() string

+ type ConsumerInterceptor interface

+ BeforeConsume func(message ConsumerMessage)

+ OnAcknowledge func(consumer Consumer, msgID MessageID)

+ OnNegativeAcksSend func(consumer Consumer, msgIDs []MessageID)

+ type ConsumerInterceptors []ConsumerInterceptor

+ func (x ConsumerInterceptors) BeforeConsume(message ConsumerMessage)

+ func (x ConsumerInterceptors) OnAcknowledge(consumer Consumer, msgID MessageID)

+ func (x ConsumerInterceptors) OnNegativeAcksSend(consumer Consumer, msgIDs []MessageID)

type ConsumerOptions

+ Interceptors ConsumerInterceptors

type Message

+ ProducerName func() string

+ type ProducerInterceptor interface

+ BeforeSend func(producer Producer, message *ProducerMessage)

+ OnSendAcknowledgement func(producer Producer, message *ProducerMessage, msgID MessageID)

+ type ProducerInterceptors []ProducerInterceptor

+ func (x ProducerInterceptors) BeforeSend(producer Producer, message *ProducerMessage)

+ func (x ProducerInterceptors) OnSendAcknowledgement(producer Producer, message *ProducerMessage, msgID MessageID)

type ProducerOptions

+ BatchingMaxSize uint

+ Interceptors ProducerInterceptors

type Reader

+ Seek func(MessageID) error

+ SeekByTime func(time time.Time) error

Jun 11, 2020

Changes in this version

type Authentication

+ func NewAuthenticationAthenz(authParams map[string]string) Authentication

type ClientOptions

+ MaxConnectionsPerBroker int

type Message

+ GetReplicatedFrom func() string

+ IsReplicated func() bool

Mar 26, 2020

Changes in this version

+ var ErrConsumerClosed = errors.New("consumer closed")

+ type Authentication interface

+ func NewAuthentication(name string, params string) (Authentication, error)

+ func NewAuthenticationTLS(certificatePath string, privateKeyPath string) Authentication

+ func NewAuthenticationToken(token string) Authentication

+ func NewAuthenticationTokenFromFile(tokenFilePath string) Authentication

+ func NewAuthenticationTokenFromSupplier(tokenSupplier func() (string, error)) Authentication

+ type Client interface

+ Close func()

+ CreateProducer func(ProducerOptions) (Producer, error)

+ CreateReader func(ReaderOptions) (Reader, error)

+ Subscribe func(ConsumerOptions) (Consumer, error)

+ TopicPartitions func(topic string) ([]string, error)

+ func NewClient(options ClientOptions) (Client, error)

+ type ClientOptions struct

+ ConnectionTimeout time.Duration

+ OperationTimeout time.Duration

+ TLSAllowInsecureConnection bool

+ TLSTrustCertsFilePath string

+ TLSValidateHostname bool

+ URL string

+ type CompressionType int

+ const LZ4

+ const NoCompression

+ const ZLib

+ const ZSTD

+ type Consumer interface

+ Ack func(Message)

+ AckID func(MessageID)

+ Chan func() <-chan ConsumerMessage

+ Close func()

+ Nack func(Message)

+ NackID func(MessageID)

+ Receive func(context.Context) (Message, error)

+ Seek func(MessageID) error

+ SeekByTime func(time time.Time) error

+ Subscription func() string

+ Unsubscribe func() error

+ type ConsumerMessage struct

+ type ConsumerOptions struct

+ AutoDiscoveryPeriod time.Duration

+ DLQ *DLQPolicy

+ MessageChannel chan ConsumerMessage

+ NackRedeliveryDelay time.Duration

+ Name string

+ Properties map[string]string

+ ReadCompacted bool

+ ReceiverQueueSize int

+ ReplicateSubscriptionState bool

+ SubscriptionName string

+ Topic string

+ Topics []string

+ TopicsPattern string

+ Type SubscriptionType

+ type DLQPolicy struct

+ MaxDeliveries uint32

+ Topic string

+ type Error struct

+ func (e *Error) Error() string

+ func (e *Error) Result() Result

+ type HashingScheme int

+ const JavaStringHash

+ const Murmur3_32Hash

+ type Message interface

+ EventTime func() time.Time

+ ID func() MessageID

+ Key func() string

+ Payload func() []byte

+ Properties func() map[string]string

+ PublishTime func() time.Time

+ RedeliveryCount func() uint32

+ Topic func() string

+ type MessageID interface

+ Serialize func() []byte

+ func DeserializeMessageID(data []byte) (MessageID, error)

+ func EarliestMessageID() MessageID

+ func LatestMessageID() MessageID

+ type Producer interface

+ Close func()

+ Flush func() error

+ LastSequenceID func() int64

+ Name func() string

+ Send func(context.Context, *ProducerMessage) (MessageID, error)

+ SendAsync func(context.Context, *ProducerMessage, func(MessageID, *ProducerMessage, error))

+ Topic func() string

+ type ProducerMessage struct

+ DeliverAfter time.Duration

+ DeliverAt time.Time

+ EventTime time.Time

+ Key string

+ Payload []byte

+ Properties map[string]string

+ ReplicationClusters []string

+ SequenceID *int64

+ type ProducerOptions struct

+ BatchingMaxMessages uint

+ BatchingMaxPublishDelay time.Duration

+ DisableBatching bool

+ MaxPendingMessages int

+ MessageRouter func(*ProducerMessage, TopicMetadata) int

+ Name string

+ Properties map[string]string

+ Topic string

+ type Reader interface

+ Close func()

+ HasNext func() bool

+ Next func(context.Context) (Message, error)

+ Topic func() string

+ type ReaderMessage struct

+ type ReaderOptions struct

+ MessageChannel chan ReaderMessage

+ Name string

+ Properties map[string]string

+ ReadCompacted bool

+ ReceiverQueueSize int

+ StartMessageID MessageID

+ StartMessageIDInclusive bool

+ SubscriptionRolePrefix string

+ Topic string

+ type Result int

+ const ConsumerBusy

+ const ResultConnectError

+ const ResultInvalidConfiguration

+ const ResultInvalidTopicName

+ const ResultLookupError

+ const ResultOk

+ const ResultTimeoutError

+ const ResultUnknownError

+ const SubscriptionNotFound

+ const TopicNotFound

+ type SubscriptionInitialPosition int

+ const SubscriptionPositionEarliest

+ const SubscriptionPositionLatest

+ type SubscriptionType int

+ const Exclusive

+ const Failover

+ const KeyShared

+ const Shared

+ type TopicMetadata interface

+ NumPartitions func() uint32