refactoring to prep for integration tests

clients.go

@@ -1 +0,0 @@
-package main

cmd/cli/client/client.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net/http"
+	"net/url"
 	"path/filepath"
 	"strings"
 	"sync"
@@ -12,6 +14,7 @@ import (
 
 	"github.com/gorilla/websocket"
 	"github.com/pion/webrtc/v3"
+	"github.com/thoas/go-funk"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 
@@ -21,18 +24,19 @@ import (
 )
 
 type RTCClient struct {
-	conn             *websocket.Conn
-	PeerConn         *webrtc.PeerConnection
-	localTracks      []webrtc.TrackLocal
-	lock             sync.Mutex
-	ctx              context.Context
-	cancel           context.CancelFunc
-	connected        bool
-	iceConnected     bool
-	paused           bool
-	me               *webrtc.MediaEngine // optional, populated only when receiving tracks
-	receivers        []*rtc.ReceiverImpl
-	localParticipant *livekit.ParticipantInfo
+	conn               *websocket.Conn
+	PeerConn           *webrtc.PeerConnection
+	localTracks        []webrtc.TrackLocal
+	lock               sync.Mutex
+	ctx                context.Context
+	cancel             context.CancelFunc
+	connected          bool
+	iceConnected       bool
+	paused             bool
+	me                 *webrtc.MediaEngine // optional, populated only when receiving tracks
+	subscribedTracks   map[string]*webrtc.TrackRemote
+	localParticipant   *livekit.ParticipantInfo
+	remoteParticipants map[string]*livekit.ParticipantInfo
 
 	// pending actions to start after connected to peer
 	pendingCandidates   []*webrtc.ICECandidate
@@ -61,6 +65,17 @@ var (
 	}
 )
 
+func NewWebSocketConn(host, token string) (*websocket.Conn, error) {
+	u, err := url.Parse(host + "/rtc")
+	if err != nil {
+		return nil, err
+	}
+	requestHeader := make(http.Header)
+	requestHeader.Set("Authorization", "Bearer "+token)
+	conn, _, err := websocket.DefaultDialer.Dial(u.String(), requestHeader)
+	return conn, err
+}
+
 func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) {
 	// Create a new RTCPeerConnection
 	peerConn, err := webrtc.NewPeerConnection(rtcConf)
@@ -70,14 +85,16 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) {
 
 	logRing := ring.New(maxLogs)
 	c := &RTCClient{
-		conn:              conn,
-		lock:              sync.Mutex{},
-		pendingCandidates: make([]*webrtc.ICECandidate, 0),
-		localTracks:       make([]webrtc.TrackLocal, 0),
-		reader:            logRing,
-		writer:            logRing,
-		PeerConn:          peerConn,
-		me:                &webrtc.MediaEngine{},
+		conn:               conn,
+		lock:               sync.Mutex{},
+		pendingCandidates:  make([]*webrtc.ICECandidate, 0),
+		localTracks:        make([]webrtc.TrackLocal, 0),
+		subscribedTracks:   make(map[string]*webrtc.TrackRemote),
+		remoteParticipants: make(map[string]*livekit.ParticipantInfo),
+		reader:             logRing,
+		writer:             logRing,
+		PeerConn:           peerConn,
+		me:                 &webrtc.MediaEngine{},
 	}
 	c.ctx, c.cancel = context.WithCancel(context.Background())
 	c.me.RegisterDefaultCodecs()
@@ -102,7 +119,7 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) {
 
 	peerConn.OnTrack(func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
 		c.AppendLog("track received", "label", track.StreamID(), "id", track.ID())
-		go c.consumeReceiver(track)
+		go c.processTrack(track)
 	})
 
 	peerConn.OnNegotiationNeeded(func() {
@@ -168,6 +185,12 @@ func (c *RTCClient) Run() error {
 		}
 		switch msg := res.Message.(type) {
 		case *livekit.SignalResponse_Join:
+			c.lock.Lock()
+			for _, p := range msg.Join.OtherParticipants {
+				c.remoteParticipants[p.Sid] = p
+			}
+			c.lock.Unlock()
+
 			c.AppendLog("join accepted, sending offer..", "participant", msg.Join.Participant.Sid)
 			c.localParticipant = msg.Join.Participant
 			c.AppendLog("other participants", "count", len(msg.Join.OtherParticipants))
@@ -216,9 +239,12 @@ func (c *RTCClient) Run() error {
 				return err
 			}
 		case *livekit.SignalResponse_Update:
+			c.lock.Lock()
 			for _, p := range msg.Update.Participants {
+				c.remoteParticipants[p.Sid] = p
 				c.AppendLog("participant update", "id", p.Sid, "state", p.State.String())
 			}
+			c.lock.Unlock()
 		}
 	}
 
@@ -259,6 +285,14 @@ func (c *RTCClient) ReadResponse() (*livekit.SignalResponse, error) {
 	}
 }
 
