Use Go in a more idiomatic way

cmd/sfu/main.go

@@ -100,11 +100,16 @@ func main() {
 		return
 	}
 
-	// Create a router to route incoming To-Device messages to the right conference.
-	routerChannel := routing.NewRouter(matrixClient, connectionFactory, config.Conference)
+	// Create a channel which we'll use to send events to the router.
+	matrixEvents := make(chan *event.Event)
+
+	// Start a router that will receive events from the matrix client and route them to the appropriate conference.
+	routing.RunRouter(matrixClient, connectionFactory, matrixEvents, config.Conference)
 
 	// Start matrix client sync. This function will block until the sync fails.
 	matrixClient.RunSyncing(func(e *event.Event) {
-		routerChannel <- e
+		matrixEvents <- e
 	})
+
+	close(matrixEvents)
 }

pkg/channel/sink.go

@@ -0,0 +1,85 @@
+package channel
+
+import (
+	"errors"
+	"sync/atomic"
+)
+
+var ErrSinkSealed = errors.New("The channel is sealed")
+
+// SinkWithSender is a helper struct that allows to send messages to a message sink.
+// The SinkWithSender abstracts the message sink which has a certain sender, so that
+// the sender does not have to be specified every time a message is sent.
+// At the same it guarantees that the caller can't alter the `sender`, which means that
+// the sender can't impersonate another sender (and we guarantee this on a compile-time).
+type SinkWithSender[SenderType comparable, MessageType any] struct {
+	// The sender of the messages. This is useful for multiple-producer-single-consumer scenarios.
+	sender SenderType
+	// The message sink to which the messages are sent.
+	messageSink chan<- Message[SenderType, MessageType]
+	// A channel that is used to indicate that our channel is considered sealed. It's akin
+	// to a close indication without really closing the channel. We don't want to close
+	// the channel here since we know that the sink is shared between multiple producers,
+	// so we only disallow sending to the sink at this point.
+	sealed chan struct{}
+	// A "mutex" that is used to protect the act of closing `sealed`.
+	alreadySealed atomic.Bool
+}
+
+// Creates a new MessageSink. The function is generic allowing us to use it for multiple use cases.
+// Note that since the current implementation accepts a channel, it's **not responsible** for closing it.
+func NewSink[S comparable, M any](sender S, messageSink chan<- Message[S, M]) *SinkWithSender[S, M] {
+	return &SinkWithSender[S, M]{
+		sender:      sender,
+		messageSink: messageSink,
+		sealed:      make(chan struct{}),
+	}
+}
+
+// Sends a message to the message sink. Blocks if the sink is full!
+func (s *SinkWithSender[S, M]) Send(message M) error {
+	if s.alreadySealed.Load() {
+		return ErrSinkSealed
+	}
+
+	messageWithSender := Message[S, M]{
+		Sender:  s.sender,
+		Content: message,
+	}
+
+	select {
+	case <-s.sealed:
+		return ErrSinkSealed
+	case s.messageSink <- messageWithSender:
+		return nil
+	}
+}
+
+// Seals the channel, which means that no messages could be sent via this channel.
+// Any attempt to send a message after `Seal()` returns will result in an error.
+// Note that it does not mean (does not guarantee) that any existing senders that are
+// waiting on the send to unblock won't send the message to the recipient (this case
+// can happen if buffered channels are used). The existing senders will either unblock
+// at this point and get an error that the channel is sealed or will unblock by sending
+// the message to the recipient (should the recipient be ready to consume at this point).
+func (s *SinkWithSender[S, M]) Seal() {
+	if !s.alreadySealed.CompareAndSwap(false, true) {
+		return
+	}
+
+	select {
+	case <-s.sealed:
+		return
+	default:
+		close(s.sealed)
+	}
+}
+
+// Messages that are sent from the peer to the conference in order to communicate with other peers.
+// Since each peer is isolated from others, it can't influence the state of other peers directly.
+type Message[SenderType comparable, MessageType any] struct {
+	// The sender of the message.
+	Sender SenderType
+	// The content of the message.
+	Content MessageType
+}

pkg/conference/matrix_message_processing.go

