Documentation
¶
Overview ¶
Package native provides native RabbitMQ stream protocol support for high-throughput scenarios.
This package uses the rabbitmq-stream-go-client library to communicate directly with RabbitMQ using the native stream protocol (port 5552), providing significantly better performance compared to AMQP 0.9.1 for stream operations:
- Sub-millisecond publish latency
- Higher throughput (100k+ messages/second)
- Efficient binary protocol
- Built-in offset tracking
- Automatic batching and compression
Usage:
import "github.com/cloudresty/go-rabbitmq/streams/native"
// Create environment
env, err := native.NewEnvironment(native.EnvironmentOptions{
Host: "localhost",
Port: 5552,
Username: "guest",
Password: "guest",
})
defer env.Close()
// Create a stream
err = env.CreateStream("my-stream", native.StreamOptions{
MaxLengthBytes: 2 * 1024 * 1024 * 1024, // 2GB
})
// Create a producer
producer, err := env.NewProducer("my-stream", native.ProducerOptions{})
defer producer.Close()
// Publish messages
err = producer.Send([]byte("Hello, World!"))
// Create a consumer
consumer, err := env.NewConsumer("my-stream", func(msg native.Message) {
fmt.Printf("Received: %s\n", msg.Body)
}, native.ConsumerOptions{
Offset: native.OffsetFirst,
})
defer consumer.Close()
Index ¶
- type CompressionType
- type ConfirmationHandler
- type Consumer
- type ConsumerOptions
- type Environment
- func (e *Environment) Close() error
- func (e *Environment) CreateStream(name string, opts StreamOptions) error
- func (e *Environment) DeleteStream(name string) error
- func (e *Environment) NewConsumer(streamName string, handler MessageHandler, opts ConsumerOptions) (*Consumer, error)
- func (e *Environment) NewProducer(streamName string, opts ProducerOptions) (*Producer, error)
- func (e *Environment) StreamExists(name string) (bool, error)
- type EnvironmentOptions
- type Message
- type MessageHandler
- type MessageProperties
- type OffsetType
- type Producer
- type ProducerOptions
- type StreamOptions
- type TLSConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CompressionType ¶
type CompressionType int
CompressionType represents the compression algorithm for messages
const ( CompressionNone CompressionType = iota CompressionGzip CompressionSnappy CompressionLz4 CompressionZstd )
type ConfirmationHandler ¶
ConfirmationHandler is the callback for publish confirmations
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer represents a native stream consumer
func (*Consumer) GetStreamName ¶
GetStreamName returns the stream name this consumer is consuming from
func (*Consumer) StoreCustomOffset ¶
StoreCustomOffset stores a custom offset for later recovery
func (*Consumer) StoreOffset ¶
StoreOffset stores the current offset for later recovery
type ConsumerOptions ¶
type ConsumerOptions struct {
Name string
OffsetType OffsetType
Offset int64 // Used when OffsetType is OffsetOffset
OffsetTime time.Time // Used when OffsetType is OffsetTimestamp
CRCCheck bool
InitialCredits int
}
ConsumerOptions configures a stream consumer
type Environment ¶
type Environment struct {
// contains filtered or unexported fields
}
Environment manages connections to RabbitMQ stream protocol
func NewEnvironment ¶
func NewEnvironment(opts EnvironmentOptions) (*Environment, error)
NewEnvironment creates a new stream environment with the given options
func (*Environment) Close ¶
func (e *Environment) Close() error
Close closes the environment and all associated producers/consumers
func (*Environment) CreateStream ¶
func (e *Environment) CreateStream(name string, opts StreamOptions) error
CreateStream creates a new stream with the given options
func (*Environment) DeleteStream ¶
func (e *Environment) DeleteStream(name string) error
DeleteStream deletes a stream
func (*Environment) NewConsumer ¶
func (e *Environment) NewConsumer(streamName string, handler MessageHandler, opts ConsumerOptions) (*Consumer, error)
NewConsumer creates a new consumer for the given stream
func (*Environment) NewProducer ¶
func (e *Environment) NewProducer(streamName string, opts ProducerOptions) (*Producer, error)
NewProducer creates a new producer for the given stream
func (*Environment) StreamExists ¶
func (e *Environment) StreamExists(name string) (bool, error)
StreamExists checks if a stream exists
type EnvironmentOptions ¶
type EnvironmentOptions struct {
Host string
Port int
Username string
Password string
VHost string
MaxProducers int
MaxConsumers int
RequestedHeartbeat time.Duration
RequestedMaxFrameSize int
// TLS configuration
TLSEnabled bool
TLSConfig *TLSConfig
}
EnvironmentOptions configures the stream environment connection
type Message ¶
type Message struct {
Body []byte
Properties MessageProperties
Offset int64
Timestamp time.Time
PublishingID int64
StreamName string
}
Message represents a message received from a stream
type MessageHandler ¶
type MessageHandler func(msg Message)
MessageHandler is the callback function for processing messages
type MessageProperties ¶
type MessageProperties struct {
MessageID string
CorrelationID string
ContentType string
ReplyTo string
Subject string
GroupID string
}
MessageProperties contains AMQP 1.0 message properties
type OffsetType ¶
type OffsetType int
OffsetType represents the type of offset specification for consumers
const ( // OffsetFirst starts consuming from the first available message OffsetFirst OffsetType = iota // OffsetLast starts consuming from the last available message OffsetLast // OffsetNext starts consuming from new messages only OffsetNext // OffsetOffset starts consuming from a specific offset OffsetOffset // OffsetTimestamp starts consuming from a specific timestamp OffsetTimestamp )
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer represents a native stream producer
func (*Producer) GetStreamName ¶
GetStreamName returns the stream name this producer is publishing to
func (*Producer) SendMessage ¶
SendMessage sends a message with properties
type ProducerOptions ¶
type ProducerOptions struct {
Name string
BatchSize int
BatchPublishingDelay time.Duration
MaxInFlight int
Compression CompressionType
ConfirmationHandler ConfirmationHandler
}
ProducerOptions configures a stream producer
Source Files