stream

Jun 26, 2024

Changes in this version

type SuperStreamConsumerOptions

+ Autocommit bool

+ func (s *SuperStreamConsumerOptions) SetManualCommit() *SuperStreamConsumerOptions

Jun 13, 2024

Changes in this version

type EnvironmentOptions

+ RPCTimeout time.Duration

+ func (envOptions *EnvironmentOptions) SetRPCTimeout(timeout time.Duration) *EnvironmentOptions

May 24, 2024

Changes in this version

type SuperStreamConsumerOptions

+ AutoCommitStrategy *AutoCommitStrategy

+ func (s *SuperStreamConsumerOptions) SetAutoCommit(autoCommitStrategy *AutoCommitStrategy) *SuperStreamConsumerOptions

May 3, 2024

Changes in this version

type SuperStreamConsumerOptions

+ Filter *ConsumerFilter

+ func (s *SuperStreamConsumerOptions) SetFilter(filter *ConsumerFilter) *SuperStreamConsumerOptions

type SuperStreamProducerOptions

+ Filter *ProducerFilter

+ func (o SuperStreamProducerOptions) SetFilter(filter *ProducerFilter) *SuperStreamProducerOptions

Apr 12, 2024

Changes in this version

+ const LeaderLocatorBalanced

+ const LeaderLocatorClientLocal

+ const SEED

+ var ErrEnvironmentNotDefined = errors.New("Environment not defined")

+ var ErrMessageRouteNotFound = errors.New("Message Route not found for the message key")

+ var ErrProducerNotFound = errors.New("Producer not found in the SuperStream Producer")

+ var ErrSuperStreamConsumerOptionsNotDefined = errors.New("SuperStreamConsumerOptions not defined.")

+ var ErrSuperStreamProducerOptionsNotDefined = errors.New(...)

+ var SingleActiveConsumerNotSupported = errors.New(...)

+ type BindingsOptions struct

+ Bindings []string

+ LeaderLocator string

+ MaxAge time.Duration

+ MaxLengthBytes *ByteCapacity

+ MaxSegmentSizeBytes *ByteCapacity

+ func NewBindingsOptions(bindings []string) *BindingsOptions

+ func (t *BindingsOptions) SetBalancedLeaderLocator() *BindingsOptions

+ func (t *BindingsOptions) SetClientLocalLocator() *BindingsOptions

+ func (t *BindingsOptions) SetMaxAge(maxAge time.Duration) *BindingsOptions

+ func (t *BindingsOptions) SetMaxLengthBytes(maxLengthBytes *ByteCapacity) *BindingsOptions

+ func (t *BindingsOptions) SetMaxSegmentSizeBytes(maxSegmentSizeBytes *ByteCapacity) *BindingsOptions

+ type CPartitionClose struct

+ Context CPartitionContext

+ Event Event

+ Partition string

+ type CPartitionContext interface

+ ConnectPartition func(partition string, offset OffsetSpecification) error

type Client

+ func (c *Client) DeclareSuperStream(superStream string, options SuperStreamOptions) error

+ func (c *Client) DeleteSuperStream(superStream string) error

+ func (c *Client) QueryPartitions(superStream string) ([]string, error)

type ConsumerOptions

+ SingleActiveConsumer *SingleActiveConsumer

+ func (c *ConsumerOptions) IsSingleActiveConsumerEnabled() bool

+ func (c *ConsumerOptions) SetSingleActiveConsumer(singleActiveConsumer *SingleActiveConsumer) *ConsumerOptions

+ type ConsumerUpdate func(streamName string, isActive bool) OffsetSpecification

type Environment

+ func (env *Environment) DeclareSuperStream(superStreamName string, options SuperStreamOptions) error

+ func (env *Environment) DeleteSuperStream(superStreamName string) error

+ func (env *Environment) NewSuperStreamConsumer(superStream string, messagesHandler MessagesHandler, ...) (*SuperStreamConsumer, error)

+ func (env *Environment) NewSuperStreamProducer(superStream string, superStreamProducerOptions *SuperStreamProducerOptions) (*SuperStreamProducer, error)

+ func (env *Environment) QueryPartitions(superStreamName string) ([]string, error)

+ func (env *Environment) QueryRoute(superStream string, routingKey string) ([]string, error)

+ type HashRoutingStrategy struct

+ RoutingKeyExtractor func(message message.StreamMessage) string

