diff --git a/clients.go b/clients.go deleted file mode 100644 index 06ab7d0f9..000000000 --- a/clients.go +++ /dev/null @@ -1 +0,0 @@ -package main diff --git a/cmd/cli/client/client.go b/cmd/cli/client/client.go index aeb88c0d4..0aef06ae5 100644 --- a/cmd/cli/client/client.go +++ b/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 diff --git a/cmd/cli/commands/rtc.go b/cmd/cli/commands/rtc.go index 2b24737fe..41ba793d8 100644 --- a/cmd/cli/commands/rtc.go +++ b/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 } diff --git a/cmd/server/main.go b/cmd/server/main.go index 6fe79593c..4ca01eb7a 100644 --- a/cmd/server/main.go +++ b/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() diff --git a/magefile.go b/magefile.go index fb1de8124..488ad4491 100644 --- a/magefile.go +++ b/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 diff --git a/pkg/service/service.go b/pkg/service/service.go index 948b6bebf..e97a34820 100644 --- a/pkg/service/service.go +++ b/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 +} diff --git a/cmd/server/wire.go b/pkg/service/wire.go similarity index 75% rename from cmd/server/wire.go rename to pkg/service/wire.go index eac024958..c219275e1 100644 --- a/cmd/server/wire.go +++ b/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 } diff --git a/cmd/server/wire_gen.go b/pkg/service/wire_gen.go similarity index 73% rename from cmd/server/wire_gen.go rename to pkg/service/wire_gen.go index d195fd11c..f9761d13b 100644 --- a/cmd/server/wire_gen.go +++ b/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 diff --git a/test/integration_helpers.go b/test/integration_helpers.go new file mode 100644 index 000000000..a923c96b7 --- /dev/null +++ b/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 "" +} diff --git a/test/integration_test.go b/test/integration_test.go new file mode 100644 index 000000000..1f1776479 --- /dev/null +++ b/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) +}