Fix race conditions (#51)

* Fix race conditions
* updates

Author: Tim Middleton

coherence/event.go

@@ -100,7 +100,7 @@ func newEventEmitter[L comparable, E any]() *eventEmitter[L, E] {
 	}
 }
 
-// on register a callback to be notified when an event associated with
+// on registers a callback to be notified when an event associated with
 // the specified label is raised.
 func (ee *eventEmitter[L, E]) on(label L, callback func(E)) {
 	cbs, present := ee.callbacks[label]

coherence/session.go

@@ -285,17 +285,22 @@ func (s *Session) ID() string {
 
 // Close closes a connection.
 func (s *Session) Close() {
+	s.mutex.Lock()
 	if !s.closed {
 		s.maps = make(map[string]interface{}, 0)
 		s.caches = make(map[string]interface{}, 0)
 		err := s.conn.Close()
 		s.closed = true
+
+		s.mutex.Unlock()
 		s.dispatch(Closed, func() SessionLifecycleEvent {
 			return newSessionLifecycleEvent(s, Closed)
 		})
 		if err != nil {
-			log.Printf("Unable to close session %s %v", s.sessionID, err)
+			log.Printf("unable to close session %s %v", s.sessionID, err)
 		}
+	} else {
+		defer s.mutex.Unlock()
 	}
 }
 
@@ -697,8 +702,7 @@ func (s *SessionOptions) String() string {
 	return sb.String()
 }
 
-func (s *Session) dispatch(eventType SessionLifecycleEventType,
-	creator func() SessionLifecycleEvent) {
+func (s *Session) dispatch(eventType SessionLifecycleEventType, creator func() SessionLifecycleEvent) {
 	if len(s.lifecycleListeners) > 0 {
 		event := creator()
 		for _, l := range s.lifecycleListeners {

examples/basic/expiry/main.go

@@ -60,5 +60,5 @@ func main() {
 	if value, err = namedCache.Get(ctx, 1); err != nil {
 		panic(err)
 	}
-	fmt.Printf("Value for key 1 is %v, should be nil pointer as entry longer exists\n", value)
+	fmt.Printf("Value for key 1 is %v, should be nil pointer as entry no longer exists\n", value)
 }

examples/basic/expiry_cache/main.go

@@ -53,7 +53,7 @@ func main() {
 	if value, err = namedCache.Get(ctx, 1); err != nil {
 		panic(err)
 	}
-	fmt.Printf("Value for key 1 is %v, should be nil pointer as entry longer exists\n", value)
+	fmt.Printf("Value for key 1 is %v, should be nil pointer as entry no longer exists\n", value)
 
 	// If we do call PutWithExpiry, this expiry value will override the default
 	fmt.Println("Issue PutWithExpiry key 1, value \"one\" with expiry 10 seconds")

test/e2e/standalone/event_test.go

@@ -104,10 +104,13 @@ func TestMapAndLifecycleEventsAll(t *testing.T) {
 	runBasicTests(g, namedMap, namedMap.Name(), &expected, -1)
 }
 
+// TestEventDisconnect tests to ensure that if we get a disconnect, then we can
 func TestEventDisconnect(t *testing.T) {
 	t.Setenv("COHERENCE_SESSION_DEBUG", "true")
 	//g, session := initTest(t)
-	g, session := initTest(t, coherence.WithDisconnectTimeout(time.Duration(130000)*time.Millisecond))
+	g, session := initTest(t,
+		coherence.WithDisconnectTimeout(time.Duration(130000)*time.Millisecond),
+		coherence.WithReadyTimeout(time.Duration(70)*time.Second))
 	defer session.Close()
 
 	namedCache := GetNamedCache[string, string](g, session, "test-reconnect-cache")
@@ -119,7 +122,7 @@ func TestEventDisconnect(t *testing.T) {
 }
 
 // TestEventDisconnectWithReadyTimeoutDelay tests that the ready timeout is honoured,
-// as we have stopped the gRPC proxy.
+// as we have stopped the gRPC proxy before the test runs.
 func TestEventDisconnectWithReadyTimeoutDelay(t *testing.T) {
 	t.Setenv("COHERENCE_SESSION_DEBUG", "true")
 
@@ -275,9 +278,6 @@ func RunTestReconnect(g *gomega.WithT, namedMap coherence.NamedMap[string, strin
 		fmt.Println("Issue stop of $GRPC:GrpcProxy")
 		_, err = IssuePostRequest("http://127.0.0.1:30000/management/coherence/cluster/services/$GRPC:GrpcProxy/members/1/stop")
 		g.Expect(err).ShouldNot(gomega.HaveOccurred())
-
-		// sleep for 70 seconds to give the shutdown time to take effect
-		Sleep(70)
 	}
 
 	// add another 'additional' mutations