repository·master·Indexed 41 days ago
https://github.com/lirm/aeron-goA Go implementation of the Aeron messaging client providing high-performance, low-latency messaging via IPC or network protocols. Includes the aeron-go/cluster package for Aeron Cluster service container functionality, supporting snapshotting, timers, and session messaging. Requires the Java media driver, archive, and consensus module to run a cluster, with protocols specified using Simple Binary Encoding (SBE).
The aeron-go/cluster package is a Go implementation of the Aeron Cluster service container. It aims for near one-to-one parity with the original Java classes and methods.
Key Requirements & Constraints:
The implementation is functional and supports the following features:
Note on the Cluster Interface:
The Cluster interface (defined in cluster.go) currently lacks some methods found in the Java equivalent, though these are considered trivial to add.
A subscription allows you to receive messages from an Aeron stream.
a.AddSubscription(channel, streamId). This returns a channel, allowing you to block until the subscription is ready or use a select statement for asynchronous handling.func(buffer *buffers.Atomic, offset int32, length int32, header *logbuffer.Header) to process incoming fragments.subscription.Poll(handler, fragmentLimit) within a loop. Use an idlestrategy (like idlestrategy.Sleeping) to manage CPU usage between polls.subscription := <-a.AddSubscription("aeron:ipc", 10)
defer subscription.Close()
handler := func(buffer *buffers.Atomic, offset int32, length int32, header *logbuffer.Header) {
bytes := buffer.GetBytesArray(offset, length)
fmt.Printf("Received a fragment with payload: %s\n", string(bytes))
}
idleStrategy := idlestrategy.Sleeping{time.Millisecond}
for {
fragmentsRead := subscription.Poll(handler, 10)
idleStrategy.Idle(fragmentsRead)
}A publication allows you to send messages to an Aeron stream.
a.AddPublication(channel, streamId). This returns a channel that yields the publication once it is registered with the driver.buffers.MakeAtomic([]byte) to wrap your message payload.publication.IsConnected() to ensure subscribers are ready before sending.publication.Offer(srcBuffer, offset, length, options).Offer Return Codes:
aeron.NotConnected: The publication is not yet connected to any subscribers.aeron.BackPressured: The publication is being throttled due to congestion.ret >= 0: Success (the specific value indicates the number of fragments written).ret < 0: An unrecognized error code.publication := <-a.AddPublication("aeron:ipc", 10)
defer publication.Close()
// Prepare message
message := "this is a message"
srcBuffer := buffers.MakeAtomic(([]byte)(message))
// Wait for connection
for !publication.IsConnected() {
time.Sleep(time.Millisecond * 10)
}
// Send message
ret := publication.Offer(srcBuffer, 0, int32(len(message)), nil)
switch ret {
case aeron.NotConnected:
log.Print("not connected yet")
case aeron.BackPressured:
log.Print("back pressured")
default:
if ret < 0 {
log.Print("Unrecognized code: %d", ret)
} else {
log.Print("success!")
}
}publication := <-a.AddPublication("aeron:ipc", 10)
defer publication.Close()
message := "this is a message"
srcBuffer := buffers.MakeAtomic(([]byte)(message))
for !publication.IsConnected() {
time.Sleep(time.Millisecond * 10)
}
ret := publication.Offer(srcBuffer, 0, int32(len(message)), nil)
switch ret {
case aeron.NotConnected:
log.Print("not connected yet")
case aeron.BackPressured:
log.Print("back pressured")
default:
if ret < 0 {
log.Print("Unrecognized code: %d", ret)
} else {
log.Print("success!")
}
}To use aeron-go, you must first instantiate an Aeron client by connecting with a Context. You can configure the context with options such as MediaDriverTimeout to control how long the client waits for the media driver.
ctx := aeron.NewContext().MediaDriverTimeout(time.Second * 10)
a := aeron.Connect(ctx)The repository provides two primary examples to help you get started:
examples/cluster/echo_service.go implements a basic echo service, serving as a Go equivalent to the Java EchoService sample.examples/cluster_client/throughput_test_client.go demonstrates how to use the cluster client for high-throughput operations.In aeron-go, flyweights are used to provide a structured view over raw memory buffers (atomic.Buffer). Instead of manually parsing bytes, you use a flyweight struct that implements the Wrap method. This method maps the struct's fields (like Int64Field or StringField) to specific offsets within the provided buffer.
Most flyweights follow a pattern where they contain fixed-width fields followed by an optional variable-length field (like a StringField) at the end. The Wrap method calculates the total size of the message and sets it using m.SetSize(size).
// Example pattern for wrapping a message
err := messageFlyweight.Wrap(buffer, offset)
// Now you can access fields directly
clientID := messageFlyweight.ClientID.Get()The PublicationMessage flyweight provides access to messages related to a publication.
Fields:
ClientID: flyweight.Int64FieldCorrelationID: flyweight.Int64FieldStreamID: flyweight.Int32FieldChannel: flyweight.StringField (Variable length)type PublicationMessage struct {
flyweight.FWBase
ClientID flyweight.Int64Field
CorrelationID flyweight.Int64Field
StreamID flyweight.Int32Field
Channel flyweight.StringField
}The DestinationMessage flyweight provides access to destination-related messages.
Fields:
ClientID: flyweight.Int64FieldCorrelationID: flyweight.Int64FieldRegistrationCorrelationID: flyweight.Int64FieldChannel: flyweight.StringField (Variable length)type DestinationMessage struct {
flyweight.FWBase
ClientID flyweight.Int64Field
CorrelationID flyweight.Int64Field
RegistrationCorrelationID flyweight.Int64Field
Channel flyweight.StringField
}The CounterMessage flyweight is unique because it contains two variable-length fields: key and label. Because the position of the second variable field (label) depends on the length of the first (key), you must use specific methods to update them to ensure the message size and offsets remain correct.
Fields:
ClientID: flyweight.Int64FieldCorrelationID: flyweight.Int64FieldCounterTypeID: flyweight.Int32Fieldkey: flyweight.LengthAndRawDataField (Variable length)label: flyweight.LengthAndRawDataField (Variable length)Methods for updating variable fields:
CopyKeyBuffer(buffer *atomic.Buffer, offset int32, length int32): Copies raw data into the key field.CopyLabelBuffer(buffer *atomic.Buffer, offset int32, length int32): Copies raw data into the label field and recalculates the total message size.CopyLabelString(label string): Copies a string into the label field and recalculates the total message size.Warning: If you wrap the flyweight with a buffer of a different length than the previous one, you must call CopyLabelBuffer to ensure the underlying data is not corrupt.
// Example: Updating a CounterMessage label
counterMsg.CopyLabelString("my-counter-label")The SubscriptionMessage flyweight provides access to subscription-related messages.
Fields:
ClientID: flyweight.Int64FieldCorrelationID: flyweight.Int64FieldRegistrationCorrelationID: flyweight.Int64FieldStreamID: flyweight.Int32FieldChannel: flyweight.StringField (Variable length)type SubscriptionMessage struct {
flyweight.FWBase
ClientID flyweight.Int64Field
CorrelationID flyweight.Int64Field
RegistrationCorrelationID flyweight.Int64Field
StreamID flyweight.Int32Field
Channel flyweight.StringField
}The ImageMessage flyweight provides access to image-related messages, including channel and stream information.
Fields:
CorrelationID: flyweight.Int64FieldSubscriptionRegistrationID: flyweight.Int64FieldStreamID: flyweight.Int32FieldChannel: flyweight.StringField (Variable length)type ImageMessage struct {
flyweight.FWBase
CorrelationID flyweight.Int64Field
SubscriptionRegistrationID flyweight.Int64Field
StreamID flyweight.Int32Field
Channel flyweight.StringField
}