Add lock for clients map (#46)

(cherry picked from commit 9a08a09)

pkg/collector/process.go

@@ -46,6 +46,8 @@ type collectingProcess struct {
 	messages []*entities.Message
 	// maps each client to its client handler (required channels)
 	clients map[string]*clientHandler
+	// clientsLock allows multiple readers or one writer to access clients map at the same time
+	clientsLock sync.RWMutex
 }
 
 type clientHandler struct {
@@ -90,11 +92,21 @@ func (cp *collectingProcess) createClient() *clientHandler {
 	}
 }
 
+func (cp *collectingProcess) addClient(address string, client *clientHandler) {
+	cp.clientsLock.Lock()
+	defer cp.clientsLock.Unlock()
+	cp.clients[address] = client
+}
+
 func (cp *collectingProcess) deleteClient(name string) {
+	cp.clientsLock.Lock()
+	defer cp.clientsLock.Unlock()
 	delete(cp.clients, name)
 }
 
 func (cp *collectingProcess) getClientCount() int {
+	cp.clientsLock.RLock()
+	defer cp.clientsLock.RUnlock()
 	return len(cp.clients)
 }
 

pkg/collector/tcp.go

@@ -44,7 +44,7 @@ func (cp *collectingProcess) handleTCPClient(conn net.Conn, wg *sync.WaitGroup)
 	defer wg.Done()
 	address := conn.RemoteAddr().String()
 	client := cp.createClient()
-	cp.clients[address] = client
+	cp.addClient(address, client)
 	go func() {
 		defer conn.Close()
 	out:

pkg/collector/udp.go

@@ -69,7 +69,7 @@ func (cp *collectingProcess) startUDPServer() {
 func (cp *collectingProcess) handleUDPClient(address net.Addr, wg *sync.WaitGroup) {
 	if _, exist := cp.clients[address.String()]; !exist {
 		client := cp.createClient()
-		cp.clients[address.String()] = client
+		cp.addClient(address.String(), client)
 		wg.Add(1)
 		defer wg.Done()
 		go func() {