fixup! Fix deadlock with ccevents

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

Author: mbrandenburger

platform/fabric/events.go

@@ -15,39 +15,41 @@ import (
 
 // EventListener models the parameters to use for chaincode listening.
 type EventListener struct {
-	chaincodeListener chan *committer.ChaincodeEvent
-	subscriber        events.Subscriber
-	chaincodeName     string
+	eventCh       chan *committer.ChaincodeEvent // this is our main event channel
+	subscriber    events.Subscriber
+	chaincodeName string
 
 	subscribeOnce sync.Once
-
-	middleCh chan *committer.ChaincodeEvent
-	closing  chan struct{}
-	closed   chan struct{}
+	middleCh      chan *committer.ChaincodeEvent
+	closing       chan struct{}
+	closed        chan struct{}
 }
 
 func newEventListener(subscriber events.Subscriber, chaincodeName string) *EventListener {
 	return &EventListener{
-		chaincodeName:     chaincodeName,
-		subscriber:        subscriber,
-		chaincodeListener: make(chan *committer.ChaincodeEvent),
-		middleCh:          make(chan *committer.ChaincodeEvent),
-		closing:           make(chan struct{}),
-		closed:            make(chan struct{}),
+		chaincodeName: chaincodeName,
+		subscriber:    subscriber,
+		eventCh:       make(chan *committer.ChaincodeEvent),
+		middleCh:      make(chan *committer.ChaincodeEvent),
+		closing:       make(chan struct{}),
+		closed:        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.subscribeOnce.Do(func() {
-
+		// when a consumer first time calls ChaincodeEvents, we set up the event subscription for the chaincode name
+		// and start this goroutine for graceful closing
 		go func() {
+			// our shutdown helper function
 			exit := func(v *committer.ChaincodeEvent, needSend bool) {
 				close(e.closed)
+				e.subscriber.Unsubscribe(e.chaincodeName, e)
 				if needSend {
-					e.chaincodeListener <- v
+					e.eventCh <- v
 				}
-				close(e.chaincodeListener)
+				close(e.eventCh)
 			}
 
 			for {
@@ -56,27 +58,30 @@ func (e *EventListener) ChaincodeEvents() <-chan *committer.ChaincodeEvent {
 					exit(nil, false)
 					return
 				case v := <-e.middleCh:
+					// we have a new event v received via OnReceive
 					select {
 					case <-e.closing:
 						exit(v, true)
 						return
-					case e.chaincodeListener <- v:
+					case e.eventCh <- v:
+						// forward event v to event channel
 					}
 				}
 			}
 		}()
 
+		// finally we create our subscription
 		e.subscriber.Subscribe(e.chaincodeName, e)
 	})
 
-	return e.chaincodeListener
+	// we always return the event channel; if it is closed
+	return e.eventCh
 }
 
 // CloseChaincodeEvents closes the channel from which chaincode events are read.
 func (e *EventListener) CloseChaincodeEvents() {
 	select {
 	case e.closing <- struct{}{}:
-		e.subscriber.Unsubscribe(e.chaincodeName, e)
 		<-e.closed
 	case <-e.closed:
 	}
@@ -88,6 +93,8 @@ func (e *EventListener) OnReceive(event events.Event) {
 		return
 	}
 
+	// we check if our event listener is already closed
+	// we do this extra select to prioritize the close channel (https://go.dev/ref/spec#Select_statements).
 	select {
 	case <-e.closed:
 		return