@@ -3,7 +3,7 @@ package conference
 import (
 	"time"
 
-	"github.com/matrix-org/waterfall/pkg/common"
+	"github.com/matrix-org/waterfall/pkg/channel"
 	"github.com/matrix-org/waterfall/pkg/conference/participant"
 	"github.com/matrix-org/waterfall/pkg/peer"
 	"github.com/matrix-org/waterfall/pkg/signaling"
@@ -15,9 +15,8 @@ import (
 type MessageContent interface{}
 
 type MatrixMessage struct {
-	Sender   participant.ID
-	Content  MessageContent
-	RawEvent *event.Event
+	Sender  participant.ID
+	Content MessageContent
 }
 
 // New participant tries to join the conference.
@@ -48,34 +47,32 @@ func (c *Conference) onNewParticipant(id participant.ID, inviteEvent *event.Call
 		}
 		sdpAnswer = answer
 	} else {
-		messageSink := common.NewMessageSink(id, c.peerMessages)
+		messageSink := channel.NewSink(id, c.peerMessages)
 
 		peerConnection, answer, err := peer.NewPeer(c.connectionFactory, inviteEvent.Offer.SDP, messageSink, logger)
 		if err != nil {
 			logger.WithError(err).Errorf("Failed to process SDP offer")
 			return err
 		}
 
-		heartbeat := common.Heartbeat{
-			Interval: time.Duration(c.config.HeartbeatConfig.Interval) * time.Second,
-			Timeout:  time.Duration(c.config.HeartbeatConfig.Timeout) * time.Second,
-			SendPing: func() bool {
-				return p.SendDataChannelMessage(event.Event{
-					Type:    event.FocusCallPing,
-					Content: event.Content{},
-				}) == nil
-			},
-			OnTimeout: func() {
-				messageSink.Send(peer.LeftTheCall{event.CallHangupKeepAliveTimeout})
-			},
+		pingEvent := event.Event{
+			Type:    event.FocusCallPing,
+			Content: event.Content{},
+		}
+
+		heartbeat := participant.HeartbeatConfig{
+			Interval:  time.Duration(c.config.HeartbeatConfig.Interval) * time.Second,
+			Timeout:   time.Duration(c.config.HeartbeatConfig.Timeout) * time.Second,
+			SendPing:  func() bool { return p.SendDataChannelMessage(pingEvent) == nil },
+			OnTimeout: func() { messageSink.Send(peer.LeftTheCall{event.CallHangupKeepAliveTimeout}) },
 		}
 
 		p = &participant.Participant{
 			ID:              id,
 			Peer:            peerConnection,
 			Logger:          logger,
 			RemoteSessionID: inviteEvent.SenderSessionID,
-			HeartbeatPong:   heartbeat.Start(),
+			Pong:            heartbeat.Start(),
 		}
 
 		c.tracker.AddParticipant(p)

pkg/conference/matrix_worker.go

@@ -3,27 +3,27 @@ package conference
 import (
 	"time"
 
-	"github.com/matrix-org/waterfall/pkg/common"
 	"github.com/matrix-org/waterfall/pkg/signaling"
+	"github.com/matrix-org/waterfall/pkg/worker"
 	"github.com/sirupsen/logrus"
 	"maunium.net/go/mautrix/id"
 )
 
 type matrixWorker struct {
-	worker   *common.Worker[signaling.MatrixMessage]
+	worker   *worker.Worker[signaling.MatrixMessage]
 	deviceID id.DeviceID
 }
 
 func newMatrixWorker(handler signaling.MatrixSignaler) *matrixWorker {
-	workerConfig := common.WorkerConfig[signaling.MatrixMessage]{
+	workerConfig := worker.Config[signaling.MatrixMessage]{
 		ChannelSize: 128,
 		Timeout:     time.Hour,
 		OnTimeout:   func() {},
 		OnTask:      func(msg signaling.MatrixMessage) { handler.SendMessage(msg) },
 	}
 
 	matrixWorker := &matrixWorker{
-		worker:   common.StartWorker(workerConfig),
+		worker:   worker.StartWorker(workerConfig),
 		deviceID: handler.DeviceID(),
 	}
 

pkg/common/heartbeat.go → pkg/conference/participant/heartbeat.go

@@ -1,13 +1,13 @@
-package common
+package participant
 
 import (
 	"time"
 )
 
 type Pong struct{}
 
-// Heartbeat defines the configuration for a heartbeat.
-type Heartbeat struct {
+// HeartbeatConfig defines the configuration for a heartbeat.
+type HeartbeatConfig struct {
 	// How often to send pings.
 	Interval time.Duration
 	// After which time to consider the communication stalled.
@@ -23,8 +23,8 @@ type Heartbeat struct {
 // on `PongChannel` for `Timeout`. If no response is received within `Timeout`, `OnTimeout` is called.
 // The goroutine stops once the channel is closed or upon handling the `OnTimeout`. The returned channel
 // is what the caller should use to inform about the reception of a pong.
-func (h *Heartbeat) Start() chan<- Pong {
-	pong := make(chan Pong, UnboundedChannelSize)
+func (h *HeartbeatConfig) Start() chan<- Pong {
+	pong := make(chan Pong, 1)
 
 	go func() {
 		ticker := time.NewTicker(h.Interval)
@@ -52,7 +52,7 @@ func (h *Heartbeat) Start() chan<- Pong {
 
 // Tries to send a ping message using `SendPing` and retry it if it fails.
 // Returns `true` if the ping was sent successfully.
-func (h *Heartbeat) sendWithRetry() bool {
+func (h *HeartbeatConfig) sendWithRetry() bool {
 	const retries = 3
 	retryInterval := h.Timeout / retries