AV1 Support

README.md

@@ -21,7 +21,7 @@ Version 2 of the <platform> SDK contains a small set of breaking changes. Read t
 ## Installation
 
 ```shell
-go get github.com/livekit/server-sdk-go/v2
+go get github.com/tryiris-ai/livekit-server-sdk-go/v2
 ```
 
 Note: since v1.0 release, this package requires Go 1.18+ in order to build.
@@ -32,7 +32,7 @@ Note: since v1.0 release, this package requires Go 1.18+ in order to build.
 import (
 	"time"
 
-	lksdk "github.com/livekit/server-sdk-go/v2"
+	lksdk "github.com/tryiris-ai/livekit-server-sdk-go/v2"
 	"github.com/livekit/protocol/auth"
 )
 
@@ -56,7 +56,7 @@ RoomService gives you complete control over rooms and participants within them.
 
 ```go
 import (
-	lksdk "github.com/livekit/server-sdk-go/v2"
+	lksdk "github.com/tryiris-ai/livekit-server-sdk-go/v2"
 	livekit "github.com/livekit/protocol/livekit"
 )
 
@@ -110,7 +110,7 @@ The Real-time SDK gives you access programmatic access as a client enabling you
 
 ```go
 import (
-  lksdk "github.com/livekit/server-sdk-go/v2"
+  lksdk "github.com/tryiris-ai/livekit-server-sdk-go/v2"
 )
 
 func main() {
@@ -210,7 +210,8 @@ if _, err = room.LocalParticipant.PublishTrack(track, &lksdk.TrackPublicationOpt
 }
 ```
 
-For a full working example, refer to [join.go](https://github.com/livekit/livekit-cli/blob/main/cmd/livekit-cli/join.go) in livekit-cli.
+For a full working example, refer to [filesender](https://github.com/livekit/server-sdk-go/blob/main/examples/filesender/main.go). This
+example sends all audio/video files in the current directory.
 
 ### Publish from other sources
 
@@ -247,6 +248,14 @@ room, err := lksdk.ConnectToRoom(hostURL, lksdk.ConnectInfo{
 }, lksdk.WithPacer(pf))
 ```
 
+## Receiving tracks from Room
+
+With the Go SDK, you can accept media from the room.
+
+For a full working example, refer to [filesaver](https://github.com/livekit/server-sdk-go/blob/main/examples/filesaver/main.go). This
+example saves the audio/video in the LiveKit room to the local disk.
+
+
 ## Receiving webhooks
 
 The Go SDK helps you to verify and decode webhook callbacks to ensure their authenticity.

callback.go

@@ -94,8 +94,17 @@ func (cb *ParticipantCallback) Merge(other *ParticipantCallback) {
 	}
 }
 
+type DisconnectionReason string
+
+const (
+	LeaveRequested    DisconnectionReason = "leave requested by room"
+	Failed            DisconnectionReason = "connection to room failed"
+	NegotiationFailed DisconnectionReason = "negotiation failed"
+)
+
 type RoomCallback struct {
 	OnDisconnected            func()
+	OnDisconnectedWithReason  func(reason DisconnectionReason)
 	OnParticipantConnected    func(*RemoteParticipant)
 	OnParticipantDisconnected func(*RemoteParticipant)
 	OnActiveSpeakersChanged   func([]Participant)
@@ -113,6 +122,7 @@ func NewRoomCallback() *RoomCallback {
 		ParticipantCallback: *pc,
 
 		OnDisconnected:            func() {},
+		OnDisconnectedWithReason:  func(reason DisconnectionReason) {},
 		OnParticipantConnected:    func(participant *RemoteParticipant) {},
 		OnParticipantDisconnected: func(participant *RemoteParticipant) {},
 		OnActiveSpeakersChanged:   func(participants []Participant) {},
@@ -130,6 +140,9 @@ func (cb *RoomCallback) Merge(other *RoomCallback) {
 	if other.OnDisconnected != nil {
 		cb.OnDisconnected = other.OnDisconnected
 	}
+	if other.OnDisconnectedWithReason != nil {
+		cb.OnDisconnectedWithReason = other.OnDisconnectedWithReason
+	}
 	if other.OnParticipantConnected != nil {
 		cb.OnParticipantConnected = other.OnParticipantConnected
 	}

engine.go

@@ -18,6 +18,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pion/dtls/v2"
 	"github.com/pion/webrtc/v3"
 	"go.uber.org/atomic"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -36,22 +37,23 @@ const (
 )
 
 type RTCEngine struct {
-	pclock                sync.Mutex
-	publisher             *PCTransport
-	subscriber            *PCTransport
-	client                *SignalClient
-	dclock                sync.RWMutex
-	reliableDC            *webrtc.DataChannel
-	lossyDC               *webrtc.DataChannel
-	reliableDCSub         *webrtc.DataChannel
-	lossyDCSub            *webrtc.DataChannel
-	trackPublishedChan    chan *livekit.TrackPublishedResponse
-	subscriberPrimary     bool
-	hasConnected          atomic.Bool
-	hasPublish            atomic.Bool
-	closed                atomic.Bool
-	reconnecting          atomic.Bool
-	requiresFullReconnect atomic.Bool
+	pclock                 sync.Mutex
+	publisher              *PCTransport
+	subscriber             *PCTransport
+	client                 *SignalClient
+	dclock                 sync.RWMutex
+	reliableDC             *webrtc.DataChannel
+	lossyDC                *webrtc.DataChannel
+	reliableDCSub          *webrtc.DataChannel
+	lossyDCSub             *webrtc.DataChannel
+	trackPublishedChan     chan *livekit.TrackPublishedResponse
+	subscriberPrimary      bool
+	hasConnected           atomic.Bool
+	hasPublish             atomic.Bool
+	closed                 atomic.Bool
+	reconnecting           atomic.Bool
+	requiresFullReconnect  atomic.Bool
+	srtpProtectionProfiles []dtls.SRTPProtectionProfile
 
 	url        string
 	token      atomic.String
@@ -60,7 +62,7 @@ type RTCEngine struct {
 	JoinTimeout time.Duration
 
 	// callbacks
-	OnDisconnected      func()
+	OnDisconnected      func(reason DisconnectionReason)
 	OnMediaTrack        func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver)
 	OnParticipantUpdate func([]*livekit.ParticipantInfo)
 	OnSpeakersChanged   func([]*livekit.SpeakerInfo)
@@ -120,6 +122,7 @@ func (e *RTCEngine) Join(url string, token string, params *connectParams) (*live
 	e.url = url
 	e.token.Store(token)
 	e.connParams = params
+	e.srtpProtectionProfiles = params.SRTPProtectionProfiles
 
 	err = e.configure(res.IceServers, res.ClientConfiguration, proto.Bool(res.SubscriberPrimary))
 	if err != nil {
@@ -205,7 +208,10 @@ func (e *RTCEngine) configure(
 	clientConfig *livekit.ClientConfiguration,
 	subscriberPrimary *bool) error {
 	rtcICEServers := FromProtoIceServers(iceServers)
-	configuration := webrtc.Configuration{ICEServers: rtcICEServers}
+	configuration := webrtc.Configuration{
+		ICEServers:         rtcICEServers,
+		ICETransportPolicy: e.connParams.ICETransportPolicy,
+	}
 	if clientConfig != nil &&
 		clientConfig.GetForceRelay() == livekit.ClientConfigSetting_ENABLED {
 		configuration.ICETransportPolicy = webrtc.ICETransportPolicyRelay
@@ -226,17 +232,21 @@ func (e *RTCEngine) configure(
 
 	var err error
 	if e.publisher, err = NewPCTransport(PCTransportParams{
-		Configuration:        configuration,
-		RetransmitBufferSize: e.connParams.RetransmitBufferSize,
-		Pacer:                e.connParams.Pacer,
-		OnRTTUpdate:          e.setRTT,
-		IsSender:             true,
+		Configuration:          configuration,
+		RetransmitBufferSize:   e.connParams.RetransmitBufferSize,
+		Pacer:                  e.connParams.Pacer,
+		Interceptors:           e.connParams.Interceptors,
+		OnRTTUpdate:            e.setRTT,
+		IsSender:               true,
+		OnNegotiationError:     e.handleNegiationError,
+		SRTPProtectionProfiles: e.srtpProtectionProfiles,
 	}); err != nil {
 		return err
 	}
 	if e.subscriber, err = NewPCTransport(PCTransportParams{
 		Configuration:        configuration,
 		RetransmitBufferSize: e.connParams.RetransmitBufferSize,
+		OnNegotiationError:   e.handleNegiationError,
 	}); err != nil {
 		return err
 	}
@@ -559,7 +569,7 @@ func (e *RTCEngine) handleDisconnect(fullReconnect bool) {
 		}
 
 		if e.OnDisconnected != nil {
-			e.OnDisconnected()
+			e.OnDisconnected(Failed)
 		}
 	}()
 }
@@ -584,9 +594,11 @@ func (e *RTCEngine) resumeConnection() error {
 	publisher := e.publisher
 	e.pclock.Unlock()
 	if sendOffer {
-		publisher.createAndSendOffer(&webrtc.OfferOptions{
+		if err := publisher.createAndSendOffer(&webrtc.OfferOptions{
 			ICERestart: true,
-		})
+		}); err != nil {
+			return err
+		}
 	}
 
 	if err = e.waitUntilConnected(); err != nil {
@@ -642,7 +654,15 @@ func (e *RTCEngine) handleLeave(leave *livekit.LeaveRequest) {
 			"canReconnect", leave.GetCanReconnect(),
 		)
 		if e.OnDisconnected != nil {
-			e.OnDisconnected()
+			e.OnDisconnected(LeaveRequested)
 		}
 	}
 }
+
+func (e *RTCEngine) handleNegiationError(err error) {
+	logger.Errorw("negotiation error", err)
+
+	if e.OnDisconnected != nil {
+		e.OnDisconnected(NegotiationFailed)
+	}
+}

examples/filesaver/main.go

@@ -31,8 +31,8 @@ import (
 	"github.com/pion/webrtc/v3/pkg/media/oggwriter"
 
 	"github.com/livekit/protocol/logger"
-	lksdk "github.com/livekit/server-sdk-go/v2"
-	"github.com/livekit/server-sdk-go/v2/pkg/samplebuilder"
+	lksdk "github.com/tryiris-ai/livekit-server-sdk-go/v2"
+	"github.com/tryiris-ai/livekit-server-sdk-go/v2/pkg/samplebuilder"
 )
 
 var (

examples/filesender/main.go

@@ -0,0 +1,97 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	lksdk "github.com/tryiris-ai/livekit-server-sdk-go/v2"
+)
+
+var (
+	host, apiKey, apiSecret, roomName, identity string
+)
+
+func init() {
+	flag.StringVar(&host, "host", "", "livekit server host")
+	flag.StringVar(&apiKey, "api-key", "", "livekit api key")
+	flag.StringVar(&apiSecret, "api-secret", "", "livekit api secret")
+	flag.StringVar(&roomName, "room-name", "", "room name")
+	flag.StringVar(&identity, "identity", "", "participant identity")
+}
+
+func main() {
+	flag.Parse()
+	if host == "" || apiKey == "" || apiSecret == "" || roomName == "" || identity == "" {
+		fmt.Println("invalid arguments.")
+		return
+	}
+	room, err := lksdk.ConnectToRoom(host, lksdk.ConnectInfo{
+		APIKey:              apiKey,
+		APISecret:           apiSecret,
+		RoomName:            roomName,
+		ParticipantIdentity: identity,
+	}, &lksdk.RoomCallback{}, lksdk.WithAutoSubscribe(false))
+	if err != nil {
+		panic(err)
+	}
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT)
+
+	files, err := os.ReadDir(".")
+	if err != nil {
+		panic(err)
+	}
+
+	for _, file := range files {
+		if file.IsDir() {
+			continue
+		} else if !strings.HasSuffix(file.Name(), ".h264") && !strings.HasSuffix(file.Name(), ".ivf") && !strings.HasSuffix(file.Name(), ".ogg") {
+			continue
+		}
+
+		frameDuration := 33 * time.Millisecond
+		if strings.HasSuffix(file.Name(), ".ogg") {
+			frameDuration = 20 * time.Millisecond
+		}
+
+		track, err := lksdk.NewLocalFileTrack(file.Name(),
+			lksdk.ReaderTrackWithFrameDuration(frameDuration),
+			lksdk.ReaderTrackWithOnWriteComplete(func() { fmt.Println("track finished") }),
+		)
+		if err != nil {
+			panic(err)
+		}
+
+		if _, err = room.LocalParticipant.PublishTrack(track, &lksdk.TrackPublicationOptions{
+			VideoWidth:  640,
+			VideoHeight: 480,
+			Name:        file.Name(),
+		}); err != nil {
+			panic(err)
+		}
+
+	}
+
+	<-sigChan
+	room.Disconnect()
+}