fix(evpn-bridge): first review

Signed-off-by: Dimitrios Markou <dimitrios.markou@ericsson.com>

pkg/infradb/subscriberframework/eventbus/eventbus.go

@@ -130,17 +130,15 @@ func (e *EventBus) subscriberExist(eventType string, moduleName string) bool {
 
 // Publish api notifies the subscribers with certain eventType
 func (e *EventBus) Publish(objectData *ObjectData, subscriber *Subscriber) error {
-	e.publishL.RLock()
-	defer e.publishL.RUnlock()
+	e.publishL.Lock()
+	defer e.publishL.Unlock()
 	var err error
-	// We need the default case here as if the subscriber is busy then we will not be able to sent to the subscriber channel
-	// and the Publish function will stuck. So the default case serves exctly this purpose.
+
 	select {
 	case subscriber.Ch <- objectData:
 		log.Printf("Publish(): Notification is sent to subscriber %s\n", subscriber.Name)
 	default:
-		log.Printf("Publish(): Channel for subsriber %s is busy. Notification not sent", subscriber.Name)
-		err = fmt.Errorf("channel is busy")
+		err = fmt.Errorf("channel for subscriber %s is busy", subscriber.Name)
 	}
 	return err
 }

pkg/infradb/taskmanager/taskmanager.go

@@ -131,7 +131,8 @@ func (t *TaskManager) processTasks() {
 				NotificationID: uuid.NewString(),
 			}
 			if err := eventbus.EBus.Publish(objectData, sub); err != nil {
-				log.Printf("processTasks(): Notification not sent to subscriber %+v with data %+v. Subscriber is busy. The Task %+v will be requeued.\n", sub, objectData, task)
+				log.Printf("processTasks(): Failed to sent notification: %+v\n", err)
+				log.Printf("processTasks(): Notification not sent to subscriber %+v with data %+v. The Task %+v will be requeued.\n", sub, objectData, task)
 				// We keep this subIndex in order to know from which subscriber to start iterating after the requeue of the Task
 				// so we do start again from the subscriber that returned an error or was unavailable for any reason.
 				task.subIndex += i