fix(event): Waiting for event bus to terminate for closing

Signed-off-by: Vincent Boutour <[email protected]>

cmd/fibr/fibr.go

@@ -187,5 +187,5 @@ func main() {
 	go appServer.Start("http", healthApp.End(), httputils.Handler(handler, healthApp, recoverer.Middleware, prometheusApp.Middleware, tracerApp.Middleware, owasp.New(owaspConfig).Middleware))
 
 	healthApp.WaitForTermination(appServer.Done())
-	server.GracefulWait(appServer.Done(), promServer.Done(), amqpExifApp.Done(), amqpShareApp.Done(), amqpWebhookApp.Done())
+	server.GracefulWait(appServer.Done(), promServer.Done(), amqpExifApp.Done(), amqpShareApp.Done(), amqpWebhookApp.Done(), eventBus.Done())
 }

pkg/provider/event.go

@@ -268,6 +268,7 @@ type EventBus struct {
 	tracer  trace.Tracer
 	counter *prometheus.CounterVec
 	bus     chan Event
+	closed  chan struct{}
 	done    chan struct{}
 }
 
@@ -287,6 +288,7 @@ func NewEventBus(size uint64, prometheusRegisterer prometheus.Registerer, tracer
 	}
 
 	return EventBus{
+		closed:  make(chan struct{}),
 		done:    make(chan struct{}),
 		bus:     make(chan Event, size),
 		counter: counter,
@@ -302,39 +304,45 @@ func (e EventBus) increaseMetric(event Event, state string) {
 	e.counter.WithLabelValues(event.Type.String(), state).Inc()
 }
 
+func (e EventBus) Done() <-chan struct{} {
+	return e.done
+}
+
 func (e EventBus) Push(event Event) error {
 	select {
-	case <-e.done:
+	case <-e.closed:
 		e.increaseMetric(event, "refused")
-		return errors.New("done signal is received")
+		return errors.New("bus is closed")
 	case e.bus <- event:
 		e.increaseMetric(event, "push")
 		return nil
 	}
 }
 
 func (e EventBus) Start(done <-chan struct{}, storageApp absto.Storage, renamers []Renamer, consumers ...EventConsumer) {
-	defer close(e.bus)
 	defer close(e.done)
 
 	go func() {
-		for event := range e.bus {
-			ctx, end := tracer.StartSpan(context.Background(), e.tracer, "event", trace.WithAttributes(attribute.String("type", event.Type.String())))
+		defer close(e.bus)
+		defer close(e.closed)
 
-			if event.Type == RenameEvent && event.Item.IsDir {
-				RenameDirectory(ctx, storageApp, renamers, event.Item, *event.New)
-			}
+		<-done
+	}()
 
-			for _, consumer := range consumers {
-				consumer(ctx, event)
-			}
+	for event := range e.bus {
+		ctx, end := tracer.StartSpan(context.Background(), e.tracer, "event", trace.WithAttributes(attribute.String("type", event.Type.String())))
 
-			end()
-			e.increaseMetric(event, "done")
+		if event.Type == RenameEvent && event.Item.IsDir {
+			RenameDirectory(ctx, storageApp, renamers, event.Item, *event.New)
+		}
+
+		for _, consumer := range consumers {
+			consumer(ctx, event)
 		}
-	}()
 
-	<-done
+		end()
+		e.increaseMetric(event, "done")
+	}
 }
 
 func RenameDirectory(ctx context.Context, storageApp absto.Storage, renamers []Renamer, old, new absto.Item) {

pkg/share/share.go

@@ -79,7 +79,6 @@ func New(config Config, storageApp absto.Storage, amqpClient *amqp.Client) (*App
 	}, nil
 }
 
-// Exclusive does action on shares with exclusive lock
 func (a *App) Exclusive(ctx context.Context, name string, duration time.Duration, action func(ctx context.Context) error) (bool, error) {
 	fn := func() error {
 		a.Lock()
@@ -111,7 +110,6 @@ exclusive:
 	return true, nil
 }
 
-// Get returns a share based on path
 func (a *App) Get(requestPath string) provider.Share {
 	cleanPath := strings.TrimPrefix(requestPath, "/")
 
@@ -127,7 +125,6 @@ func (a *App) Get(requestPath string) provider.Share {
 	return provider.Share{}
 }
 
-// Start worker
 func (a *App) Start(done <-chan struct{}) {
 	if err := a.loadShares(context.Background()); err != nil {
 		logger.Error("refresh shares: %s", err)

pkg/webhook/webhook.go

@@ -93,7 +93,6 @@ func New(config Config, storageApp absto.Storage, prometheusRegisterer prometheu
 	}, nil
 }
 
-// Exclusive does action on webhook with exclusive lock
 func (a *App) Exclusive(ctx context.Context, name string, duration time.Duration, action func(ctx context.Context) error) error {
 	fn := func() error {
 		a.Lock()
@@ -125,7 +124,6 @@ exclusive:
 	return nil
 }
 
-// Start worker
 func (a *App) Start(_ <-chan struct{}) {
 	a.Lock()
 	defer a.Unlock()