internal

Documentation

Index

• Constants
• func GetPrimaryTokenIndex(token Token, tokenRangeLength int) int
• func GetTokenAtIndex(length int, index int) Token
• func HashToken(key string) Token
• func Murmur3H1(data []byte) int64
• func PrimaryBroker(partitionKey string, brokersLength int) int
• type BackoffPolicy
• type BrokerStatusInfo
  • func NewBrokerStatusInfo() *BrokerStatusInfo
  • func (b *BrokerStatusInfo) IsUp() bool
  • func (b *BrokerStatusInfo) SetAsDown()
  • func (b *BrokerStatusInfo) SetAsReconnected()
  • func (b *BrokerStatusInfo) SetAsReconnecting() bool
  • func (b *BrokerStatusInfo) SetAsUp()
• type Client
  • func NewClient(serviceUrl string, options *ClientOptions) (*Client, error)
  • func (c *Client) Close()
  • func (c *Client) ManualCommit() ConsumerCommitResult
  • func (c *Client) Poll() ConsumerPollResult
  • func (c *Client) ProduceJson(topic string, message io.Reader, partitionKey string) error
  • func (c *Client) RegisterAsConsumer(options ConsumerOptions)
  • func (c *Client) SendStatusRequestToConsumer()
  • func (c *Client) Topology() *Topology
• type ClientOptions
• type Topology

Constants

const DefaultTopologyPollInterval = 10 * time.Second

Functions

func GetPrimaryTokenIndex

func GetPrimaryTokenIndex(token Token, tokenRangeLength int) int

GetPrimaryTokenIndex returns the broker index of the start token in a given range

func GetTokenAtIndex

func GetTokenAtIndex(length int, index int) Token

func HashToken

func HashToken(key string) Token

Gets a token based on a murmur3 hash

func Murmur3H1

func Murmur3H1(data []byte) int64

func PrimaryBroker

func PrimaryBroker(partitionKey string, brokersLength int) int

Types

type BackoffPolicy

type BackoffPolicy interface {
	Reset()

	// Returns the next delay
	Next() time.Duration
}

type BrokerStatusInfo

type BrokerStatusInfo struct {
	// contains filtered or unexported fields
}

func NewBrokerStatusInfo

func NewBrokerStatusInfo() *BrokerStatusInfo

func (*BrokerStatusInfo) IsUp

func (b *BrokerStatusInfo) IsUp() bool

func (*BrokerStatusInfo) SetAsDown

func (b *BrokerStatusInfo) SetAsDown()

func (*BrokerStatusInfo) SetAsReconnected

func (b *BrokerStatusInfo) SetAsReconnected()

func (*BrokerStatusInfo) SetAsReconnecting

func (b *BrokerStatusInfo) SetAsReconnecting() bool

func (*BrokerStatusInfo) SetAsUp

func (b *BrokerStatusInfo) SetAsUp()

type Client

type Client struct {
	// contains filtered or unexported fields
}

func NewClient

func NewClient(serviceUrl string, options *ClientOptions) (*Client, error)

func (*Client) Close

func (c *Client) Close()

func (*Client) ManualCommit

func (c *Client) ManualCommit() ConsumerCommitResult

Performs a manual commit operation on all brokers in the cluster

func (*Client) Poll

func (c *Client) Poll() ConsumerPollResult

func (*Client) ProduceJson

func (c *Client) ProduceJson(topic string, message io.Reader, partitionKey string) error

func (*Client) RegisterAsConsumer

func (c *Client) RegisterAsConsumer(options ConsumerOptions)

func (*Client) SendStatusRequestToConsumer

func (c *Client) SendStatusRequestToConsumer()

func (*Client) Topology

func (c *Client) Topology() *Topology

type ClientOptions

type ClientOptions struct {
	Logger                      Logger
	TopologyPollInterval        time.Duration
	FixedReconnectionDelay      time.Duration
	ProducerInitialize          bool
	ProducerFlushThresholdBytes int
	ProducerConnectionsPerHost  int
}

type Topology

type Topology struct {
	BaseName           string   `json:"baseName,omitempty"` // When defined, it represents the base name to build the broker names, e.g. "polar-"
	Length             int      `json:"length"`             // The ring size
	BrokerNames        []string `json:"names,omitempty"`
	ProducerPort       int      `json:"producerPort"`
	ProducerBinaryPort int      `json:"producerBinaryPort"`
	ConsumerPort       int      `json:"consumerPort"`
}