+func (c *RTCClient) SubscribedTracks() map[string]*webrtc.TrackRemote {
+	return c.subscribedTracks
+}
+
+func (c *RTCClient) RemoteParticipants() []*livekit.ParticipantInfo {
+	return funk.Values(c.remoteParticipants).([]*livekit.ParticipantInfo)
+}
+
 func (c *RTCClient) Stop() {
 	c.conn.Close()
 	c.cancel()
@@ -272,12 +306,6 @@ func (c *RTCClient) ResumeLogs() {
 	c.paused = false
 }
 
-func (c *RTCClient) Receivers() []*rtc.ReceiverImpl {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-	return append([]*rtc.ReceiverImpl{}, c.receivers...)
-}
-
 func (c *RTCClient) SendRequest(msg *livekit.SignalRequest) error {
 	payload, err := protojson.Marshal(msg)
 	if err != nil {
@@ -471,7 +499,7 @@ func (c *RTCClient) logLoop() {
 	}
 }
 
-func (c *RTCClient) consumeReceiver(track *webrtc.TrackRemote) {
+func (c *RTCClient) processTrack(track *webrtc.TrackRemote) {
 	lastUpdate := time.Time{}
 	peerId, trackId := rtc.UnpackTrackId(track.ID())
 	numBytes := 0

cmd/cli/commands/rtc.go

@@ -4,14 +4,11 @@ import (
 	"bufio"
 	"errors"
 	"fmt"
-	"net/http"
-	"net/url"
 	"os"
 	"os/signal"
 	"path/filepath"
 	"syscall"
 
-	"github.com/gorilla/websocket"
 	"github.com/manifoldco/promptui"
 	"github.com/pion/webrtc/v3"
 	"github.com/urfave/cli/v2"
@@ -53,11 +50,6 @@ var (
 )
 
 func joinRoom(c *cli.Context) error {
-	u, err := url.Parse(c.String("host") + "/rtc")
-	if err != nil {
-		return err
-	}
-
 	name := c.String("name")
 	roomId := c.String("room")
 	token := c.String("token")
@@ -72,6 +64,7 @@ func joinRoom(c *cli.Context) error {
 			return fmt.Errorf("--name is required")
 		}
 		// token may be nil in dev mode
+		var err error
 		token, err = accessToken(c, &auth.VideoGrant{
 			RoomJoin: true,
 			Room:     roomId,
@@ -82,21 +75,10 @@ func joinRoom(c *cli.Context) error {
 	}
 
 	log := logger.GetLogger()
-	var requestHeader http.Header
-	if token != "" {
-		// set this as a header
-		requestHeader = make(http.Header)
-		requestHeader.Set("Authorization", "Bearer "+token)
-	} else {
-		// dev mode, will pass these as overrides
-		v := url.Values{}
-		v.Set("room", roomId)
-		v.Set("name", name)
-		u.RawQuery = v.Encode()
-	}
 
-	log.Infow("connecting to Websocket signal", "url", u.String())
-	conn, _, err := websocket.DefaultDialer.Dial(u.String(), requestHeader)
+	host := c.String("host")
+	log.Infow("connecting to Websocket signal", "host", host)
+	conn, err := client.NewWebSocketConn(host, token)
 	if err != nil {
 		return err
 	}

cmd/server/main.go

@@ -2,31 +2,23 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
 	"math/rand"
-	"net"
-	"net/http"
 	"os"
 	"os/signal"
 	"runtime"
 	"runtime/pprof"
-	"sync"
 	"syscall"
 	"time"
 
 	"github.com/urfave/cli/v2"
-	"github.com/urfave/negroni"
 
 	"github.com/livekit/livekit-server/pkg/auth"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/logger"
-	"github.com/livekit/livekit-server/pkg/node"
-	"github.com/livekit/livekit-server/pkg/rtc"
 	"github.com/livekit/livekit-server/pkg/service"
 	"github.com/livekit/livekit-server/pkg/utils"
-	"github.com/livekit/livekit-server/proto/livekit"
 )
 
 func main() {
@@ -127,7 +119,7 @@ func startServer(c *cli.Context) error {
 	service.AuthRequired = true
 	logger.GetLogger().Infow("auth enabled", "num_keys", keyProvider.NumKeys())
 
-	server, err := InitializeServer(conf, keyProvider)
+	server, err := service.InitializeServer(conf, keyProvider)
 	if err != nil {
 		return err
 	}
@@ -168,113 +160,6 @@ func createKeyProvider(keyFile, keys string) (auth.KeyProvider, error) {
 	return nil, errors.New("one of key-file or keys must be provided in order to support a secure installation")
 }
 
-type LivekitServer struct {
-	config     *config.Config
-	roomServer livekit.TwirpServer
-	rtcService *service.RTCService
-	roomHttp   *http.Server
-	rtcHttp    *http.Server
-	running    bool
-	doneChan   chan bool
-}
-
-func NewLivekitServer(conf *config.Config,
-	roomService livekit.RoomService,
-	rtcService *service.RTCService,
-	keyProvider auth.KeyProvider) (s *LivekitServer, err error) {
-	s = &LivekitServer{
-		config:     conf,
-		roomServer: livekit.NewRoomServiceServer(roomService),
-		rtcService: rtcService,
-	}
-
-	middlewares := make([]negroni.Handler, 0)
-	if keyProvider != nil {
-		middlewares = append(middlewares, service.NewAPIKeyAuthMiddleware(keyProvider))
-	}
-
-	s.roomHttp = &http.Server{
-		Addr:    fmt.Sprintf(":%d", conf.APIPort),
-		Handler: configureMiddlewares(conf, s.roomServer, middlewares...),
-	}
-
-	rtcHandler := http.NewServeMux()
-	rtcHandler.Handle("/rtc", rtcService)
-	s.rtcHttp = &http.Server{
-		Addr:    fmt.Sprintf(":%d", conf.RTCPort),
-		Handler: configureMiddlewares(conf, rtcHandler, middlewares...),
-	}
-
-	return
-}
-
-func (s *LivekitServer) Start() error {
-	if s.running {
-		return errors.New("already running")
-	}
-	s.running = true
-	s.doneChan = make(chan bool, 1)
-
-	// ensure we could listen
-	roomLn, err := net.Listen("tcp", s.roomHttp.Addr)
-	if err != nil {
-		return err
-	}
-
-	rtcAddr := fmt.Sprintf(":%d", s.config.RTCPort)
-	rtcLn, err := net.Listen("tcp", rtcAddr)
-	if err != nil {
-		return err
-	}
-
-	go func() {
-		logger.GetLogger().Infow("starting Room service", "address", s.roomHttp.Addr)
-		s.roomHttp.Serve(roomLn)
-	}()
-	go func() {
-		logger.GetLogger().Infow("starting RTC service", "address", rtcAddr)
-		s.rtcHttp.Serve(rtcLn)
-	}()
-
-	<-s.doneChan
-
-	// wait for shutdown
-	ctx, _ := context.WithTimeout(context.Background(), time.Second*5)
-	wg := sync.WaitGroup{}
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		s.rtcHttp.Shutdown(ctx)
-	}()
-	go func() {
-		defer wg.Done()
-		s.roomHttp.Shutdown(ctx)
-	}()
-	wg.Wait()
-
-	return nil
-}
-
-func (s *LivekitServer) Stop() {
-	s.running = false
-
-	s.doneChan <- true
-}
-
-func configureMiddlewares(conf *config.Config, handler http.Handler, middlewares ...negroni.Handler) *negroni.Negroni {
-	n := negroni.New()
-	n.Use(negroni.NewRecovery())
-	for _, m := range middlewares {
-		n.Use(m)
-	}
-	n.UseHandler(handler)
-	return n
-}
-
-func newManager(conf *config.Config, localNode *node.Node) (*rtc.RoomManager, error) {
-	return rtc.NewRoomManager(conf.RTC, localNode.Ip)
-}
-
 func generateKeys(c *cli.Context) error {
 	apiKey := utils.NewGuid(utils.APIKeyPrefix)
 	secret := utils.RandomSecret()

magefile.go

@@ -29,7 +29,7 @@ var checksummer = NewChecksummer(".", goChecksumFile, ".go")
 
 func init() {
 	checksummer.IgnoredPaths = []string{
-		"cmd/server/wire_gen.go",
+		"pkg/service/wire_gen.go",
 		"pkg/rtc/types/typesfakes",
 	}
 }
@@ -163,7 +163,16 @@ func generateCmd() error {
 
 	cmd := exec.Command("go", "generate", "./cmd/...")
 	connectStd(cmd)
-	return cmd.Run()
+	if err := cmd.Run(); err != nil {
+		return err
+	}
+
+	cmd = exec.Command("go", "generate", "./pkg/service/...")
+	connectStd(cmd)
+	if err := cmd.Run(); err != nil {
+		return err
+	}
+	return nil
 }
 
 // implicitly install deps

pkg/service/service.go

@@ -1,11 +1,20 @@
 package service
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"net"
+	"net/http"
+	"sync"
+	"time"
 
 	"github.com/google/wire"
+	"github.com/urfave/negroni"
 
+	"github.com/livekit/livekit-server/pkg/auth"
 	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/logger"
 	"github.com/livekit/livekit-server/pkg/node"
 	"github.com/livekit/livekit-server/pkg/rtc"
 	"github.com/livekit/livekit-server/proto/livekit"
@@ -14,6 +23,8 @@ import (
 var ServiceSet = wire.NewSet(
 	NewRoomService,
 	NewRTCService,
+	NewLivekitServer,
+	newRoomManagerWithNode,
 )
 
 func NewRoomService(conf *config.Config, manager *rtc.RoomManager, localNode *node.Node) (livekit.RoomService, error) {
@@ -23,3 +34,114 @@ func NewRoomService(conf *config.Config, manager *rtc.RoomManager, localNode *no
 		return NewSimpleRoomService(manager, localNode)
 	}
 }
+
+type LivekitServer struct {
+	config     *config.Config
+	roomServer livekit.TwirpServer
+	rtcService *RTCService
+	roomHttp   *http.Server
+	rtcHttp    *http.Server
+	running    bool
+	doneChan   chan bool
+}
+
+func newRoomManagerWithNode(conf *config.Config, localNode *node.Node) (*rtc.RoomManager, error) {
+	return rtc.NewRoomManager(conf.RTC, localNode.Ip)
+}
+
+func NewLivekitServer(conf *config.Config,
+	roomService livekit.RoomService,
+	rtcService *RTCService,
+	keyProvider auth.KeyProvider) (s *LivekitServer, err error) {
+	s = &LivekitServer{
+		config:     conf,
+		roomServer: livekit.NewRoomServiceServer(roomService),
+		rtcService: rtcService,
+	}
+
+	middlewares := make([]negroni.Handler, 0)
+	if keyProvider != nil {
+		middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider))
+	}
+
+	s.roomHttp = &http.Server{
+		Addr:    fmt.Sprintf(":%d", conf.APIPort),
+		Handler: configureMiddlewares(s.roomServer, middlewares...),
+	}
+
+	rtcHandler := http.NewServeMux()
+	rtcHandler.Handle("/rtc", rtcService)
+	s.rtcHttp = &http.Server{
+		Addr:    fmt.Sprintf(":%d", conf.RTCPort),
+		Handler: configureMiddlewares(rtcHandler, middlewares...),
+	}
+
+	return
+}
+
+func (s *LivekitServer) IsRunning() bool {
+	return s.running
+}
+
+func (s *LivekitServer) Start() error {
+	if s.running {
+		return errors.New("already running")
+	}
+	s.doneChan = make(chan bool, 1)
+
+	// ensure we could listen
+	roomLn, err := net.Listen("tcp", s.roomHttp.Addr)
+	if err != nil {
+		return err
+	}
+
+	rtcAddr := fmt.Sprintf(":%d", s.config.RTCPort)
+	rtcLn, err := net.Listen("tcp", rtcAddr)
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		logger.GetLogger().Infow("starting Room service", "address", s.roomHttp.Addr)
+		s.roomHttp.Serve(roomLn)
+	}()
+	go func() {
+		logger.GetLogger().Infow("starting RTC service", "address", rtcAddr)
+		s.rtcHttp.Serve(rtcLn)
+	}()
+
+	s.running = true
+
+	<-s.doneChan
+
+	// wait for shutdown
+	ctx, _ := context.WithTimeout(context.Background(), time.Second*5)
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		s.rtcHttp.Shutdown(ctx)
+	}()
+	go func() {
+		defer wg.Done()
+		s.roomHttp.Shutdown(ctx)
+	}()
+	wg.Wait()
+
+	return nil
+}
+
+func (s *LivekitServer) Stop() {
+	s.running = false
+	s.doneChan <- true
+}
+
+func configureMiddlewares(handler http.Handler, middlewares ...negroni.Handler) *negroni.Negroni {
+	n := negroni.New()
+	n.Use(negroni.NewRecovery())
+	for _, m := range middlewares {
+		n.Use(m)
+	}
+	n.UseHandler(handler)
+	return n
+}

pkg/service/wire.go

@@ -1,6 +1,6 @@
 //+build wireinject
 
-package main
+package service
 
 import (
 	"github.com/google/wire"
@@ -8,15 +8,12 @@ import (
 	"github.com/livekit/livekit-server/pkg/auth"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/node"
-	"github.com/livekit/livekit-server/pkg/service"
 )
 
 func InitializeServer(conf *config.Config, keyProvider auth.KeyProvider) (*LivekitServer, error) {
 	wire.Build(
-		NewLivekitServer,
 		node.NodeSet,
-		service.ServiceSet,
-		newManager,
+		ServiceSet,
 	)
 	return &LivekitServer{}, nil
 }

pkg/service/wire_gen.go

@@ -3,13 +3,12 @@
 //go:generate wire
 //+build !wireinject
 
-package main
+package service
 
 import (
 	"github.com/livekit/livekit-server/pkg/auth"
 	"github.com/livekit/livekit-server/pkg/config"
 	"github.com/livekit/livekit-server/pkg/node"
-	"github.com/livekit/livekit-server/pkg/service"
 )
 
 // Injectors from wire.go:
@@ -19,15 +18,15 @@ func InitializeServer(conf *config.Config, keyProvider auth.KeyProvider) (*Livek
 	if err != nil {
 		return nil, err
 	}
-	roomManager, err := newManager(conf, nodeNode)
+	roomManager, err := newRoomManagerWithNode(conf, nodeNode)
 	if err != nil {
 		return nil, err
 	}
-	roomService, err := service.NewRoomService(conf, roomManager, nodeNode)
+	roomService, err := NewRoomService(conf, roomManager, nodeNode)
 	if err != nil {
 		return nil, err
 	}
-	rtcService := service.NewRTCService(conf, roomManager)
+	rtcService := NewRTCService(conf, roomManager)
 	livekitServer, err := NewLivekitServer(conf, roomService, rtcService, keyProvider)
 	if err != nil {
 		return nil, err

test/integration_helpers.go

@@ -0,0 +1,84 @@
+package test
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/livekit/livekit-server/cmd/cli/client"
+	"github.com/livekit/livekit-server/pkg/auth"
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/service"
+)
+
+const (
+	testApiKey    = "apikey"
+	testApiSecret = "apiSecret"
+)
+
+func waitForServerToStart(s *service.LivekitServer) {
+	// wait till ready
+	ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
+	for {
+		select {
+		case <-ctx.Done():
+			panic("could not start server after timeout")
+		case <-time.After(10 * time.Millisecond):
+			if s.IsRunning() {
+				return
+			}
+		}
+	}
+}
+
+func createServer() *service.LivekitServer {
+	conf, err := config.NewConfig("")
+	if err != nil {
+		panic(fmt.Sprintf("could not create config: %v", err))
+	}
+	s, err := service.InitializeServer(conf, &StaticKeyProvider{})
+	if err != nil {
+		panic(fmt.Sprintf("could not create server: %v", err))
+	}
+	return s
+}
+
+func createClient(room, name string) *client.RTCClient {
+	return nil
+}
+
+func joinToken(room, name string) string {
+	at := auth.NewAccessToken(testApiKey, testApiSecret).
+		AddGrant(&auth.VideoGrant{RoomJoin: true, Room: room}).
+		SetIdentity(name)
+	t, err := at.ToJWT()
+	if err != nil {
+		panic(err)
+	}
+	return t
+}
+
+func createRoomToken() string {
+	at := auth.NewAccessToken(testApiKey, testApiSecret).
+		AddGrant(&auth.VideoGrant{RoomCreate: true}).
+		SetIdentity("testuser")
+	t, err := at.ToJWT()
+	if err != nil {
+		panic(err)
+	}
+	return t
+}
+
+type StaticKeyProvider struct {
+}
+
+func (p *StaticKeyProvider) NumKeys() int {
+	return 1
+}
+
+func (p *StaticKeyProvider) GetSecret(key string) string {
+	if key == testApiKey {
+		return testApiSecret
+	}
+	return ""
+}

test/integration_test.go

@@ -0,0 +1,26 @@
+package test
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestScenarioDefault(t *testing.T) {
+	assert.True(t, true)
+}
+
+func TestMain(m *testing.M) {
+	s := createServer()
+	go func() {
+		s.Start()
+	}()
+
+	waitForServerToStart(s)
+
+	code := m.Run()
+
+	s.Stop()
+	os.Exit(code)
+}