IBM/sarama Go client library for Apache Kafka support

.gitignore

@@ -3,5 +3,8 @@ pki/*
 !pki/create_pki.sh
 redis-data/
 .idea/
+kraft[0-9]*-data/
 *.orig
 **/.*.orig
+.env
+

Makefile

@@ -75,6 +75,8 @@ build-ui:
 
 docker-build: docker-build-verifier docker-build-registry docker-build-persistent docker-build-mockas docker-build-apigw docker-build-issuer docker-build-ui
 
+docker-build-goland-debug: docker-build-verifier docker-build-registry docker-build-persistent docker-build-mockas docker-build-apigw docker-build-issuer docker-build-ui-goland-debug
+
 docker-build-gobuild:
 	$(info Docker Building gobuild with tag: $(VERSION))
 	docker build --tag $(DOCKER_TAG_GOBUILD) --file dockerfiles/gobuild .
@@ -107,6 +109,10 @@ docker-build-ui:
 	$(info Docker building ui with tag: $(VERSION))
 	docker build --build-arg SERVICE_NAME=ui --tag $(DOCKER_TAG_UI) --file dockerfiles/ui_worker .
 
+docker-build-ui-goland-debug:
+	$(info Docker building ui with tag: $(VERSION))
+	docker build --build-arg SERVICE_NAME=ui --tag $(DOCKER_TAG_UI) --file dockerfiles/ui_worker_goland_debug .
+
 docker-push-gobuild:
 	$(info Pushing docker images)
 	docker push $(DOCKER_TAG_GOBUILD)

cmd/apigw/main.go

@@ -9,6 +9,8 @@ import (
 	"vc/internal/apigw/apiv1"
 	"vc/internal/apigw/db"
 	"vc/internal/apigw/httpserver"
+	"vc/internal/apigw/inbound"
+	"vc/internal/apigw/outbound"
 	"vc/internal/apigw/simplequeue"
 	"vc/pkg/configuration"
 	"vc/pkg/kvclient"
@@ -59,16 +61,36 @@ func main() {
 		panic(err)
 	}
 
+	var eventPublisher apiv1.EventPublisher
+	if cfg.Common.Kafka.Enabled {
+		var err error
+		eventPublisher, err = outbound.New(ctx, cfg, tracer, log)
+		if err != nil {
+			panic(err)
+		}
+		services["eventPublisher"] = eventPublisher
+	} else {
+		log.Info("EventPublisher disabled in config")
+	}
+
 	apiv1Client, err := apiv1.New(ctx, kvClient, dbService, simpleQueueService, tracer, cfg, log.New("apiv1"))
 	if err != nil {
 		panic(err)
 	}
-	httpService, err := httpserver.New(ctx, cfg, apiv1Client, tracer, log.New("httpserver"))
+	httpService, err := httpserver.New(ctx, cfg, apiv1Client, tracer, log.New("httpserver"), eventPublisher)
 	services["httpService"] = httpService
 	if err != nil {
 		panic(err)
 	}
 
+	eventConsumer, err := inbound.New(ctx, cfg, log, apiv1Client, tracer)
+	if err != nil {
+		panic(err)
+	}
+	if eventConsumer != nil {
+		services["eventConsumer"] = eventConsumer
+	}
+
 	// Handle sigterm and await termChan signal
 	termChan := make(chan os.Signal, 1)
 	signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)

cmd/mockas/main.go

@@ -8,6 +8,7 @@ import (
 	"syscall"
 	"vc/internal/mockas/apiv1"
 	"vc/internal/mockas/httpserver"
+	"vc/internal/mockas/inbound"
 	"vc/pkg/configuration"
 	"vc/pkg/logger"
 	"vc/pkg/trace"
@@ -32,6 +33,8 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
+	mainLog := log.New("main")
+
 	tracer, err := trace.New(ctx, cfg, log, "vc", "mock_as")
 	if err != nil {
 		panic(err)
@@ -41,19 +44,27 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
+
 	httpService, err := httpserver.New(ctx, cfg, apiv1Client, tracer, log.New("httpserver"))
 	services["httpService"] = httpService
 	if err != nil {
 		panic(err)
 	}
 
+	eventConsumer, err := inbound.New(ctx, cfg, log, apiv1Client, tracer)
+	if err != nil {
+		panic(err)
+	}
+	if eventConsumer != nil {
+		services["eventConsumer"] = eventConsumer
+	}
+
 	// Handle sigterm and await termChan signal
 	termChan := make(chan os.Signal, 1)
 	signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)
 
 	<-termChan // Blocks here until interrupted
 
-	mainLog := log.New("main")
 	mainLog.Info("HALTING SIGNAL!")
 
 	for serviceName, service := range services {

cmd/ui/main.go

@@ -10,6 +10,7 @@ import (
 	"time"
 	"vc/internal/ui/apiv1"
 	"vc/internal/ui/httpserver"
+	"vc/internal/ui/outbound"
 	"vc/pkg/configuration"
 	"vc/pkg/logger"
 	"vc/pkg/trace"
@@ -45,7 +46,20 @@ func main() {
 		panic(err)
 	}
 
-	apiClient, err := apiv1.New(ctx, cfg, tracer, log.New("ui_api_client"))
+	var eventPublisher apiv1.EventPublisher
+	if cfg.Common.Kafka.Enabled {
+		var err error
+		eventPublisher, err = outbound.New(ctx, cfg, tracer, log)
+		if err != nil {
+			panic(err)
+		}
+		services["eventPublisher"] = eventPublisher
+	} else {
+		log.Info("EventPublisher disabled in config")
+	}
+
+	apiClient, err := apiv1.New(ctx, cfg, tracer, eventPublisher, log.New("api_client"))
+	services["apiClient"] = apiClient
 	if err != nil {
 		panic(err)
 	}

config.yaml

@@ -25,6 +25,11 @@ common:
     base_url: "https://deutsche-rentenversicherung.de"
     recovery_level: 2
     size: 256
+  kafka:
+    enabled: true
+    brokers:
+      - "kafka0:9092"
+      - "kafka1:9092"
 
 authentic_sources:
   SUNET: