diff --git a/cmd/cli/client/client.go b/cmd/cli/client/client.go index ff4b8b918..f57e27c43 100644 --- a/cmd/cli/client/client.go +++ b/cmd/cli/client/client.go @@ -72,11 +72,15 @@ func NewWebSocketConn(host, token string) (*websocket.Conn, error) { return nil, err } requestHeader := make(http.Header) - requestHeader.Set("Authorization", "Bearer "+token) + SetAuthorizationToken(requestHeader, token) conn, _, err := websocket.DefaultDialer.Dial(u.String(), requestHeader) return conn, err } +func SetAuthorizationToken(header http.Header, token string) { + header.Set("Authorization", "Bearer "+token) +} + func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) { // Create a new RTCPeerConnection peerConn, err := webrtc.NewPeerConnection(rtcConf) diff --git a/cmd/cli/commands/room.go b/cmd/cli/commands/room.go index 00efac814..5c7256308 100644 --- a/cmd/cli/commands/room.go +++ b/cmd/cli/commands/room.go @@ -8,6 +8,7 @@ import ( "github.com/twitchtv/twirp" "github.com/urfave/cli/v2" + "github.com/livekit/livekit-server/cmd/cli/client" "github.com/livekit/livekit-server/pkg/auth" "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/proto/livekit" @@ -109,7 +110,7 @@ func contextWithAccessToken(c *cli.Context, grant *auth.VideoGrant) context.Cont } if token != "" { header := make(http.Header) - header.Set("Authorization", "Bearer "+token) + client.SetAuthorizationToken(header, token) if tctx, err := twirp.WithHTTPRequestHeaders(ctx, header); err == nil { logger.Debugw("requesting with token") ctx = tctx diff --git a/cmd/server/main.go b/cmd/server/main.go index 5a3b1b064..0ed1fed70 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -2,7 +2,6 @@ package main import ( "context" - "errors" "fmt" "io/ioutil" "math/rand" @@ -14,6 +13,7 @@ import ( "time" "github.com/go-redis/redis/v8" + "github.com/pkg/errors" "github.com/urfave/cli/v2" "github.com/livekit/livekit-server/pkg/auth" @@ -163,11 +163,13 @@ func startServer(c *cli.Context) error { func createRouterAndStore(config *config.Config, node routing.LocalNode) (router routing.Router, store service.RoomStore, err error) { if config.MultiNode { + logger.Infow("using multi-node routing via redis", "address", config.Redis.Address) rc := redis.NewClient(&redis.Options{ Addr: config.Redis.Address, Password: config.Redis.Password, }) if err = rc.Ping(context.Background()).Err(); err != nil { + err = errors.Wrap(err, "unable to connect to redis") return } @@ -175,6 +177,7 @@ func createRouterAndStore(config *config.Config, node routing.LocalNode) (router store = service.NewRedisRoomStore(rc) } else { // local routing and store + logger.Infow("using single-node routing") router = routing.NewLocalRouter(node) store = service.NewLocalRoomStore() } diff --git a/proto/livekit/rtc.pb.go b/proto/livekit/rtc.pb.go index b3f723961..9cb2169d6 100644 --- a/proto/livekit/rtc.pb.go +++ b/proto/livekit/rtc.pb.go @@ -126,7 +126,7 @@ type isSignalRequest_Message interface { } type SignalRequest_Offer struct { - // participant joining initially + // participant joining initially, and during negotiations Offer *SessionDescription `protobuf:"bytes,1,opt,name=offer,proto3,oneof"` } diff --git a/test/integration_helpers.go b/test/integration_helpers.go index 3c20f9da1..0ab44e18a 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -8,11 +8,14 @@ import ( "testing" "time" + "github.com/go-redis/redis/v8" "github.com/stretchr/testify/assert" + "github.com/twitchtv/twirp" "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/logger" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/service" "github.com/livekit/livekit-server/proto/livekit" @@ -29,6 +32,63 @@ var ( roomClient livekit.RoomService ) +func setupSingleNodeTest(roomName string) *service.LivekitServer { + logger.InitDevelopment("") + s := createSingleNodeServer() + go func() { + s.Start() + }() + + waitForServerToStart(s) + + // create test room + header := make(http.Header) + client.SetAuthorizationToken(header, createRoomToken()) + tctx, err := twirp.WithHTTPRequestHeaders(context.Background(), header) + if err != nil { + panic(err) + } + _, err = roomClient.CreateRoom(tctx, &livekit.CreateRoomRequest{Name: roomName}) + if err != nil { + panic(err) + } + return s +} + +func setupMultiNodeTest(roomName string) *service.LivekitServer { + logger.InitDevelopment("") + s := createMultiNodeServer() + go func() { + s.Start() + }() + + waitForServerToStart(s) + + // create test room + header := make(http.Header) + client.SetAuthorizationToken(header, createRoomToken()) + tctx, err := twirp.WithHTTPRequestHeaders(context.Background(), header) + if err != nil { + panic(err) + } + _, err = roomClient.CreateRoom(tctx, &livekit.CreateRoomRequest{Name: roomName}) + if err != nil { + panic(err) + } + return s +} + +func teardownTest(s *service.LivekitServer, roomName string) { + header := make(http.Header) + client.SetAuthorizationToken(header, createRoomToken()) + tctx, err := twirp.WithHTTPRequestHeaders(context.Background(), header) + if err != nil { + panic(err) + } + roomClient.DeleteRoom(tctx, &livekit.DeleteRoomRequest{Room: roomName}) + s.Stop() +} + func waitForServerToStart(s *service.LivekitServer) { // wait till ready ctx, _ := context.WithTimeout(context.Background(), 10*time.Second) @@ -71,7 +131,7 @@ func waitUntilConnected(t *testing.T, clients ...*client.RTCClient) { wg.Wait() } -func createServer() *service.LivekitServer { +func createSingleNodeServer() *service.LivekitServer { var err error serverConfig, err = config.NewConfig("") if err != nil { @@ -95,8 +155,42 @@ func createServer() *service.LivekitServer { return s } +func createMultiNodeServer() *service.LivekitServer { + var err error + serverConfig, err = config.NewConfig("") + if err != nil { + panic(fmt.Sprintf("could not create config: %v", err)) + } + serverConfig.MultiNode = true + serverConfig.Redis.Address = "localhost:6379" + + currentNode, err := routing.NewLocalNode(serverConfig) + if err != nil { + panic(err) + } + + // local routing and store + rc := redis.NewClient(&redis.Options{ + Addr: serverConfig.Redis.Address, + Password: serverConfig.Redis.Password, + }) + if err = rc.Ping(context.Background()).Err(); err != nil { + panic(err) + } + + router := routing.NewRedisRouter(currentNode, rc, false) + roomStore := service.NewRedisRoomStore(rc) + s, err := service.InitializeServer(serverConfig, &StaticKeyProvider{}, roomStore, router, currentNode) + if err != nil { + panic(fmt.Sprintf("could not create server: %v", err)) + } + + roomClient = livekit.NewRoomServiceJSONClient(fmt.Sprintf("http://localhost:%d", serverConfig.Port), &http.Client{}) + return s +} + // creates a client and runs against server -func createClient(name string) *client.RTCClient { +func createRTCClient(name string) *client.RTCClient { token := joinToken(testRoom, name) ws, err := client.NewWebSocketConn(fmt.Sprintf("ws://localhost:%d", serverConfig.Port), token) if err != nil { diff --git a/test/integration_test.go b/test/integration_test.go index 9d70a7ee5..30d04a296 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -1,21 +1,15 @@ package test import ( - "context" - "net/http" "os" "testing" "github.com/stretchr/testify/assert" - "github.com/twitchtv/twirp" - - "github.com/livekit/livekit-server/pkg/logger" - "github.com/livekit/livekit-server/proto/livekit" ) func TestClientCouldConnect(t *testing.T) { - c1 := createClient("c1") - c2 := createClient("c2") + c1 := createRTCClient("c1") + c2 := createRTCClient("c2") waitUntilConnected(t, c1, c2) // ensure they both see each other @@ -29,8 +23,8 @@ func TestClientCouldConnect(t *testing.T) { } func TestSinglePublisher(t *testing.T) { - c1 := createClient("c1") - c2 := createClient("c2") + c1 := createRTCClient("c1") + c2 := createRTCClient("c2") waitUntilConnected(t, c1, c2) // publish a track and ensure clients receive it ok @@ -42,7 +36,7 @@ func TestSinglePublisher(t *testing.T) { defer t2.Stop() // a new client joins and should get the initial stream - c3 := createClient("c3") + c3 := createRTCClient("c3") withTimeout(t, "c2 should receive two tracks", func() bool { if len(c2.SubscribedTracks()) == 0 { @@ -73,31 +67,10 @@ func TestSinglePublisher(t *testing.T) { } func TestMain(m *testing.M) { - logger.InitDevelopment("") - s := createServer() - go func() { - s.Start() - }() - - waitForServerToStart(s) - - // create test room - token := createRoomToken() - header := make(http.Header) - logger.Debugw("auth token", "token", token) - header.Set("Authorization", "Bearer "+token) - tctx, err := twirp.WithHTTPRequestHeaders(context.Background(), header) - if err != nil { - panic(err) - } - _, err = roomClient.CreateRoom(tctx, &livekit.CreateRoomRequest{Name: testRoom}) - if err != nil { - panic(err) - } + s := setupSingleNodeTest(testRoom) code := m.Run() - roomClient.DeleteRoom(tctx, &livekit.DeleteRoomRequest{Room: testRoom}) - s.Stop() + teardownTest(s, testRoom) os.Exit(code) }