feat(sda-validator-orchestrator): update broker.ConnectionWatcher -> broker.Monitor and also listen to channel close events

Author: KarlG-nbis

sda-validator/orchestrator/api/api_test.go

@@ -192,9 +192,9 @@ func (m *mockBroker) Close() error {
 	panic("broker.close call not expected in unit tests")
 }
 
-func (m *mockBroker) ConnectionWatcher() chan *amqp.Error {
+func (m *mockBroker) Monitor() chan *amqp.Error {
 	// Function not needed for unit test, but to implement interface
-	panic("broker.ConnectionWatcher call not expected in unit tests")
+	panic("broker.Monitor call not expected in unit tests")
 }
 
 func mockAuthenticator(c *gin.Context) {

sda-validator/orchestrator/internal/broker/broker.go

@@ -207,7 +207,18 @@ func (broker *amqpBroker) Close() error {
 	return broker.connection.Close()
 }
 
-// ConnectionWatcher listens to events from the server
-func (broker *amqpBroker) ConnectionWatcher() chan *amqp.Error {
-	return broker.connection.NotifyClose(make(chan *amqp.Error))
+// Monitor listens to close events on the broker connection and channel
+func (broker *amqpBroker) Monitor() chan *amqp.Error {
+	errChan := make(chan *amqp.Error)
+
+	go func() {
+		select {
+		case err := <-broker.connection.NotifyClose(make(chan *amqp.Error, 1)):
+			errChan <- err
+		case err := <-broker.channel.NotifyClose(make(chan *amqp.Error, 1)):
+			errChan <- err
+		}
+	}()
+
+	return errChan
 }

sda-validator/orchestrator/internal/broker/interface_broker.go

@@ -13,6 +13,6 @@ type AMQPBrokerI interface {
 	Subscribe(ctx context.Context, queue, consumerID string, handleFunc func(context.Context, amqp.Delivery) error) error
 	// Close the broker connection
 	Close() error
-	// ConnectionWatcher returns a chan watching the broker connection events
-	ConnectionWatcher() chan *amqp.Error
+	// Monitor returns a chan watching the broker connection and channel close events
+	Monitor() chan *amqp.Error
 }

sda-validator/orchestrator/jobpreparationworker/worker.go

@@ -118,7 +118,7 @@ func (w *worker) close() {
 	close(w.stopCh)
 }
 
-// ShutdownWorkers shutdowns and waits for all workers to have closed
+// Shutdown shutdowns and waits for all workers to have closed
 func (w *Workers) Shutdown() {
 	wg := sync.WaitGroup{}
 	for _, w := range w.workers {

sda-validator/orchestrator/jobpreparationworker/worker_test.go

@@ -206,9 +206,9 @@ func (m *mockBroker) Close() error {
 	panic("broker.close call not expected in unit tests")
 }
 
-func (m *mockBroker) ConnectionWatcher() chan *amqp.Error {
+func (m *mockBroker) Monitor() chan *amqp.Error {
 	// Function not needed for unit test, but to implement interface
-	panic("broker.ConnectionWatcher call not expected in unit tests")
+	panic("broker.Monitor call not expected in unit tests")
 }
 
 func (ts *JobPreparationWorkerTestSuite) TestInitWorkers() {

sda-validator/orchestrator/jobworker/worker_test.go

@@ -167,9 +167,9 @@ func (m *mockBroker) Close() error {
 	panic("broker.close call not expected in unit tests")
 }
 
-func (m *mockBroker) ConnectionWatcher() chan *amqp.Error {
+func (m *mockBroker) Monitor() chan *amqp.Error {
 	// Function not needed for unit test, but to implement interface
-	panic("broker.ConnectionWatcher call not expected in unit tests")
+	panic("broker.Monitor call not expected in unit tests")
 }
 
 func (ts *JobWorkerTestSuite) TestInitWorkers() {

sda-validator/orchestrator/main.go

@@ -162,7 +162,7 @@ func main() {
 	}()
 
 	go func() {
-		if err := <-amqpBroker.ConnectionWatcher(); err != nil {
+		if err := <-amqpBroker.Monitor(); err != nil {
 			log.Errorf("broker connection error: %v", err)
 			sigc <- syscall.SIGTERM
 		}