Merge pull request hyperledger-labs#366 from perun-network/365-close-…

…ml-ch

Fix Multi-Adjudicator Subscription

channel/multi/subscription.go

@@ -26,6 +26,7 @@ func (a *Adjudicator) Subscribe(ctx context.Context, chID channel.ID) (channel.A
 		events: make(chan channel.AdjudicatorEvent),
 		errors: make(chan error),
 		subs:   []channel.AdjudicatorSubscription{},
+		done:   make(chan struct{}),
 	}
 
 	for _, la := range a.adjudicators {
@@ -37,7 +38,13 @@ func (a *Adjudicator) Subscribe(ctx context.Context, chID channel.ID) (channel.A
 		asub.subs = append(asub.subs, sub)
 
 		go func() {
-			asub.events <- sub.Next()
+			for {
+				select {
+				case asub.events <- sub.Next():
+				case <-asub.done:
+					return
+				}
+			}
 		}()
 
 		go func() {
@@ -53,11 +60,17 @@ type AdjudicatorSubscription struct {
 	subs   []channel.AdjudicatorSubscription
 	events chan channel.AdjudicatorEvent
 	errors chan error
+	done   chan struct{}
 }
 
 // Next returns the next event.
 func (s *AdjudicatorSubscription) Next() channel.AdjudicatorEvent {
-	return <-s.events
+	select {
+	case e := <-s.events:
+		return e
+	case <-s.done:
+		return nil
+	}
 }
 
 // Err blocks until an error occurred and returns it.
@@ -76,5 +89,6 @@ func (s *AdjudicatorSubscription) Close() error {
 	for _, sub := range s.subs {
 		sub.Close()
 	}
+	close(s.done)
 	return nil
 }

client/test/multiledger_dispute.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
 	"perun.network/go-perun/channel"
 	"perun.network/go-perun/client"
 	"perun.network/go-perun/wire"
@@ -131,6 +132,10 @@ func TestMultiLedgerDispute(
 	err = chBobAlice.Settle(ctx, false)
 	require.NoError(err)
 
+	// Close the channels.
+	require.NoError(chAliceBob.Close())
+	require.NoError(chBobAlice.Close())
+
 	// Check final balances.
 	balancesAfter := channel.Balances{
 		{