Documentation
¶
Overview ¶
Package dispatcher publishes messages to Kafka and waits for the broker to confirm each one.
A dispatcher wraps a producer client:
client, err := dispatcher.NewClient(
dispatcher.WithBootstrapServers("localhost:9092"),
)
if err != nil {
return err
}
d := dispatcher.New(client)
defer d.Shutdown()
err = d.Dispatch(ctx, "orders", order.ID, stream.NewJSONMessage(order))
Dispatch is synchronous: it blocks on the delivery report and returns only once the broker has acknowledged the message or reported a failure. A delivery timeout surfaces as stream.ErrProcessMessageTimedOut, so callers can distinguish "the broker is slow" from a malformed message.
The producer is configured for ordering and durability over throughput - acks=all with a single in-flight request - which combined with the synchronous wait caps the rate at roughly one message per round trip. It is the right default for event streams that must not reorder, and the wrong one for bulk loading.
Dispatch opens a Datadog span named "stream.dispatcher" tagged with the topic and key. The span does not currently propagate to the consumer.
Shutdown flushes whatever the producer still holds and closes it. Call it before the process exits, or buffered messages are lost.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client interface {
Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error
Close()
Flush(timeoutMs int) int
}
Client is the Kafka producer driver a Dispatcher writes through.
type Dispatcher ¶
type Dispatcher interface {
Dispatch(ctx context.Context, topic string, key string, content stream.Message) error
Shutdown()
}
Dispatcher publishes messages to Kafka.
Dispatch blocks until the broker confirms the message. Shutdown flushes and closes the underlying producer and must be called before the process exits.
func New ¶
func New(client Client) Dispatcher
New returns a Dispatcher publishing through client.
Example ¶
Requires a Kafka broker, so this example is compiled but not run.
package main
import (
"context"
"github.com/diegodesousas/go-devkit/pkg/stream"
"github.com/diegodesousas/go-devkit/pkg/stream/dispatcher"
)
type orderPlaced struct {
ID string `json:"id"`
Total int `json:"total"`
}
func main() {
client, err := dispatcher.NewClient(
dispatcher.WithBootstrapServers("localhost:9092"),
)
if err != nil {
panic(err)
}
d := dispatcher.New(client)
defer d.Shutdown() // flushes buffered messages - skipping it loses them
order := orderPlaced{ID: "42", Total: 250}
err = d.Dispatch(context.Background(), "orders", order.ID, stream.NewJSONMessage(order))
if err != nil {
panic(err)
}
}
Output:
type Option ¶
type Option func(settings dispatcherSettings) dispatcherSettings
Option configures the producer built by NewClient.
func WithBootstrapServers ¶
WithBootstrapServers sets the broker list, as a comma-separated "host:port" string.
func WithDispatchTimeoutMs ¶
WithDispatchTimeoutMs sets message.timeout.ms, the deadline for a produced message to be acknowledged. Defaults to 1000, far below the librdkafka default of 300000, so a brief broker hiccup surfaces as stream.ErrProcessMessageTimedOut rather than being ridden out.
func WithLogLevel ¶
WithLogLevel sets the librdkafka syslog level. Defaults to 6 (informational).
Source Files