From: elinuxhenrik <henrik.b.andersson@est.tech>
Date: Fri, 5 Nov 2021 07:00:12 +0000 (+0100)
Subject: Fix locking in DMaaP Mediator Producer
X-Git-Tag: 1.2.0~53
X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=commitdiff_plain;h=d289079b1b15286e994bc29b98cd732fc32fb06f;hp=3923a2bd9989a6988cdb29d464c86cb32ced5439;p=nonrtric.git

Fix locking in DMaaP Mediator Producer

Issue-ID: NONRTRIC-631
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>
Change-Id: I38d2255be7f180a10559152c385ee88fc8471ce4
---

diff --git a/dmaap-mediator-producer/internal/jobs/jobs.go b/dmaap-mediator-producer/internal/jobs/jobs.go
index b856e296..1c429424 100644
--- a/dmaap-mediator-producer/internal/jobs/jobs.go
+++ b/dmaap-mediator-producer/internal/jobs/jobs.go
@@ -182,8 +182,6 @@ func (jh *jobHandler) start(mRAddress string) {
 }
 
 func (jh *jobHandler) pollAndDistributeMessages(mRAddress string) {
-	jh.mu.Lock()
-	defer jh.mu.Unlock()
 	log.Debugf("Processing jobs for type: %v", jh.typeId)
 	messagesBody, error := restclient.Get(mRAddress+jh.topicUrl, jh.pollClient)
 	if error != nil {
@@ -195,6 +193,8 @@ func (jh *jobHandler) pollAndDistributeMessages(mRAddress string) {
 
 func (jh *jobHandler) distributeMessages(messages []byte) {
 	if len(messages) > 2 {
+		jh.mu.Lock()
+		defer jh.mu.Unlock()
 		for _, jobInfo := range jh.jobs {
 			go jh.sendMessagesToConsumer(messages, jobInfo)
 		}