+ func NewHashRoutingStrategy(routingKeyExtractor func(message message.StreamMessage) string) *HashRoutingStrategy

+ func (h *HashRoutingStrategy) Route(message message.StreamMessage, partitions []string) ([]string, error)

+ func (h *HashRoutingStrategy) SetRouteParameters(_ string, _ func(superStream string, routingKey string) ([]string, error))

+ type KeyRoutingStrategy struct

+ RoutingKeyExtractor func(message message.StreamMessage) string

+ func NewKeyRoutingStrategy(routingKeyExtractor func(message message.StreamMessage) string) *KeyRoutingStrategy

+ func (k *KeyRoutingStrategy) Route(message message.StreamMessage, partitions []string) ([]string, error)

+ func (k *KeyRoutingStrategy) SetRouteParameters(superStream string, ...)

+ type PPartitionClose struct

+ Context PPartitionContext

+ Event Event

+ Partition string

+ type PPartitionContext interface

+ ConnectPartition func(partition string) error

+ type PartitionPublishConfirm struct

+ ConfirmationStatus []*ConfirmationStatus

+ Partition string

+ type PartitionsOptions struct

+ LeaderLocator string

+ MaxAge time.Duration

+ MaxLengthBytes *ByteCapacity

+ MaxSegmentSizeBytes *ByteCapacity

+ Partitions int

+ func NewPartitionsOptions(partitions int) *PartitionsOptions

+ func (t *PartitionsOptions) SetBalancedLeaderLocator() *PartitionsOptions

+ func (t *PartitionsOptions) SetClientLocalLocator() *PartitionsOptions

+ func (t *PartitionsOptions) SetMaxAge(maxAge time.Duration) *PartitionsOptions

+ func (t *PartitionsOptions) SetMaxLengthBytes(maxLengthBytes *ByteCapacity) *PartitionsOptions

+ func (t *PartitionsOptions) SetMaxSegmentSizeBytes(maxSegmentSizeBytes *ByteCapacity) *PartitionsOptions

+ type RoutingStrategy interface

+ Route func(message message.StreamMessage, partitions []string) ([]string, error)

+ SetRouteParameters func(superStream string, ...)

+ type SingleActiveConsumer struct

+ ConsumerUpdate ConsumerUpdate

+ Enabled bool

+ func NewSingleActiveConsumer(ConsumerUpdate ConsumerUpdate) *SingleActiveConsumer

+ func (s *SingleActiveConsumer) SetEnabled(enabled bool) *SingleActiveConsumer

+ type SuperStreamConsumer struct

+ MessagesHandler MessagesHandler

+ SuperStream string

+ SuperStreamConsumerOptions *SuperStreamConsumerOptions

+ func (s *SuperStreamConsumer) Close() error

+ func (s *SuperStreamConsumer) ConnectPartition(partition string, offset OffsetSpecification) error

+ func (s *SuperStreamConsumer) NotifyPartitionClose(size int) chan CPartitionClose

+ type SuperStreamConsumerOptions struct

+ ClientProvidedName string

+ ConsumerName string

+ Offset OffsetSpecification

+ SingleActiveConsumer *SingleActiveConsumer

+ func NewSuperStreamConsumerOptions() *SuperStreamConsumerOptions

+ func (s *SuperStreamConsumerOptions) SetClientProvidedName(clientProvidedName string) *SuperStreamConsumerOptions

+ func (s *SuperStreamConsumerOptions) SetConsumerName(consumerName string) *SuperStreamConsumerOptions

+ func (s *SuperStreamConsumerOptions) SetOffset(offset OffsetSpecification) *SuperStreamConsumerOptions

+ func (s *SuperStreamConsumerOptions) SetSingleActiveConsumer(singleActiveConsumer *SingleActiveConsumer) *SuperStreamConsumerOptions

+ type SuperStreamOptions interface

+ type SuperStreamProducer struct

+ SuperStream string

+ SuperStreamProducerOptions *SuperStreamProducerOptions

+ func (s *SuperStreamProducer) Close() error

+ func (s *SuperStreamProducer) ConnectPartition(partition string) error

+ func (s *SuperStreamProducer) GetPartitions() []string

+ func (s *SuperStreamProducer) NotifyPartitionClose(size int) chan PPartitionClose

+ func (s *SuperStreamProducer) NotifyPublishConfirmation(size int) chan PartitionPublishConfirm

