channel: improve documentation and API surface

Author: daniel-abramov

pkg/common/channel.go

@@ -2,22 +2,34 @@ package common
 
 import "sync/atomic"
 
+// In Go, unbounded channel means something different than what it means in Rust.
+// I.e. unlike Rust, "unbounded" in Go means that the channel has **no buffer**,
+// meaning that each attempt to send will block the channel until the receiver
+// reads it. Majority of primitives here in `waterfall` are designed under assumption
+// that sending is not blocking.
+const UnboundedChannelSize = 128
+
 // Creates a new channel, returns two counterparts of it where one can only send and another can only receive.
 // Unlike traditional Go channels, these allow the receiver to mark the channel as closed which would then fail
 // to send any messages to the channel over `Send“.
 func NewChannel[M any]() (Sender[M], Receiver[M]) {
-	channel := make(chan M, 128)
+	channel := make(chan M, UnboundedChannelSize)
 	closed := &atomic.Bool{}
 	sender := Sender[M]{channel, closed}
 	receiver := Receiver[M]{channel, closed}
 	return sender, receiver
 }
 
+// Sender counterpart of the channel.
 type Sender[M any] struct {
-	channel        chan<- M
+	// The channel itself.
+	channel chan<- M
+	// Atomic variable that indicates whether the channel is closed.
 	receiverClosed *atomic.Bool
 }
 
+// Tries to send a message if the channel is not closed.
+// Returns the message back if the channel is closed.
 func (s *Sender[M]) Send(message M) *M {
 	if !s.receiverClosed.Load() {
 		s.channel <- message
@@ -27,11 +39,32 @@ func (s *Sender[M]) Send(message M) *M {
 	}
 }
 
+// The receiver counterpart of the channel.
 type Receiver[M any] struct {
-	Channel        <-chan M
+	// The channel itself. It's public, so that we can combine it in `select` statements.
+	Channel <-chan M
+	// Atomic variable that indicates whether the channel is closed.
 	receiverClosed *atomic.Bool
 }
 
-func (r *Receiver[M]) Close() {
+// Marks the channel as closed, which means that no messages could be sent via this channel.
+// Any attempt to send a message would result in an error. This is similar to closing the
+// channel except that we don't close the underlying channel (since in Go receivers can't
+// close the channel).
+//
+// This function reads (in a non-blocking way) all pending messages until blocking. Otherwise,
+// they will stay forver in a channel and get lost.
+func (r *Receiver[M]) Close() []M {
 	r.receiverClosed.Store(true)
+
+	messages := make([]M, 0)
+	for {
+		msg, ok := <-r.Channel
+		if !ok {
+			break
+		}
+		messages = append(messages, msg)
+	}
+
+	return messages
 }

pkg/conference/processor.go

@@ -26,22 +26,11 @@ func (c *Conference) processMessages() {
 		if len(c.participants) == 0 {
 			c.logger.Info("No more participants, stopping the conference")
 			// Close the channel so that the sender can't push any messages.
-			c.matrixMessages.Close()
-
-			// Let's read remaining messages from the channel (otherwise the caller will be
-			// blocked in case of unbuffered channels). We must read **all** pending messages.
-			messages := make([]MatrixMessage, 0)
-			for {
-				msg, ok := <-c.matrixMessages.Channel
-				if !ok {
-					break
-				}
-				messages = append(messages, msg)
-			}
+			unreadMessages := c.matrixMessages.Close()
 
 			// Send the information that we ended to the owner and pass the message
 			// that we did not process (so that we don't drop it silently).
-			c.endNotifier.Notify(messages)
+			c.endNotifier.Notify(unreadMessages)
 			return
 		}
 	}

pkg/conference/start.go

@@ -43,7 +43,7 @@ func StartConference(
 		matrixMessages: receiver,
 		endNotifier:    conferenceEndNotifier,
 		participants:   make(map[ParticipantID]*Participant),
-		peerMessages:   make(chan common.Message[ParticipantID, peer.MessageContent], 128),
+		peerMessages:   make(chan common.Message[ParticipantID, peer.MessageContent], common.UnboundedChannelSize),
 		logger:         logrus.WithFields(logrus.Fields{"conf_id": confID}),
 	}
 

pkg/router.go

@@ -45,7 +45,7 @@ func newRouter(matrix *signaling.MatrixClient, config conf.Config) chan<- Router
 		matrix:          matrix,
 		conferenceSinks: make(map[string]*common.Sender[conf.MatrixMessage]),
 		config:          config,
-		channel:         make(chan RouterMessage, 128),
+		channel:         make(chan RouterMessage, common.UnboundedChannelSize),
 	}
 
 	// Start the main loop of the Router.