Resolve "Implement gNMI subscription management"

Description

Related Issue

Motivation and Context

How Has This Been Tested?

Screenshots (if appropriate):

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

Checklist:

• My code follows the code style of this project.
• My change requires a change to the documentation.
• I have updated the documentation accordingly.
• I have read the CONTRIBUTING document.
• I have added tests to cover my changes.
• All new and existing tests passed.

Related to #382

controller/nucleus/networkElementWatcher.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"sync"
 
 	"code.fbi.h-da.de/danet/gosdn/controller/config"
 	"code.fbi.h-da.de/danet/gosdn/controller/customerrs"
@@ -33,9 +34,10 @@ const (
 // NetworkElementWatcher is a component that subscribes to network elements via gNMI from within the controller and handles
 // responses by triggering the internal event process.
 type NetworkElementWatcher struct {
-	mneService                 networkelement.Service
-	networkelementSubcriptions map[uuid.UUID]*networkelementSubscriptionHelper // TODO: Mutex stuff here!
-	eventService               eventInterfaces.Service
+	mneService                       networkelement.Service
+	networkelementSubscriptionsMutex sync.Mutex
+	networkelementSubcriptions       map[uuid.UUID]*networkelementSubscriptionHelper
+	eventService                     eventInterfaces.Service
 }
 
 // networkelementSubscriptionHelper is used to store information to stop a running subscribe go routine.
@@ -150,6 +152,9 @@ func (n *NetworkElementWatcher) callSubscribe(stopContext context.Context, mne n
 }
 
 func (n *NetworkElementWatcher) addToNetworkElementSubscriptions(subID uuid.UUID, devSub *networkelementSubscriptionHelper) {
+	n.networkelementSubscriptionsMutex.Lock()
+	defer n.networkelementSubscriptionsMutex.Unlock()
+
 	n.networkelementSubcriptions[subID] = devSub
 }
 
@@ -163,6 +168,9 @@ func (n *NetworkElementWatcher) StopAndRemoveAllNetworkElementSubscriptions() {
 // StopAndRemoveNetworkElementSubscription passes a subscription uuid to stop the running subscription go routing and removes the entry from the map
 // of network element subscriptions.
 func (n *NetworkElementWatcher) StopAndRemoveNetworkElementSubscription(subID uuid.UUID) {
+	n.networkelementSubscriptionsMutex.Lock()
+	defer n.networkelementSubscriptionsMutex.Unlock()
+
 	n.networkelementSubcriptions[subID].stopFunc()
 	delete(n.networkelementSubcriptions, subID)
 }
@@ -253,6 +261,8 @@ func (n *NetworkElementWatcher) mergeGnmiSubscriptions(gNMISusbcriptionPathsFrom
 // GetAllSubscriptionInformations returns the information of all running sunscriptions.
 func (n *NetworkElementWatcher) GetAllSubscriptionInformations() []*networkelementSubscriptionHelper {
 	var information []*networkelementSubscriptionHelper
+	n.networkelementSubscriptionsMutex.Lock()
+	defer n.networkelementSubscriptionsMutex.Unlock()
 
 	for _, info := range n.networkelementSubcriptions {
 		information = append(information, info)
@@ -263,6 +273,9 @@ func (n *NetworkElementWatcher) GetAllSubscriptionInformations() []*networkeleme
 
 // GetSubscriptionInformations returns the information for one specific subscription referenced by its ID.
 func (n *NetworkElementWatcher) GetSubscriptionInformations(subID uuid.UUID) (*networkelementSubscriptionHelper, error) {
+	n.networkelementSubscriptionsMutex.Lock()
+	defer n.networkelementSubscriptionsMutex.Unlock()
+
 	information, ok := n.networkelementSubcriptions[subID]
 	if !ok {
 		return nil, errors.New(fmt.Sprintf("Couldn't retrieve information for subscription with ID: %s", subID.String()))