+ func (s *SuperStreamProducer) Send(message message.StreamMessage) error

+ type SuperStreamProducerOptions struct

+ ClientProvidedName string

+ RoutingStrategy RoutingStrategy

+ func NewSuperStreamProducerOptions(routingStrategy RoutingStrategy) *SuperStreamProducerOptions

+ func (o SuperStreamProducerOptions) SetClientProvidedName(clientProvidedName string) *SuperStreamProducerOptions

Mar 25, 2024

Changes in this version

+ var FilterNotSupported = errors.New(...)

+ func IsVersionGreaterOrEqual(version, target string) bool

+ type ConsumerFilter struct

+ MatchUnfiltered bool

+ PostFilter PostFilter

+ Values []string

+ func NewConsumerFilter(values []string, matchUnfiltered bool, postFilter PostFilter) *ConsumerFilter

type ConsumerOptions

+ Filter *ConsumerFilter

+ func (c *ConsumerOptions) IsFilterEnabled() bool

+ func (c *ConsumerOptions) SetFilter(filter *ConsumerFilter) *ConsumerOptions

+ type FilterValue func(message message.StreamMessage) string

+ type PostFilter func(message *amqp.Message) bool

+ type ProducerFilter struct

+ FilterValue FilterValue

+ func NewProducerFilter(filterValue FilterValue) *ProducerFilter

type ProducerOptions

+ Filter *ProducerFilter

+ func (po *ProducerOptions) IsFilterEnabled() bool

+ func (po *ProducerOptions) SetFilter(filter *ProducerFilter) *ProducerOptions

+ type PublishFilter struct

+ func (p PublishFilter) GetCommandKey() uint16

+ func (p PublishFilter) GetMaxVersion() uint16

+ func (p PublishFilter) GetMinVersion() uint16

+ type Version struct

+ Major int

+ Minor int

+ Patch int

+ func (v Version) Compare(other Version) int

Mar 18, 2024

Changes in this version

type ConsumerContext

+ func (cc ConsumerContext) GetEntriesCount() uint16

Mar 4, 2024

Changes in this version

+ const MetaDataUpdate

+ const SocketClosed

+ var ConfirmationTimoutError = errors.New("Confirmation Timeout Error")

type ConsumerOptions

+ ClientProvidedName string

+ func (c *ConsumerOptions) SetClientProvidedName(clientProvidedName string) *ConsumerOptions

type ProducerOptions

+ ClientProvidedName string

+ ConfirmationTimeOut time.Duration

+ func (po *ProducerOptions) SetClientProvidedName(name string) *ProducerOptions

+ func (po *ProducerOptions) SetConfirmationTimeOut(duration time.Duration) *ProducerOptions

Jun 15, 2023

Changes in this version

+ const SaslConfigurationExternal

+ const SaslConfigurationPlain

type EnvironmentOptions

+ SaslConfiguration *SaslConfiguration

+ func (envOptions *EnvironmentOptions) SetSaslConfiguration(value string) *EnvironmentOptions

+ type SaslConfiguration struct

+ Mechanism string

Apr 26, 2023

Changes in this version

type Consumer

+ func (consumer *Consumer) GetCloseHandler() chan Event

Mar 14, 2023

Changes in this version

type Client

+ func (c *Client) StreamStats(streamName string) (*StreamStats, error)

type Environment

+ func (env *Environment) StreamStats(streamName string) (*StreamStats, error)

+ type StreamStats struct

+ func (s *StreamStats) CommittedChunkId() (int64, error)

+ func (s *StreamStats) FirstOffset() (int64, error)

+ func (s *StreamStats) LastOffset() (int64, error)

Jan 16, 2023

Changes in this version

type ConsumerOptions

+ func (c *ConsumerOptions) SetInitialCredits(initialCredits int16) *ConsumerOptions

Dec 12, 2022

Changes in this version

+ const ClientVersion

+ const CommandClose

+ const CommandDeletePublisher

+ const CommandMetadataUpdate

+ const CommandQueryOffset

+ const CommandUnsubscribe

+ const GZIP

+ const LZ4

+ const LocalhostUriConnection

+ const None

+ const SNAPPY

+ const StreamTcpPort

+ const UnitGb

+ const UnitKb

+ const UnitMb

+ const UnitTb

+ const ZSTD

+ var AlreadyClosed = errors.New("Already Closed")

+ var AuthenticationFailure = errors.New("Authentication Failure")

