Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  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

protocol

package

Go to main page

Versions in this module

v2

v2.0.42

Mar 18, 2025

v2.0.41

Mar 18, 2025

v2.0.40

Mar 18, 2025

v2.0.39

Mar 14, 2025

v2.0.38

Mar 13, 2025

v2.0.37

Mar 13, 2025

v2.0.36

Mar 13, 2025

v2.0.35

Mar 13, 2025

v2.0.34

Mar 12, 2025

v2.0.33

Mar 11, 2025

v2.0.32

Mar 11, 2025

v2.0.31

Mar 3, 2025

v2.0.30

Mar 3, 2025

v2.0.29

Mar 2, 2025

Changes in this version

+ const CurrentVersion

+ const DefaultBufferSize

+ const DefaultMaxConns

+ const DefaultQueueSize

+ const DefaultTimeout

+ const ErrCodeChecksum

+ const ErrCodeCompression

+ const ErrCodeConnection

+ const ErrCodeDeserialize

+ const ErrCodeFlowControl

+ const ErrCodeHandshake

+ const ErrCodeHeartbeat

+ const ErrCodeInvalidLength

+ const ErrCodeMessageAge

+ const ErrCodePanic

+ const ErrCodeQueueFull

+ const ErrCodeRead

+ const ErrCodeReconnect

+ const ErrCodeSequence

+ const ErrCodeSerialize

+ const ErrCodeSize

+ const ErrCodeTimeout

+ const ErrCodeVersion

+ const ErrCodeWrite

+ const FeatureCompression

+ const FeatureEncryption

+ const FeatureHeartbeat

+ const FeatureReconnect

+ const HandshakeStatusAuthError

+ const HandshakeStatusError

+ const HandshakeStatusOK

+ const HandshakeStatusVersionError

+ const HandshakeTimeout

+ const HeaderSize

+ const MaxLogDataLength

+ const MaxMessageAge

+ const MaxMessageSize

+ const MaxRetries

+ const RateLimitBurst

+ const RateLimitRate

+ const StatusConnected

+ const StatusDisconnected

+ const StatusReconnecting

+ func IsRecoverable(err error) bool

+ func IsTemporaryError(err error) bool

+ func RetryWithBackoff(operation func() error) error

+ func WithTimeout(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc)

+ type AtomicDuration struct

+ func (d *AtomicDuration) Add(val time.Duration) time.Duration

+ func (d *AtomicDuration) Load() time.Duration

+ func (d *AtomicDuration) Store(val time.Duration)

+ type BackPressure struct

+ type CompressedMessage struct

+ Compressed []byte

+ Original uint32

+ Type CompressionType

+ func (m *CompressedMessage) Marshal() ([]byte, error)

+ func (m *CompressedMessage) Unmarshal(data []byte) error

+ type CompressionType uint8

+ const CompressGzip

+ const CompressLZ4

+ const CompressNone

+ const CompressSnappy

+ type Compressor interface

+ Compress func([]byte) ([]byte, error)

+ Decompress func([]byte) ([]byte, error)

+ Type func() CompressionType

+ type FlowController struct

+ func NewFlowController(rate, window, threshold int64) *FlowController

+ func (fc *FlowController) Acquire(size int64) error

+ func (fc *FlowController) Release(size int64)

+ type GzipCompressor struct

+ func NewGzipCompressor(level int) *GzipCompressor

+ func (c *GzipCompressor) Compress(data []byte) ([]byte, error)

+ func (c *GzipCompressor) Decompress(data []byte) ([]byte, error)

+ func (c *GzipCompressor) Type() CompressionType

+ type Handler struct

+ func NewHandler(conn net.Conn, opts *Options) *Handler

+ func (h *Handler) Close() error

+ func (h *Handler) GetFeatures() uint32

+ func (h *Handler) GetMetrics() *Metrics

+ func (h *Handler) GetQueueMetrics() (send, receive QueueMetrics)

+ func (h *Handler) GracefulClose(timeout time.Duration) error

+ func (h *Handler) Handshake(authData []byte) error

+ func (h *Handler) IsFeatureEnabled(feature uint32) bool

+ func (h *Handler) ReceiveMessage(msg Message) error

+ func (h *Handler) Reconnect(ctx context.Context) error

