Add method which allows to remove a receiver

See merge request !142

goKMS/kms/akms/server/receiver.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/google/uuid"
+	"github.com/sirupsen/logrus"
 )
 
 type Receiver struct {
@@ -22,12 +23,27 @@ func (r *Receiver) RequestReceiverChannel(pathId uuid.UUID) (<-chan struct{}, er
 	return newSubChan, nil
 }
 
+func (r *Receiver) RemoveReceiver(pathId uuid.UUID) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if channel, ok := r.receivers[pathId]; ok {
+		close(channel)
+		delete(r.receivers, pathId)
+	} else {
+		return fmt.Errorf("There are no active subscribers for pathId: %s", pathId)
+	}
+
+	logrus.Debugf("Current active receivers after removing receiver for pathId: %s; Receivers: %v", pathId, r.receivers)
+	return nil
+}
+
 func (r *Receiver) InformReceiver(pathId uuid.UUID) error {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 
 	if receiver, ok := r.receivers[pathId]; ok {
-		receiver <- struct{}{}
+		go func() { receiver <- struct{}{} }()
 	} else {
 		return fmt.Errorf("There are no active subscribers for pathId: %s", pathId)
 	}

goKMS/kms/akms/server/server.go

@@ -87,6 +87,9 @@ func ksaReqHandler(eventBus *event.EventBus, receiver *Receiver, generateAndSend
 		select {
 		case <-receiverChan:
 		case <-time.After(20 * time.Second):
+			if err := receiver.RemoveReceiver(pathId); err != nil {
+				logrus.Errorf("Failed removing receiver for pathId: %s ; err: %v", pathId, err)
+			}
 			http.Error(w, fmt.Sprintf("Platform Key exchange failed for RequestID: %s", kmsKeyRequest.RequestID), http.StatusInternalServerError)
 			logrus.Errorf("Platform Key exchange failed for RequestID: %s", kmsKeyRequest.RequestID)
 			return