+ var AuthenticationFailureLoopbackError = errors.New("Authentication Failure Loopback Error")

+ var CodeAccessRefused = errors.New("Resources Access Refused")

+ var ConnectionClosed = errors.New("Can't send the message, connection closed")

+ var FrameTooLarge = errors.New("Frame Too Large, the buffer is too big")

+ var InternalError = errors.New("Internal Error")

+ var LeaderNotReady = errors.New("Leader not Ready yet")

+ var OffsetNotFoundError = errors.New("Offset not found")

+ var PreconditionFailed = errors.New("Precondition Failed")

+ var PublisherDoesNotExist = errors.New("Publisher Does Not Exist")

+ var StreamAlreadyExists = errors.New("Stream Already Exists")

+ var StreamDoesNotExist = errors.New("Stream Does Not Exist")

+ var StreamNotAvailable = errors.New("Stream Not Available")

+ var SubscriptionIdDoesNotExist = errors.New("Subscription Id Does Not Exist")

+ var UnknownFrame = errors.New("Unknown Frame")

+ var VirtualHostAccessFailure = errors.New("Virtual Host Access Failure")

+ func SetLevelInfo(value int8)

+ type AddressResolver struct

+ Host string

+ Port int

+ type AutoCommitStrategy struct

+ func NewAutoCommitStrategy() *AutoCommitStrategy

+ func (ac *AutoCommitStrategy) SetCountBeforeStorage(messageCountBeforeStorage int) *AutoCommitStrategy

+ func (ac *AutoCommitStrategy) SetFlushInterval(flushInterval time.Duration) *AutoCommitStrategy

+ type Broker struct

+ Host string

+ Password string

+ Port string

+ Scheme string

+ Uri string

+ User string

+ Vhost string

+ func (br *Broker) GetUri() string

+ type Brokers struct

+ func (brs *Brokers) Add(brokerReference int16, host string, port uint32) *Broker

+ func (brs *Brokers) Get(brokerReference int16) *Broker

+ type ByteCapacity struct

+ func (byteCapacity ByteCapacity) B(value int64) *ByteCapacity

+ func (byteCapacity ByteCapacity) From(value string) *ByteCapacity

+ func (byteCapacity ByteCapacity) GB(value int64) *ByteCapacity

+ func (byteCapacity ByteCapacity) KB(value int64) *ByteCapacity

+ func (byteCapacity ByteCapacity) MB(value int64) *ByteCapacity

+ func (byteCapacity ByteCapacity) TB(value int64) *ByteCapacity

+ type ChannelClose = <-chan Event

+ type ChannelPublishConfirm chan []*ConfirmationStatus

+ type Client struct

+ func (c *Client) BrokerForConsumer(stream string) (*Broker, error)

+ func (c *Client) BrokerLeader(stream string) (*Broker, error)

+ func (c *Client) Close() error

+ func (c *Client) DeclarePublisher(streamName string, options *ProducerOptions) (*Producer, error)

+ func (c *Client) DeclareStream(streamName string, options *StreamOptions) error

+ func (c *Client) DeclareSubscriber(streamName string, messagesHandler MessagesHandler, options *ConsumerOptions) (*Consumer, error)

+ func (c *Client) DeleteStream(streamName string) error

+ func (c *Client) ReusePublisher(streamName string, existingProducer *Producer) (*Producer, error)

+ func (c *Client) StreamExists(stream string) bool

+ type ClientProperties struct

+ type Code struct

+ type Compression struct

+ func (compression Compression) Gzip() Compression

+ func (compression Compression) Lz4() Compression

+ func (compression Compression) None() Compression

+ func (compression Compression) Snappy() Compression

+ func (compression Compression) String() string

+ func (compression Compression) Zstd() Compression

+ type ConfirmationStatus struct

+ func (cs *ConfirmationStatus) GetError() error

+ func (cs *ConfirmationStatus) GetErrorCode() uint16

+ func (cs *ConfirmationStatus) GetMessage() message.StreamMessage

+ func (cs *ConfirmationStatus) GetProducerID() uint8

+ func (cs *ConfirmationStatus) GetPublishingId() int64

+ func (cs *ConfirmationStatus) IsConfirmed() bool

+ func (cs *ConfirmationStatus) LinkedMessages() []*ConfirmationStatus

+ type ConnectionProperties struct

+ type Consumer struct

