Fix deadlock with ccevents

Signed-off-by: Marcus Brandenburger <bur@zurich.ibm.com>

Author: mbrandenburger

platform/fabric/events.go

@@ -8,49 +8,73 @@ package fabric
 
 import (
 	"sync"
+	"sync/atomic"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/fabric/core/generic/committer"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/events"
 )
 
 // EventListener models the parameters to use for chaincode listening.
 type EventListener struct {
-	sync.RWMutex
 	chaincodeListener chan *committer.ChaincodeEvent
 	subscriber        events.Subscriber
 	chaincodeName     string
-	closing           bool
+
+	isClosed       atomic.Bool
+	wg             sync.WaitGroup
+	shutdownOnce   sync.Once
+	shutdownSignal chan struct{}
+	subscribeOnce  sync.Once
 }
 
 func newEventListener(subscriber events.Subscriber, chaincodeName string) *EventListener {
 	return &EventListener{
-		chaincodeName: chaincodeName,
-		subscriber:    subscriber,
+		chaincodeName:  chaincodeName,
+		subscriber:     subscriber,
+		shutdownSignal: make(chan struct{}),
 	}
 }
 
 // ChaincodeEvents returns a channel from which chaincode events emitted by transaction functions in the specified chaincode can be read.
 func (e *EventListener) ChaincodeEvents() chan *committer.ChaincodeEvent {
-	e.chaincodeListener = make(chan *committer.ChaincodeEvent, 1)
-	e.subscriber.Subscribe(e.chaincodeName, e)
+	e.subscribeOnce.Do(func() {
+		e.chaincodeListener = make(chan *committer.ChaincodeEvent, 1)
+		e.subscriber.Subscribe(e.chaincodeName, e)
+	})
+
 	return e.chaincodeListener
 }
 
 // CloseChaincodeEvents closes the channel from which chaincode events are read.
 func (e *EventListener) CloseChaincodeEvents() {
-	e.Lock()
-	e.closing = true
-	e.Unlock()
+	e.shutdownOnce.Do(func() {
+		e.isClosed.Store(true)
+
+		// first unsubscribe
+		e.subscriber.Unsubscribe(e.chaincodeName, e)
+
+		// tell every ongoing OnReceive to stop writing into chaincodeListener channel
+		close(e.shutdownSignal)
 
-	e.subscriber.Unsubscribe(e.chaincodeName, e)
-	close(e.chaincodeListener)
+		// wait until all OnReceive calls have finished
+		e.wg.Wait()
+
+		// signal consumer that there is no more coming
+		close(e.chaincodeListener)
+	})
 }
 
 // OnReceive pushes events to the listener
 func (e *EventListener) OnReceive(event events.Event) {
-	e.RLock()
-	defer e.RUnlock()
-	if !e.closing {
-		e.chaincodeListener <- event.Message().(*committer.ChaincodeEvent)
+	if event == nil || e.isClosed.Load() {
+		return
+	}
+
+	e.wg.Add(1)
+	defer e.wg.Done()
+
+	select {
+	case <-e.shutdownSignal:
+	case e.chaincodeListener <- event.Message().(*committer.ChaincodeEvent):
 	}
 }