consumer

.gitignore

@@ -0,0 +1 @@
+.idea

config/config.go

@@ -0,0 +1,8 @@
+package config
+
+type Configuration struct {
+}
+
+func (c Configuration) Invalidate() error {
+	return Invalidate(c)
+}

config/iconfig.go

@@ -0,0 +1,93 @@
+package config
+
+import (
+	"fmt"
+	"github.com/urfave/cli/v2"
+	"gopkg.in/yaml.v3"
+	"os"
+	"reflect"
+)
+
+const cliMetaKey = "cli.config"
+
+type IConfig interface {
+	Invalidate() error
+}
+
+func Invalidate(config IConfig) error {
+	if config == nil {
+		return fmt.Errorf("config must not be a nil")
+	}
+
+	val := reflect.ValueOf(config)
+	if val.Kind() != reflect.Struct {
+		if val.IsZero() {
+			return fmt.Errorf("config is empty")
+		}
+
+		val = reflect.ValueOf(config).Elem()
+	}
+
+	for i := 0; i < val.NumField(); i++ {
+		if !val.Field(i).CanInterface() || val.Field(i).IsZero() {
+			continue
+		}
+
+		if elm, ok := val.Field(i).Interface().(IConfig); ok {
+			if er := elm.Invalidate(); er != nil {
+				return er
+			}
+		}
+	}
+
+	return nil
+}
+
+func Read[T IConfig](path string) (*T, error) {
+	var (
+		cfg = new(T)
+		fd  []byte
+		err error
+	)
+
+	if fd, err = os.ReadFile(path); err != nil {
+		return nil, err
+	}
+
+	if err = yaml.Unmarshal(fd, cfg); err != nil {
+		return nil, err
+	}
+
+	if err = (*cfg).Invalidate(); err != nil {
+		return nil, err
+	}
+
+	if err = Invalidate(*cfg); err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}
+
+func Load[T IConfig](ctx *cli.Context, path cli.Path) error {
+	cfg, err := Read[T](path)
+	if err != nil {
+		return fmt.Errorf("could not load config from: %s; %v", path, err)
+	}
+
+	if cfg == nil {
+		return fmt.Errorf("could not load config from: %s", path)
+	}
+
+	ctx.App.Metadata[cliMetaKey] = cfg
+	return nil
+}
+
+func Get[T IConfig](ctx *cli.Context) *T {
+	cfg, ok := ctx.App.Metadata[cliMetaKey].(*T)
+	if !ok {
+		return nil
+	}
+
+	return cfg
+}

config/iconfig_test.go

