Merge branch 'master' into remove-auto-subscription-id

Author: yuce

internal/invocation/invocation_service.go

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License")
  * you may not use this file except in compliance with the License.
@@ -19,7 +19,7 @@ package invocation
 import (
 	"context"
 	"fmt"
-	"sync/atomic"
+	"sync"
 	"time"
 
 	"github.com/hazelcast/hazelcast-go-client/hzerrors"
@@ -29,11 +29,6 @@ import (
 	"github.com/hazelcast/hazelcast-go-client/internal/proto"
 )
 
-const (
-	ready   = 0
-	stopped = 1
-)
-
 var serviceSubID = event.NextSubscriptionID()
 
 type Handler interface {
@@ -53,13 +48,11 @@ type Service struct {
 	removeCh chan int64
 	executor *stripeExecutor
 	logger   logger.LogAdaptor
-	state    int32
+	stateMu  *sync.RWMutex
+	running  bool
 }
 
-func NewService(
-	handler Handler,
-	eventDispatcher *event.DispatchService,
-	logger logger.LogAdaptor) *Service {
+func NewService(handler Handler, ed *event.DispatchService, lg logger.LogAdaptor) *Service {
 	s := &Service{
 		requestCh:       make(chan Invocation),
 		urgentRequestCh: make(chan Invocation),
@@ -69,14 +62,20 @@ func NewService(
 		groupLostCh:     make(chan *GroupLostEvent),
 		invocations:     map[int64]Invocation{},
 		handler:         handler,
-		eventDispatcher: eventDispatcher,
-		logger:          logger,
-		state:           ready,
+		eventDispatcher: ed,
+		logger:          lg,
+		stateMu:         &sync.RWMutex{},
+		running:         true,
 		executor:        newStripeExecutor(),
 	}
 	s.eventDispatcher.Subscribe(EventGroupLost, serviceSubID, func(event event.Event) {
 		go func() {
-			s.groupLostCh <- event.(*GroupLostEvent)
+			select {
+			case s.groupLostCh <- event.(*GroupLostEvent):
+				return
+			case <-s.doneCh:
+				return
+			}
 		}()
 	})
 	s.executor.start()
@@ -85,9 +84,12 @@ func NewService(
 }
 
 func (s *Service) Stop() {
-	if !atomic.CompareAndSwapInt32(&s.state, ready, stopped) {
+	s.stateMu.Lock()
+	defer s.stateMu.Unlock()
+	if !s.running {
 		return
 	}
+	s.running = false
 	s.executor.stop()
 	close(s.doneCh)
 }
@@ -249,7 +251,9 @@ func (s *Service) unregisterInvocation(correlationID int64) Invocation {
 }
 
 func (s *Service) handleGroupLost(e *GroupLostEvent) {
-	if atomic.LoadInt32(&s.state) != ready {
+	s.stateMu.RLock()
+	defer s.stateMu.RUnlock()
+	if !s.running {
 		return
 	}
 	for corrID, inv := range s.invocations {