Add mechanism to multi-channel to notify the txer that the underlay has closed. Fixes #176

handler.go

@@ -132,7 +132,7 @@ func (self CloseHandlerF) HandleClose(ch Channel) {
 	self(ch)
 }
 
-type MessageSourceF func() (Sendable, error)
+type MessageSourceF func(notifer *CloseNotifier) (Sendable, error)
 
 type UnderlayHandler interface {
 	// Start is called after the MultiChannel has been created with the first underlay

impl.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/foundation/v2/concurrenz"
+	"github.com/openziti/foundation/v2/debugz"
 	"github.com/openziti/foundation/v2/info"
 	"github.com/openziti/foundation/v2/sequence"
 	"github.com/pkg/errors"
@@ -108,6 +109,7 @@ func NewChannelWithUnderlay(logicalName string, underlay Underlay, bindHandler B
 		return nil, err
 	}
 
+	debugz.DumpLocalStack()
 	impl.startMultiplex()
 
 	return impl, nil

multi.go

@@ -122,8 +122,7 @@ func NewMultiChannel(config *MultiChannelConfig) (MultiChannel, error) {
 		return nil, err
 	}
 
-	go impl.Rxer(config.Underlay)
-	go impl.Txer(config.Underlay)
+	impl.startMultiplex(config.Underlay)
 	go impl.underlayHandler.Start(impl)
 
 	impl.underlayHandler.HandleUnderlayAccepted(impl, config.Underlay)
@@ -146,14 +145,19 @@ func (self *multiChannelImpl) AcceptUnderlay(underlay Underlay) bool {
 	self.headers.Store(underlay.Headers())
 	self.underlays.Append(underlay)
 
-	go self.Rxer(underlay)
-	go self.Txer(underlay)
+	self.startMultiplex(underlay)
 
 	self.underlayHandler.HandleUnderlayAccepted(self, underlay)
 
 	return true
 }
 
+func (self *multiChannelImpl) startMultiplex(underlay Underlay) {
+	notifier := NewCloseNotifier()
+	go self.Rxer(underlay, notifier)
+	go self.Txer(underlay, notifier)
+}
+
 func (self *multiChannelImpl) GetUnderlayCountsByType() map[string]int {
 	result := map[string]int{}
 	for _, u := range self.underlays.Value() {
@@ -402,12 +406,14 @@ func (self *multiChannelImpl) Tx(underlay Underlay, sendable Sendable, writeTime
 	return nil
 }
 
-func (self *multiChannelImpl) CloseUnderlay(underlay Underlay) {
+func (self *multiChannelImpl) closeUnderlay(underlay Underlay, notifier *CloseNotifier) {
 	self.lock.Lock()
 	if err := underlay.Close(); err != nil {
 		pfxlog.Logger().WithField("context", self.Label()).WithError(err).Error("error closing underlay")
 	}
 
+	notifier.NotifyClosed()
+
 	underlayRemoved := false
 	self.underlays.DeleteIf(func(element Underlay) bool {
 		if underlay == element {
@@ -427,8 +433,8 @@ func (self *multiChannelImpl) GetTimeSinceLastRead() time.Duration {
 	return time.Duration(info.NowInMilliseconds()-atomic.LoadInt64(&self.lastRead)) * time.Millisecond
 }
 
-func (self *multiChannelImpl) Txer(underlay Underlay) {
-	defer self.CloseUnderlay(underlay)
+func (self *multiChannelImpl) Txer(underlay Underlay, notifier *CloseNotifier) {
+	defer self.closeUnderlay(underlay, notifier)
 
 	log := pfxlog.ContextLogger(self.Label())
 
@@ -440,7 +446,7 @@ func (self *multiChannelImpl) Txer(underlay Underlay) {
 	messageSource := self.underlayHandler.GetMessageSource(underlay)
 
 	for {
-		sendable, err := messageSource()
+		sendable, err := messageSource(notifier)
 		if err != nil {
 			return
 		}
@@ -456,8 +462,8 @@ func (self *multiChannelImpl) Txer(underlay Underlay) {
 	}
 }
 
-func (self *multiChannelImpl) Rxer(underlay Underlay) {
-	defer self.CloseUnderlay(underlay)
+func (self *multiChannelImpl) Rxer(underlay Underlay, notifier *CloseNotifier) {
+	defer self.closeUnderlay(underlay, notifier)
 
 	log := pfxlog.ContextLogger(self.Label())
 	log.Debug("started")
@@ -592,3 +598,24 @@ func (self *UnderlayConstraints) Apply(ch MultiChannel, factory GroupedUnderlayF
 		}
 	}
 }
+
+func NewCloseNotifier() *CloseNotifier {
+	return &CloseNotifier{
+		c: make(chan struct{}),
+	}
+}
+
+type CloseNotifier struct {
+	c        chan struct{}
+	notified atomic.Bool
+}
+
+func (self *CloseNotifier) NotifyClosed() {
+	if self.notified.CompareAndSwap(false, true) {
+		close(self.c)
+	}
+}
+
+func (self *CloseNotifier) GetCloseNotify() <-chan struct{} {
+	return self.c
+}

multi_test.go

@@ -219,7 +219,7 @@ func (self *priorityChannelBase) GetPrioritySender() Sender {
 	return self.prioritySender
 }
 
-func (self *priorityChannelBase) GetNextMsgDefault() (Sendable, error) {
+func (self *priorityChannelBase) GetNextMsgDefault(notifier *CloseNotifier) (Sendable, error) {
 	select {
 	case msg := <-self.defaultMsgChan:
 		return msg, nil
@@ -229,17 +229,21 @@ func (self *priorityChannelBase) GetNextMsgDefault() (Sendable, error) {
 		return msg, nil
 	case <-self.GetCloseNotify():
 		return nil, io.EOF
+	case <-notifier.GetCloseNotify():
+		return nil, io.EOF
 	}
 }
 
-func (self *priorityChannelBase) GetNextPriorityMsg() (Sendable, error) {
+func (self *priorityChannelBase) GetNextPriorityMsg(notifier *CloseNotifier) (Sendable, error) {
 	select {
 	case msg := <-self.priorityMsgChan:
 		return msg, nil
 	case msg := <-self.retryMsgChan:
 		return msg, nil
 	case <-self.GetCloseNotify():
 		return nil, io.EOF
+	case <-notifier.GetCloseNotify():
+		return nil, io.EOF
 	}
 }