Documentation
¶
Overview ¶
Package stream defines the message payloads exchanged over Kafka by the consumer and dispatcher packages.
A Message knows how to serialize itself, how to decode a payload back into a Go value, and how to name its own encoding. Two implementations ship here:
stream.NewJSONMessage(order) // encodes with encoding/json
stream.NewTextMessage("plain") // passes bytes through unchanged
The encoding travels with the message. Dispatch writes the Type of the message into the DEVKIT_CONTENT_TYPE header, and on the way back NewMessageType reads that header to pick the matching implementation. A message that arrives without the header, or with a value nobody recognises, yields ErrUnknownMessageType - which is what makes the consumer route it to the dead letter topic as raw text instead of guessing.
NewWithData produces a new message of the same kind carrying different data. The consumer uses it to re-encode a decoded payload when forwarding it to the dead letter topic.
Index ¶
Examples ¶
Constants ¶
const ( // ContentTypeHeaderKey is the Kafka header naming the encoding of a // message. The dispatcher writes it on produce and NewMessageType reads // it on consume. ContentTypeHeaderKey = "DEVKIT_CONTENT_TYPE" )
Variables ¶
var ( // ErrUnknownMessageType is returned by NewMessageType when the message // carries no ContentTypeHeaderKey, or names an encoding this package does // not implement. ErrUnknownMessageType = errors.New("stream: unknown message type") // ErrStringDeserialization is returned when a text message is decoded // into a destination that is not a *string. ErrStringDeserialization = errors.New("stream: string serialization error") )
Errors returned when a payload cannot be interpreted.
var ( // ErrProcessMessageTimedOut is reported by the dispatcher when the broker // does not acknowledge a produced message within message.timeout.ms. It // separates an unreachable or overloaded broker from a message the broker // actively rejected. ErrProcessMessageTimedOut = errors.New("processing message timed out") )
Functions ¶
This section is empty.
Types ¶
type Message ¶
type Message interface {
Type() string
Serialize() ([]byte, error)
Deserialize(payload []byte, input any) error
NewWithData(data any) Message
}
Message is a payload that knows its own encoding.
Type names the encoding and is written to the DEVKIT_CONTENT_TYPE header on dispatch, which is how NewMessageType picks the matching implementation on the way back. Deserialize decodes a payload into input, which must be a pointer. NewWithData returns a message of the same kind carrying data.
func NewJSONMessage ¶
NewJSONMessage returns a Message that encodes body as JSON.
Example ¶
package main
import (
"fmt"
"github.com/diegodesousas/go-devkit/pkg/stream"
)
type orderPlaced struct {
ID string `json:"id"`
Total int `json:"total"`
}
func main() {
msg := stream.NewJSONMessage(orderPlaced{ID: "42", Total: 250})
payload, err := msg.Serialize()
if err != nil {
panic(err)
}
fmt.Println(msg.Type())
fmt.Println(string(payload))
var decoded orderPlaced
if err := msg.Deserialize(payload, &decoded); err != nil {
panic(err)
}
fmt.Println(decoded.ID, decoded.Total)
}
Output: json {"id":"42","total":250} 42 250
func NewMessageType ¶
NewMessageType selects the Message implementation matching the DEVKIT_CONTENT_TYPE header of a Kafka message.
It returns ErrUnknownMessageType when the header is absent or names an encoding this package does not implement - which is what makes the consumer route the message to the dead letter topic instead of guessing.
func NewTextMessage ¶
NewTextMessage returns a Message that carries body as raw bytes.
Its Deserialize only accepts a *string destination and returns ErrStringDeserialization for anything else.
Example ¶
A text message carries bytes unchanged, and only decodes into a *string.
package main
import (
"fmt"
"github.com/diegodesousas/go-devkit/pkg/stream"
"github.com/pkg/errors"
)
func main() {
msg := stream.NewTextMessage("plain payload")
payload, err := msg.Serialize()
if err != nil {
panic(err)
}
fmt.Println(msg.Type(), string(payload))
var s string
if err := msg.Deserialize(payload, &s); err != nil {
panic(err)
}
fmt.Println(s)
var wrongType int
fmt.Println(errors.Is(msg.Deserialize(payload, &wrongType), stream.ErrStringDeserialization))
}
Output: text plain payload plain payload true
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package consumer runs a typed Kafka consumer loop around a user-supplied handler.
|
Package consumer runs a typed Kafka consumer loop around a user-supplied handler. |
|
Package dispatcher publishes messages to Kafka and waits for the broker to confirm each one.
|
Package dispatcher publishes messages to Kafka and waits for the broker to confirm each one. |