Kafka

package main

import (
    "context"
    "fmt"

    "github.com/IBM/sarama"
    "github.com/xraph/forge"
    "github.com/xraph/forge/extensions/kafka"
)

func main() {
    app := forge.NewApp(forge.AppConfig{Name: "my-app", Version: "1.0.0"})

    app.RegisterExtension(kafka.NewExtension(
        kafka.WithBrokers("localhost:9092", "localhost:9093"),
        kafka.WithClientID("my-service"),
        kafka.WithProducer(true),
        kafka.WithConsumer(true),
        kafka.WithConsumerGroup("my-group"),
    ))

    ctx := context.Background()
    app.Start(ctx)
    defer app.Stop(ctx)

    // Resolve the Kafka client from DI
    svc, _ := forge.InjectType[*kafka.KafkaService](app.Container())
    client := svc // KafkaService satisfies the Kafka interface

    // Create a topic
    client.CreateTopic("events", kafka.TopicConfig{
        NumPartitions:     6,
        ReplicationFactor: 3,
    })

    // Send a message synchronously
    client.SendMessage("events", []byte("user-123"), []byte(`{"action":"login"}`))

    // Send a message asynchronously (fire-and-forget)
    client.SendMessageAsync("events", []byte("user-456"), []byte(`{"action":"signup"}`))

    // Consume messages from a topic
    client.Consume(ctx, []string{"events"}, func(msg *sarama.ConsumerMessage) error {
        fmt.Printf("Received: key=%s value=%s partition=%d offset=%d\n",
            string(msg.Key), string(msg.Value), msg.Partition, msg.Offset)
        return nil
    })
}

client.JoinConsumerGroup(ctx, "analytics-group", []string{"events", "metrics"},
    &myGroupHandler{})

// Handler implements kafka.ConsumerGroupHandler
type myGroupHandler struct{}

func (h *myGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
    return nil
}

func (h *myGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
    return nil
}

func (h *myGroupHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for msg := range claim.Messages() {
        fmt.Printf("Group message: %s\n", string(msg.Value))
        session.MarkMessage(msg, "")
    }
    return nil
}

type EventPublisher struct {
    kafka  kafka.Kafka
    logger forge.Logger
}

func NewEventPublisher(k *kafka.KafkaService, logger forge.Logger) *EventPublisher {
    return &EventPublisher{kafka: k, logger: logger}
}

func (p *EventPublisher) PublishUserEvent(action string, userID string, data []byte) error {
    return p.kafka.SendMessage("user-events", []byte(userID), data)
}

forge.ProvideConstructor(app.Container(), NewEventPublisher)