update protocol

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/google/wire v0.5.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b
-	github.com/livekit/protocol v0.9.2
+	github.com/livekit/protocol v0.9.3
 	github.com/magefile/mage v1.11.0
 	github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0
 	github.com/mitchellh/go-homedir v1.1.0

go.sum

@@ -246,8 +246,8 @@ github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY
 github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
 github.com/livekit/ion-sfu v1.20.9 h1:Ih+tE5CXqdL1o6iXfZn+uBoddBXWceaOKVIun6aVuZM=
 github.com/livekit/ion-sfu v1.20.9/go.mod h1:g8hwobZI5fvX1RXvayf4ZXkgP7spV5YGE4yTSsumpB4=
-github.com/livekit/protocol v0.9.2 h1:ZX5m7DjmRMHtYH5oM85QRMuJlr4feTfD5UL22h9HmtM=
-github.com/livekit/protocol v0.9.2/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo=
+github.com/livekit/protocol v0.9.3 h1:rzJwes4AddFF/h47G6kU1/GQF8/meqZ3uihFIKPPxz0=
+github.com/livekit/protocol v0.9.3/go.mod h1:MEKn847Iu/2U8ClZyUmEm2oHn8k9fnSHy81Wv8kkSDo=
 github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ=
 github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE=
 github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ=

pkg/service/recordingservice.go

@@ -17,11 +17,11 @@ const lockExpiration = time.Second * 5
 
 type RecordingService struct {
 	mb       utils.MessageBus
-	notifier *webhook.Notifier
+	notifier webhook.Notifier
 	shutdown chan struct{}
 }
 
-func NewRecordingService(mb utils.MessageBus, notifier *webhook.Notifier) *RecordingService {
+func NewRecordingService(mb utils.MessageBus, notifier webhook.Notifier) *RecordingService {
 	return &RecordingService{
 		mb:       mb,
 		notifier: notifier,
@@ -110,7 +110,7 @@ func (s *RecordingService) EndRecording(ctx context.Context, req *livekit.EndRec
 }
 
 func (s *RecordingService) resultsWorker() {
-	sub, err := s.mb.Subscribe(context.Background(), utils.RecordingResultChannel)
+	sub, err := s.mb.SubscribeQueue(context.Background(), utils.RecordingResultChannel)
 	if err != nil {
 		logger.Errorw("failed to subscribe to results channel", err)
 		return
@@ -129,22 +129,13 @@ func (s *RecordingService) resultsWorker() {
 			}
 			s.notify(res)
 		case <-s.shutdown:
-			sub.Close()
+			_ = sub.Close()
 			return
 		}
 	}
 }
 
 func (s *RecordingService) notify(res *livekit.RecordingResult) {
-	acquired, err := s.mb.Lock(context.Background(), res.Id, lockExpiration)
-	if err != nil {
-		logger.Errorw("failed to lock", err)
-		return
-	}
-	if !acquired {
-		return
-	}
-
 	// log results
 	if res.Error != "" {
 		logger.Errorw("recording failed", errors.New(res.Error), "id", res.Id)

pkg/service/roommanager.go

@@ -30,7 +30,7 @@ type LocalRoomManager struct {
 	selector    routing.NodeSelector
 	router      routing.Router
 	currentNode routing.LocalNode
-	notifier    *webhook.Notifier
+	notifier    webhook.Notifier
 	rtcConfig   *rtc.WebRTCConfig
 	config      *config.Config
 	webhookPool *workerpool.WorkerPool
@@ -38,7 +38,7 @@ type LocalRoomManager struct {
 }
 
 func NewLocalRoomManager(rp RoomStore, router routing.Router, currentNode routing.LocalNode, selector routing.NodeSelector,
-	notifier *webhook.Notifier, conf *config.Config) (*LocalRoomManager, error) {
+	notifier webhook.Notifier, conf *config.Config) (*LocalRoomManager, error) {
 	rtcConf, err := rtc.NewWebRTCConfig(conf, currentNode.Ip)
 	if err != nil {
 		return nil, err

pkg/service/utils.go

@@ -65,7 +65,7 @@ func CreateKeyProvider(conf *config.Config) (auth.KeyProvider, error) {
 	return auth.NewFileBasedKeyProviderFromMap(conf.Keys), nil
 }
 
-func CreateWebhookNotifier(conf *config.Config, provider auth.KeyProvider) (*webhook.Notifier, error) {
+func CreateWebhookNotifier(conf *config.Config, provider auth.KeyProvider) (webhook.Notifier, error) {
 	wc := conf.WebHook
 	if len(wc.URLs) == 0 {
 		return nil, nil