refactor bearer token on the client side

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)

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

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()
 	}

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"`
 }
 

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 {

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)
 }