"Add mutex to avoid race condition (#90)

Signed-off-by: Aneesh Puttur <aputtur@redhat.com>

Author: aneeshkp

v2/server.go

@@ -61,7 +61,7 @@ var once sync.Once
 var ServerInstance *Server
 var healthCheckPause = 2 * time.Second
 
-type serverStatus int
+type ServerStatus int
 
 const (
 	API_VERSION           = "2.0"
@@ -88,8 +88,9 @@ type Server struct {
 	httpServer              *http.Server
 	pubSubAPI               *pubsubv1.API
 	subscriberAPI           *subscriberApi.API
-	status                  serverStatus
+	status                  ServerStatus
 	statusReceiveOverrideFn func(e cloudevents.Event, dataChan *channel.DataChan) error
+	statusLock              sync.RWMutex
 }
 
 // SubscriptionInfo
@@ -279,21 +280,37 @@ func (s *Server) Port() int {
 
 // Ready gives the status of the server
 func (s *Server) Ready() bool {
+	s.statusLock.RLock()
+	defer s.statusLock.RUnlock()
 	return s.status == started
 }
 
+// SetStatus safely updates the server status
+func (s *Server) SetStatus(newStatus ServerStatus) {
+	s.statusLock.Lock()
+	defer s.statusLock.Unlock()
+	s.status = newStatus
+}
+
+func (s *Server) GetStatus() ServerStatus {
+	s.statusLock.RLock()
+	defer s.statusLock.RUnlock()
+	return s.status
+}
+
 // GetHostPath  returns hostpath
 func (s *Server) GetHostPath() *types.URI {
 	return types.ParseURI(fmt.Sprintf("http://localhost:%d%s", s.port, s.apiPath))
 }
 
 // Start will start res routes service
 func (s *Server) Start() {
-	if s.status == started || s.status == starting {
+	currentStatus := s.GetStatus()
+	if currentStatus == started || currentStatus == starting {
 		log.Infof("Server is already running at port %d", s.port)
 		return
 	}
-	s.status = starting
+	s.SetStatus(starting)
 	r := mux.NewRouter()
 
 	api := r.PathPrefix(s.apiPath).Subrouter()
@@ -482,7 +499,7 @@ func (s *Server) Start() {
 
 	log.Infof("starting v2 rest api server at port %d, endpoint %s", s.port, s.apiPath)
 	go wait.Until(func() {
-		s.status = started
+		s.SetStatus(started)
 		s.httpServer = &http.Server{
 			ReadHeaderTimeout: HTTPReadHeaderTimeout,
 			Addr:              fmt.Sprintf(":%d", s.port),
@@ -491,7 +508,7 @@ func (s *Server) Start() {
 		err := s.httpServer.ListenAndServe()
 		if err != nil {
 			log.Errorf("restarting due to error with api server %s\n", err.Error())
-			s.status = failed
+			s.SetStatus(failed)
 		}
 	}, 1*time.Second, s.closeCh)
 }

v2/server_test.go

@@ -46,20 +46,22 @@ import (
 var (
 	server *restapi.Server
 
-	eventOutCh      chan *channel.DataChan
-	closeCh         chan struct{}
-	wg              sync.WaitGroup
-	port            = 8990
-	apHost          = "localhost"
-	apPath          = "/api/ocloudNotifications/v2/"
-	resource        = "/east-edge-10/Node3/sync/sync-status/sync-state"
-	resourceInvalid = "/east-edge-10/Node3/invalid"
-	storePath       = "."
-	ObjSub          pubsub.PubSub
-	ObjPub          pubsub.PubSub
-	testSource      = "/sync/sync-status/sync-state"
-	testType        = "event.synchronization-state-change"
-	endpoint        = "http://localhost:8990//api/ocloudNotifications/v2/dummy"
+	eventOutCh       chan *channel.DataChan
+	closeCh          chan struct{}
+	wg               sync.WaitGroup
+	port             = 8990
+	apHost           = "localhost"
+	apPath           = "/api/ocloudNotifications/v2/"
+	resource         = "/east-edge-10/Node3/sync/sync-status/sync-state"
+	resourceInvalid  = "/east-edge-10/Node3/invalid"
+	storePath        = "."
+	ObjSub           pubsub.PubSub
+	ObjPub           pubsub.PubSub
+	testSource       = "/sync/sync-status/sync-state"
+	testType         = "event.synchronization-state-change"
+	endpoint         = "http://localhost:8990//api/ocloudNotifications/v2/dummy"
+	onceCloseEvent   sync.Once
+	onceCloseCloseCh sync.Once
 )
 
 func onReceiveOverrideFn(e cloudevents.Event, d *channel.DataChan) error {
@@ -772,8 +774,15 @@ func TestServer_End(*testing.T) {
 	}
 	os.Remove("pub.json")
 	os.Remove("sub.json")
-	close(eventOutCh)
-	close(closeCh)
+	// hanlding go test -race ./...
+	// by closing channel only once
+	onceCloseEvent.Do(func() {
+		close(eventOutCh)
+	})
+
+	onceCloseCloseCh.Do(func() {
+		close(closeCh)
+	})
 }
 
 // Rest client to make http request
@@ -829,3 +838,35 @@ func (r *Rest) PostEvent(url *types.URI, e event.Event) error {
 	}
 	return nil
 }
+
+func TestServerStatusConcurrency(*testing.T) {
+	s := &restapi.Server{} // Adjust import as needed if your package name differs
+	statuses := []restapi.ServerStatus{
+		0,
+		1,
+		2,
+		3,
+	}
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	// Writer goroutine: updates status multiple times
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			s.SetStatus(statuses[i%len(statuses)])
+			time.Sleep(1 * time.Millisecond)
+		}
+	}()
+
+	// Reader goroutine: reads Ready() status
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			_ = s.Ready()
+			time.Sleep(1 * time.Millisecond)
+		}
+	}()
+
+	wg.Wait()
+}