VOL-3734 optimise rpc events to be send to queue and then to kafka from the queue
Change-Id: I5e068722412b6d9526760900d9173aaf51e00946
diff --git a/rw_core/core/kafka.go b/rw_core/core/kafka.go
index ae33090..5793eb5 100644
--- a/rw_core/core/kafka.go
+++ b/rw_core/core/kafka.go
@@ -87,6 +87,7 @@
return nil, ctx.Err()
}
}
+ go ep.Start()
if updateProbeService {
probe.UpdateStatusFromContext(ctx, clusterMessageBus, probe.ServiceStatusRunning)
}
@@ -96,6 +97,11 @@
return ep, nil
}
+func stopEventProxy(ctx context.Context, kafkaClient kafka.Client, ep *events.EventProxy) {
+ defer kafkaClient.Stop(ctx)
+ ep.Stop()
+}
+
// Interface that is valid for both EventProxy and InterContainerProxy
type KafkaProxy interface {
EnableLivenessChannel(ctx context.Context, enable bool) chan bool