+ ID uint8

+ MessagesHandler MessagesHandler

+ func (consumer *Consumer) Close() error

+ func (consumer *Consumer) GetLastStoredOffset() int64

+ func (consumer *Consumer) GetName() string

+ func (consumer *Consumer) GetOffset() int64

+ func (consumer *Consumer) GetStreamName() string

+ func (consumer *Consumer) NotifyClose() ChannelClose

+ func (consumer *Consumer) QueryOffset() (int64, error)

+ func (consumer *Consumer) StoreCustomOffset(offset int64) error

+ func (consumer *Consumer) StoreOffset() error

+ type ConsumerContext struct

+ Consumer *Consumer

+ type ConsumerOptions struct

+ CRCCheck bool

+ ConsumerName string

+ Offset OffsetSpecification

+ func NewConsumerOptions() *ConsumerOptions

+ func (c *ConsumerOptions) SetAutoCommit(autoCommitStrategy *AutoCommitStrategy) *ConsumerOptions

+ func (c *ConsumerOptions) SetCRCCheck(CRCCheck bool) *ConsumerOptions

+ func (c *ConsumerOptions) SetConsumerName(consumerName string) *ConsumerOptions

+ func (c *ConsumerOptions) SetManualCommit() *ConsumerOptions

+ func (c *ConsumerOptions) SetOffset(offset OffsetSpecification) *ConsumerOptions

+ type Coordinator struct

+ func NewCoordinator() *Coordinator

+ func (coordinator *Coordinator) ConsumersCount() int

+ func (coordinator *Coordinator) GetConsumerById(id interface{}) (*Consumer, error)

+ func (coordinator *Coordinator) GetProducerById(id interface{}) (*Producer, error)

+ func (coordinator *Coordinator) GetResponseById(id uint32) (*Response, error)

+ func (coordinator *Coordinator) GetResponseByName(id string) (*Response, error)

+ func (coordinator *Coordinator) NewConsumer(messagesHandler MessagesHandler, parameters *ConsumerOptions) *Consumer

+ func (coordinator *Coordinator) NewProducer(parameters *ProducerOptions) (*Producer, error)

+ func (coordinator *Coordinator) NewResponse(commandId uint16, info ...string) *Response

+ func (coordinator *Coordinator) NewResponseWitName(id string) *Response

+ func (coordinator *Coordinator) Producers() map[interface{}]interface{}

+ func (coordinator *Coordinator) ProducersCount() int

+ func (coordinator *Coordinator) RemoveConsumerById(id interface{}, reason Event) error

+ func (coordinator *Coordinator) RemoveProducerById(id uint8, reason Event) error

+ func (coordinator *Coordinator) RemoveResponseById(id interface{}) error

+ func (coordinator *Coordinator) RemoveResponseByName(id string) error

+ type Environment struct

+ func NewEnvironment(options *EnvironmentOptions) (*Environment, error)

+ func (env *Environment) Close() error

+ func (env *Environment) DeclareStream(streamName string, options *StreamOptions) error

+ func (env *Environment) DeleteStream(streamName string) error

+ func (env *Environment) IsClosed() bool

+ func (env *Environment) NewConsumer(streamName string, messagesHandler MessagesHandler, options *ConsumerOptions) (*Consumer, error)

+ func (env *Environment) NewProducer(streamName string, producerOptions *ProducerOptions) (*Producer, error)

+ func (env *Environment) QueryOffset(consumerName string, streamName string) (int64, error)

+ func (env *Environment) QuerySequence(publisherReference string, streamName string) (int64, error)

+ func (env *Environment) StreamExists(streamName string) (bool, error)

+ func (env *Environment) StreamMetaData(streamName string) (*StreamMetadata, error)

+ type EnvironmentOptions struct

+ AddressResolver *AddressResolver

+ ConnectionParameters []*Broker

+ MaxConsumersPerClient int

+ MaxProducersPerClient int

+ TCPParameters *TCPParameters

+ func NewEnvironmentOptions() *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) IsTLS(val bool) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetAddressResolver(addressResolver AddressResolver) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetHost(host string) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetMaxConsumersPerClient(maxConsumersPerClient int) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetMaxProducersPerClient(maxProducersPerClient int) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetNoDelay(noDelay bool) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetPassword(password string) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetPort(port int) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetReadBuffer(readBuffer int) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetRequestedHeartbeat(requestedHeartbeat time.Duration) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetRequestedMaxFrameSize(requestedMaxFrameSize int) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetTLSConfig(config *tls.Config) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetUri(uri string) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetUris(uris []string) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetUser(user string) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetVHost(vhost string) *EnvironmentOptions

+ func (envOptions *EnvironmentOptions) SetWriteBuffer(writeBuffer int) *EnvironmentOptions

+ type Event struct

+ Command uint16

+ Err error

+ Name string

+ Reason string

+ StreamName string

+ type HeartBeat struct

+ type MessagesHandler func(consumerContext ConsumerContext, message *amqp.Message)

+ type OffsetSpecification struct

+ func (o OffsetSpecification) First() OffsetSpecification

+ func (o OffsetSpecification) Last() OffsetSpecification

+ func (o OffsetSpecification) LastConsumed() OffsetSpecification

+ func (o OffsetSpecification) Next() OffsetSpecification

+ func (o OffsetSpecification) Offset(offset int64) OffsetSpecification

+ func (o OffsetSpecification) String() string

+ func (o OffsetSpecification) Timestamp(offset int64) OffsetSpecification

+ type Producer struct

+ func (producer *Producer) BatchSend(batchMessages []message.StreamMessage) error

+ func (producer *Producer) Close() error

+ func (producer *Producer) FlushUnConfirmedMessages()

+ func (producer *Producer) GetBroker() *Broker

+ func (producer *Producer) GetID() uint8

+ func (producer *Producer) GetLastPublishingId() (int64, error)

+ func (producer *Producer) GetName() string

+ func (producer *Producer) GetOptions() *ProducerOptions

+ func (producer *Producer) GetStreamName() string

+ func (producer *Producer) GetUnConfirmed() map[int64]*ConfirmationStatus

+ func (producer *Producer) NotifyClose() ChannelClose

+ func (producer *Producer) NotifyPublishConfirmation() ChannelPublishConfirm

+ func (producer *Producer) Send(streamMessage message.StreamMessage) error

+ type ProducerOptions struct

+ BatchPublishingDelay int

+ BatchSize int

+ Compression Compression

+ Name string

+ QueueSize int

+ SubEntrySize int

+ func NewProducerOptions() *ProducerOptions

+ func (po *ProducerOptions) SetBatchPublishingDelay(size int) *ProducerOptions

+ func (po *ProducerOptions) SetBatchSize(size int) *ProducerOptions

+ func (po *ProducerOptions) SetCompression(compression Compression) *ProducerOptions

+ func (po *ProducerOptions) SetProducerName(name string) *ProducerOptions

+ func (po *ProducerOptions) SetQueueSize(size int) *ProducerOptions

+ func (po *ProducerOptions) SetSubEntrySize(size int) *ProducerOptions

+ type ReaderProtocol struct

+ CommandID uint16

+ CorrelationId uint32

+ FrameLen uint32

+ Key uint16

+ PublishID uint8

+ PublishingIdCount uint64

+ ResponseCode uint16

+ Version uint16

+ type Response struct

+ type StreamMetadata struct

+ Leader *Broker

+ Replicas []*Broker

+ func (StreamMetadata) New(stream string, responseCode uint16, leader *Broker, replicas []*Broker) *StreamMetadata

+ func (sm StreamMetadata) String() string

+ type StreamOptions struct

+ MaxAge time.Duration

+ MaxLengthBytes *ByteCapacity

+ MaxSegmentSizeBytes *ByteCapacity

+ func NewStreamOptions() *StreamOptions

+ func (s *StreamOptions) SetMaxAge(maxAge time.Duration) *StreamOptions

+ func (s *StreamOptions) SetMaxLengthBytes(maxLength *ByteCapacity) *StreamOptions

+ func (s *StreamOptions) SetMaxSegmentSizeBytes(segmentSize *ByteCapacity) *StreamOptions

+ type StreamsMetadata struct

+ func (StreamsMetadata) New() *StreamsMetadata

+ func (smd *StreamsMetadata) Add(stream string, responseCode uint16, leader *Broker, replicas []*Broker) *StreamMetadata

+ func (smd *StreamsMetadata) Get(stream string) *StreamMetadata

+ type TCPParameters struct

+ NoDelay bool

+ ReadBuffer int

+ RequestedHeartbeat time.Duration

+ RequestedMaxFrameSize int

+ WriteBuffer int

+ type TuneState struct