@@ -0,0 +1,43 @@
+package config
+
+import "testing"
+
+type _config struct {
+	Debug bool
+}
+
+func (_ _config) Invalidate() error {
+	return nil
+}
+
+func TestInvalidate(t *testing.T) {
+	tests := []struct {
+		config  IConfig
+		wantErr bool
+	}{
+		{
+			config:  nil,
+			wantErr: true,
+		},
+		{
+			config: _config{
+				Debug: false,
+			},
+			wantErr: false,
+		},
+		{
+			config: &_config{
+				Debug: false,
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run("config.Invalidate", func(t *testing.T) {
+			if err := Invalidate(tt.config); (err != nil) != tt.wantErr {
+				t.Errorf("Invalidate() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}

consumer/config.go

@@ -0,0 +1,25 @@
+package consumer
+
+import "fmt"
+
+type Config struct {
+	Hosts   []string `yaml:"hosts"`
+	Group   string   `yaml:"group"`
+	Timeout int      `yaml:"timeout"`
+}
+
+func (c Config) Invalidate() error {
+	if len(c.Hosts) < 1 {
+		return fmt.Errorf("at least one bootstrap server / host must be provided")
+	}
+
+	if c.Group == "" {
+		return fmt.Errorf("group name must not be an empty string")
+	}
+
+	if c.Timeout < 1 {
+		c.Timeout = 100
+	}
+
+	return nil
+}

consumer/errors.go

@@ -0,0 +1,7 @@
+package consumer
+
+import "errors"
+
+var (
+	ErrNoMessage = errors.New("no message")
+)

consumer/kafka_consumer.go

@@ -0,0 +1,273 @@
+package consumer
+
+import (
+	"errors"
+	"fmt"
+	"git.beejay.kim/tool/service"
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"github.com/google/uuid"
+	"github.com/kelindar/bitmap"
+	"github.com/rs/zerolog/log"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	flagSubscribed uint32 = iota
+)
+
+//goland:noinspection ALL
+type _consumer struct {
+	config   *Config
+	state    bitmap.Bitmap
+	handlers []service.ConsumerHandler
+	session  *kafka.Consumer
+}
+
+func NewKafkaConsumer(cfg *Config) (service.Consumer, error) {
+	var (
+		c = &_consumer{
+			config: cfg,
+		}
+		err error
+	)
+
+	if cfg == nil {
+		return nil, fmt.Errorf("config must be provided")
+	}
+
+	opts := &kafka.ConfigMap{
+		"broker.address.family":         "v4",
+		"bootstrap.servers":             strings.Join(c.config.Hosts, ","),
+		"group.id":                      c.config.Group,
+		"partition.assignment.strategy": "cooperative-sticky",
+		"auto.offset.reset":             "earliest",
+		"log_level":                     0,
+		"enable.auto.commit":            false,
+	}
+
+	if c.session, err = kafka.NewConsumer(opts); err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+func (c *_consumer) ID() uuid.UUID {
+	return uuid.NewSHA1(uuid.NameSpaceDNS, []byte("consumer.kafka"))
+}
+
+func (c *_consumer) Subscribed() bool {
+	return c.state.Contains(flagSubscribed)
+}
+
+func (c *_consumer) RegisterHandlers(handlers ...service.ConsumerHandler) {
+	c.handlers = append(c.handlers, handlers...)
+}
+
+func (c *_consumer) Subscribe(topics []string, ch chan *kafka.Message) error {
+	if c.Subscribed() {
+		return fmt.Errorf("illegal state: already subscribed")
+	}
+
+	if err := c.session.SubscribeTopics(topics, rebalanceCallback); err != nil {
+		return err
+	}
+
+	c.state.Set(flagSubscribed)
+	for c.Subscribed() {
+		message, err := c.poll()
+		if err != nil {
+			// silently wait for a next message
+			if errors.Is(err, ErrNoMessage) {
+				continue
+			}
+
+			log.Debug().
+				Str("service", "consumer").
+				Err(err).
+				Send()
+
+			c.state.Remove(flagSubscribed)
+			return err
+		}
+
+		if message != nil {
+			ch <- message
+		}
+
+	}
+
+	log.Debug().Msg("consumer closed gracefully")
+	return nil
+}
+
+func (c *_consumer) Close() error {
+	if c.Subscribed() {
+		c.state.Remove(flagSubscribed)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+
+		if c.session != nil {
+			time.Sleep(time.Second)
+			_ = c.session.Close() //nolint:errcheck
+		}
+	}()
+
+	wg.Wait()
+	return nil
+}
+
+func (c *_consumer) poll() (*kafka.Message, error) {
+	var (
+		ev  = c.session.Poll(c.config.Timeout)
+		err error
+	)
+
+	switch e := ev.(type) {
+	case *kafka.Message:
+		for i := range c.handlers {
+			if err = c.handlers[i](e); err != nil {
+				return nil, err
+			}
+		}
+
+		// Handle manual commit since enable.auto.commit is unset.
+		if err = maybeCommit(c.session, e.TopicPartition); err != nil {
+			return nil, err
+		}
+
+		return e, nil
+	case kafka.Error:
+		log.Debug().
+			Str("service", "consumer").
+			Err(e).
+			Send()
+		return nil, e
+	default:
+		if e != nil {
+			log.Debug().
+				Str("service", "consumer").
+				Any("event", e).
+				Send()
+		}
+
+		return nil, ErrNoMessage
+	}
+}
+
+// maybeCommit is called for each message we receive from a Kafka topic.
+// This method can be used to apply some arbitary logic/processing to the
+// offsets, write the offsets into some external storage, and finally, to
+// decide when we want to commit already-stored offsets into Kafka.
+func maybeCommit(c *kafka.Consumer, topicPartition kafka.TopicPartition) error {
+	// Commit the already-stored offsets to Kafka whenever the offset is divisible
+	// by 10, otherwise return early.
+	// This logic is completely arbitrary. We can use any other internal or
+	// external variables to decide when we commit the already-stored offsets.
+	if topicPartition.Offset%10 != 0 {
+		return nil
+	}
+
+	commitedOffsets, err := c.Commit()
+
+	// ErrNoOffset occurs when there are no stored offsets to commit. This
+	// can happen if we haven't stored anything since the last commit.
+	// While this will never happen for this example since we call this method
+	// per-message, and thus, always have something to commit, the error
+	// handling is illustrative of how to handle it in cases we call Commit()
+	// in another way, for example, every N seconds.
+	if err != nil && err.(kafka.Error).Code() != kafka.ErrNoOffset {
+		return err
+	}
+
+	log.Debug().
+		Str("service", "consumer").
+		Msgf("commited offsets to Kafka: %v", commitedOffsets)
+	return nil
+}
+
+// rebalanceCallback is called on each group rebalance to assign additional
+// partitions, or remove existing partitions, from the consumer's current
+// assignment.
+//
+// A rebalance occurs when a consumer joins or leaves a consumer group, if it
+// changes the topic(s) it's subscribed to, or if there's a change in one of
+// the topics it's subscribed to, for example, the total number of partitions
+// increases.
+//
+// The application may use this optional callback to inspect the assignment,
+// alter the initial start offset (the .Offset field of each assigned partition),
+// and read/write offsets to commit to an alternative store outside of Kafka.
+func rebalanceCallback(c *kafka.Consumer, event kafka.Event) error {
+	switch ev := event.(type) {
+	case kafka.AssignedPartitions:
+		log.Debug().
+			Str("service", "consumer").
+			Msgf("%s rebalance: %d new partition(s) assigned: %v",
+				c.GetRebalanceProtocol(),
+				len(ev.Partitions),
+				ev.Partitions)
+
+		err := c.Assign(ev.Partitions)
+		if err != nil {
+			return err
+		}
+
+	case kafka.RevokedPartitions:
+		log.Debug().
+			Str("service", "consumer").
+			Msgf("%s rebalance: %d partition(s) revoked: %v",
+				c.GetRebalanceProtocol(),
+				len(ev.Partitions),
+				ev.Partitions)
+
+		// Usually, the rebalance callback for `RevokedPartitions` is called
+		// just before the partitions are revoked. We can be certain that a
+		// partition being revoked is not yet owned by any other consumer.
+		// This way, logic like storing any pending offsets or committing
+		// offsets can be handled.
+		// However, there can be cases where the assignment is lost
+		// involuntarily. In this case, the partition might already be owned
+		// by another consumer, and operations including committing
+		// offsets may not work.
+		if c.AssignmentLost() {
+			// Our consumer has been kicked out of the group and the
+			// entire assignment is thus lost.
+			log.Debug().
+				Str("service", "consumer").
+				Msg("Assignment lost involuntarily, commit may fail")
+		}
+
+		// Since enable.auto.commit is unset, we need to commit offsets manually
+		// before the partition is revoked.
+		commitedOffsets, err := c.Commit()
+		if err != nil && err.(kafka.Error).Code() != kafka.ErrNoOffset {
+			log.Debug().
+				Str("service", "consumer").
+				Err(err).
+				Msg("failed to commit offsets")
+			return err
+		}
+
+		log.Debug().
+			Str("service", "consumer").
+			Msgf("commited offsets to Kafka: %v", commitedOffsets)
+
+		// Similar to Assign, client automatically calls Unassign() unless the
+		// callback has already called that method. Here, we don't call it.
+
+	default:
+		log.Debug().
+			Str("service", "consumer").
+			Msgf("unxpected event type: %v", event)
+	}
+
+	return nil
+}

go.mod

@@ -0,0 +1,23 @@
+module git.beejay.kim/tool/service
+
+go 1.21
+
+require (
+	github.com/confluentinc/confluent-kafka-go/v2 v2.3.0
+	github.com/google/uuid v1.5.0
+	github.com/kelindar/bitmap v1.5.2
+	github.com/rs/zerolog v1.31.0
+	github.com/urfave/cli/v2 v2.26.0
+	gopkg.in/yaml.v3 v3.0.1
+)
+
+require (
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
+	github.com/kelindar/simd v1.1.2 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
+	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
+	golang.org/x/sys v0.12.0 // indirect
+)

go.sum

@@ -0,0 +1,101 @@
+github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8=
+github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
+github.com/Microsoft/go-winio v0.5.2 h1:a9IhgEQBCUEk6QCdml9CiJGhAws+YwffDHEMp1VMrpA=
+github.com/Microsoft/go-winio v0.5.2/go.mod h1:WpS1mjBmmwHBEWmogvA2mj8546UReBk4v8QkMxJ6pZY=
+github.com/Microsoft/hcsshim v0.9.4 h1:mnUj0ivWy6UzbB1uLFqKR6F+ZyiDc7j4iGgHTpO+5+I=
+github.com/Microsoft/hcsshim v0.9.4/go.mod h1:7pLA8lDk46WKDWlVsENo92gC0XFa8rbKfyFRBqxEbCc=
+github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8UtC4=
+github.com/cenkalti/backoff/v4 v4.1.3/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
+github.com/confluentinc/confluent-kafka-go/v2 v2.3.0 h1:icCHutJouWlQREayFwCc7lxDAhws08td+W3/gdqgZts=
+github.com/confluentinc/confluent-kafka-go/v2 v2.3.0/go.mod h1:/VTy8iEpe6mD9pkCH5BhijlUl8ulUXymKv1Qig5Rgb8=
+github.com/containerd/cgroups v1.0.4 h1:jN/mbWBEaz+T1pi5OFtnkQ+8qnmEbAr1Oo1FRm5B0dA=
+github.com/containerd/cgroups v1.0.4/go.mod h1:nLNQtsF7Sl2HxNebu77i1R0oDlhiTG+kO4JTrUzo6IA=
+github.com/containerd/containerd v1.6.8 h1:h4dOFDwzHmqFEP754PgfgTeVXFnLiRc6kiqC7tplDJs=
+github.com/containerd/containerd v1.6.8/go.mod h1:By6p5KqPK0/7/CgO/A6t/Gz+CUYUu2zf1hUaaymVXB0=
+github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/cpuguy83/go-md2man/v2 v2.0.2 h1:p1EgwI/C7NhT0JmVkwCD2ZBK8j4aeHQX2pMHHBfMQ6w=
+github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/docker/distribution v2.8.1+incompatible h1:Q50tZOPR6T/hjNsyc9g8/syEs6bk8XXApsHjKukMl68=
+github.com/docker/distribution v2.8.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
+github.com/docker/docker v20.10.17+incompatible h1:JYCuMrWaVNophQTOrMMoSwudOVEfcegoZZrleKc1xwE=
+github.com/docker/docker v20.10.17+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
+github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
+github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
+github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
+github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
+github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
+github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/kelindar/bitmap v1.5.2 h1:XwX7CTvJtetQZ64zrOkApoZZHBJRkjE23NfqUALA/HE=
+github.com/kelindar/bitmap v1.5.2/go.mod h1:j3qZjxH9s4OtvsnFTP2bmPkjqil9Y2xQlxPYHexasEA=
+github.com/kelindar/simd v1.1.2 h1:KduKb+M9cMY2HIH8S/cdJyD+5n5EGgq+Aeeleos55To=
+github.com/kelindar/simd v1.1.2/go.mod h1:inq4DFudC7W8L5fhxoeZflLRNpWSs0GNx6MlWFvuvr0=
+github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
+github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
+github.com/magiconair/properties v1.8.6 h1:5ibWZ6iY0NctNGWo87LalDlEZ6R41TqbbDamhfG/Qzo=
+github.com/magiconair/properties v1.8.6/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60=
+github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
+github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
+github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
+github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/moby/sys/mount v0.3.3 h1:fX1SVkXFJ47XWDoeFW4Sq7PdQJnV2QIDZAqjNqgEjUs=
+github.com/moby/sys/mount v0.3.3/go.mod h1:PBaEorSNTLG5t/+4EgukEQVlAvVEc6ZjTySwKdqp5K0=
+github.com/moby/sys/mountinfo v0.6.2 h1:BzJjoreD5BMFNmD9Rus6gdd1pLuecOFPt8wC+Vygl78=
+github.com/moby/sys/mountinfo v0.6.2/go.mod h1:IJb6JQeOklcdMU9F5xQ8ZALD+CUr5VlGpwtX+VE0rpI=
+github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 h1:dcztxKSvZ4Id8iPpHERQBbIJfabdt4wUm5qy3wOL2Zc=
+github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6/go.mod h1:E2VnQOmVuvZB6UYnnDB0qG5Nq/1tD9acaOpo6xmt0Kw=
+github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
+github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
+github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
+github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
+github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799 h1:rc3tiVYb5z54aKaDfakKn0dDjIyPpTtszkjuMzyt7ec=
+github.com/opencontainers/image-spec v1.0.3-0.20211202183452-c5a74bcca799/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0=
+github.com/opencontainers/runc v1.1.3 h1:vIXrkId+0/J2Ymu2m7VjGvbSlAId9XNRPhn2p4b+d8w=
+github.com/opencontainers/runc v1.1.3/go.mod h1:1J5XiS+vdZ3wCyZybsuxXZWGrgSr8fFJHLXuG2PsnNg=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
+github.com/rs/zerolog v1.31.0 h1:FcTR3NnLWW+NnTwwhFWiJSZr4ECLpqCm6QsEnyvbV4A=
+github.com/rs/zerolog v1.31.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
+github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
+github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/testcontainers/testcontainers-go v0.14.0 h1:h0D5GaYG9mhOWr2qHdEKDXpkce/VlvaYOCzTRi6UBi8=
+github.com/testcontainers/testcontainers-go v0.14.0/go.mod h1:hSRGJ1G8Q5Bw2gXgPulJOLlEBaYJHeBSOkQM5JLG+JQ=
+github.com/urfave/cli/v2 v2.26.0 h1:3f3AMg3HpThFNT4I++TKOejZO8yU55t3JnnSr4S4QEI=
+github.com/urfave/cli/v2 v2.26.0/go.mod h1:8qnjx1vcq5s2/wpsqoZFndg2CE5tNFyrTvS6SinrnYQ=
+github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 h1:bAn7/zixMGCfxrRTfdpNzjtPYqr8smhKouy9mxVdGPU=
+github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673/go.mod h1:N3UwUGtsrSj3ccvlPHLoLsHnpR27oXr4ZE984MbSER8=
+go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M=
+go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
+golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
+golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.12.0 h1:CM0HF96J0hcLAwsHPJZjfdNzs0gftsLfgKt57wWHJ0o=
+golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+google.golang.org/genproto v0.0.0-20230331144136-dcfb400f0633 h1:0BOZf6qNozI3pkN3fJLwNubheHJYHhMh91GRFOWWK08=
+google.golang.org/genproto v0.0.0-20230331144136-dcfb400f0633/go.mod h1:UUQDJDOlWu4KYeJZffbWgBkS1YFobzKbLVfK69pe0Ak=
+google.golang.org/grpc v1.54.0 h1:EhTqbhiYeixwWQtAEZAxmV9MGqcjEU2mFx52xCzNyag=
+google.golang.org/grpc v1.54.0/go.mod h1:PUSEXI6iWghWaB6lXM4knEgpJNu2qUcKfDtNci3EC2g=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

kafka.go

@@ -0,0 +1,20 @@
+package service
+
+import (
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+)
+
+type ConsumerHandler func(*kafka.Message) error
+
+type Consumer interface {
+	Service
+
+	Subscribe([]string, chan *kafka.Message) error
+	Subscribed() bool
+	RegisterHandlers(...ConsumerHandler)
+}
+
+type Producer interface {
+	Service
+	Produce(*kafka.Message) error
+}

service.go

@@ -0,0 +1,8 @@
+package service
+
+import "github.com/google/uuid"
+
+type Service interface {
+	ID() uuid.UUID
+	Close() error
+}