+ func (h *Handler) SafeHandle(operation func() error) (err error)

+ func (h *Handler) SendMessage(msg Message) error

+ func (h *Handler) SetCompressor(c Compressor)

+ func (h *Handler) SetDeadline(t time.Time) error

+ func (h *Handler) StartHeartbeat()

+ func (h *Handler) StartQueueProcessor(ctx context.Context)

+ func (h *Handler) StopHeartbeat()

+ type HandshakeRequest struct

+ AuthData []byte

+ Features uint32

+ Timestamp int64

+ Version uint16

+ func (r *HandshakeRequest) Marshal() ([]byte, error)

+ func (r *HandshakeRequest) Unmarshal(data []byte) error

+ type HandshakeResponse struct

+ Features uint32

+ Message string

+ Status uint16

+ Version uint16

+ func (r *HandshakeResponse) Marshal() ([]byte, error)

+ func (r *HandshakeResponse) Unmarshal(data []byte) error

+ type HeartbeatMessage struct

+ Timestamp int64

+ func (m *HeartbeatMessage) Marshal() ([]byte, error)

+ func (m *HeartbeatMessage) Unmarshal(data []byte) error

+ type Message interface

+ Marshal func() ([]byte, error)

+ Unmarshal func([]byte) error

+ type MessageQueue struct

+ func NewMessageQueue(size int, policy QueuePolicy) *MessageQueue

+ func (q *MessageQueue) Clear()

+ func (q *MessageQueue) Close()

+ func (q *MessageQueue) Get(ctx context.Context) (*QueueItem, error)

+ func (q *MessageQueue) GetWithTimeout(ctx context.Context, timeout time.Duration) (*QueueItem, error)

+ func (q *MessageQueue) IsFull() bool

+ func (q *MessageQueue) Len() int

+ func (q *MessageQueue) Metrics() QueueMetrics

+ func (q *MessageQueue) Put(ctx context.Context, msg Message, priority int) error

+ func (q *MessageQueue) PutWithTimeout(ctx context.Context, msg Message, priority int, timeout time.Duration) error

+ type MessageTracker struct

+ func NewMessageTracker(maxSize int) *MessageTracker

+ func (t *MessageTracker) IsDuplicate(seq uint64) bool

+ func (t *MessageTracker) IsOrdered(seq uint64) bool

+ func (t *MessageTracker) Track(seq uint64)

+ type MessageWrapper struct

+ Checksum uint32

+ Payload []byte

+ Sequence uint64

+ Timestamp int64

+ Version uint16

+ func (w *MessageWrapper) Marshal() ([]byte, error)

+ func (w *MessageWrapper) Unmarshal(data []byte) error

+ func (w *MessageWrapper) Validate() error

+ type Metrics struct

+ ActiveConnections atomic.Int64

+ BufferUsage atomic.Int64

+ Errors atomic.Int64

+ LastError error

+ MessagesReceived atomic.Int64

+ MessagesSent atomic.Int64

+ Retries atomic.Int64

+ type Options struct

+ DialTimeout time.Duration

+ HeartBeat time.Duration

+ MaxConns int

+ MaxRetries int

+ ProcessTimeout time.Duration

+ QueuePolicy QueuePolicy

+ QueueSize int

+ Rate int64

+ ReadTimeout time.Duration

+ Reconnect bool

+ Recovery *RecoveryHandler

+ RetryConfig *RetryConfig

+ RetryDelay time.Duration

+ Threshold int64

+ Window int64

+ WriteTimeout time.Duration

+ type ProtocolError struct

+ Code int

+ Err error

+ Message string

+ func (e *ProtocolError) Error() string

+ type QueueItem struct

+ Context context.Context

+ Deadline time.Time

+ Message Message

+ Priority int

+ type QueueMetrics struct

+ Dropped atomic.Int64

+ Length atomic.Int64

+ Processed atomic.Int64

+ WaitTime AtomicDuration

+ type QueuePolicy int

+ const PolicyBlock

+ const PolicyDrop

+ const PolicyOverwrite

+ type RecoveryHandler struct

+ MaxRetries int

+ OnDisconnect func(error)

+ OnPanic func(interface{}, []byte)

+ RetryInterval time.Duration

+ type RetryConfig struct

+ MaxRetries int

+ RetryDelay time.Duration