deblan-mirror/nextcloud-spreed-signaling
deblan-mirror/nextcloud-spreed-signaling
1
0
Fork 0
mirror of https://github.com/strukturag/nextcloud-spreed-signaling synced 2026-03-14 14:35:44 +01:00
Code Issues Releases Wiki Activity

Simplify API for waiter releasing.

Browse source
This commit is contained in:
Joachim Bauch 2026-03-10 10:04:18 +01:00
commit 182a0b78e2
No known key found for this signature in database
GPG key ID: 77C1D22D53E15F02
3 changed files with 26 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

11
async/notifier.go
View file

@ -58,7 +58,9 @@ type Notifier struct {
waiterMap map[string]map[*Waiter]bool waiterMap map[string]map[*Waiter]bool
} }
func (n *Notifier) NewWaiter(key string) *Waiter { type ReleaseFunc func()
func (n *Notifier) NewWaiter(key string) (*Waiter, ReleaseFunc) {
n.Lock() n.Lock()
defer n.Unlock() defer n.Unlock()
@ -86,7 +88,10 @@ func (n *Notifier) NewWaiter(key string) *Waiter {
ch: waiter.ch, ch: waiter.ch,
} }
n.waiterMap[key][w] = true n.waiterMap[key][w] = true
return w releaseFunc := func() {
n.release(w)
}
return w, releaseFunc
} }
func (n *Notifier) Reset() { func (n *Notifier) Reset() {
@ -100,7 +105,7 @@ func (n *Notifier) Reset() {
n.waiterMap = nil n.waiterMap = nil
} }
func (n *Notifier) Release(w *Waiter) { func (n *Notifier) release(w *Waiter) {
n.Lock() n.Lock()
defer n.Unlock() defer n.Unlock()

32
async/notifier_test.go
View file

@ -54,8 +54,8 @@ func TestNotifierWaitTimeout(t *testing.T) {
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
defer cancel() defer cancel()
waiter := notifier.NewWaiter("foo") waiter, release := notifier.NewWaiter("foo")
defer notifier.Release(waiter) defer release()
err := waiter.Wait(ctx) err := waiter.Wait(ctx)
assert.ErrorIs(t, err, context.DeadlineExceeded) assert.ErrorIs(t, err, context.DeadlineExceeded)
@ -69,8 +69,8 @@ func TestNotifierSimple(t *testing.T) {
t.Parallel() t.Parallel()
var notifier Notifier var notifier Notifier
waiter := notifier.NewWaiter("foo") waiter, release := notifier.NewWaiter("foo")
defer notifier.Release(waiter) defer release()
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Go(func() { wg.Go(func() {
@ -87,8 +87,8 @@ func TestNotifierMultiNotify(t *testing.T) {
t.Parallel() t.Parallel()
var notifier Notifier var notifier Notifier
waiter := notifier.NewWaiter("foo") _, release := notifier.NewWaiter("foo")
defer notifier.Release(waiter) defer release()
notifier.Notify("foo") notifier.Notify("foo")
// The second notification will be ignored while the first is still pending. // The second notification will be ignored while the first is still pending.
@ -99,8 +99,8 @@ func TestNotifierWaitClosed(t *testing.T) {
t.Parallel() t.Parallel()
var notifier Notifier var notifier Notifier
waiter := notifier.NewWaiter("foo") waiter, release := notifier.NewWaiter("foo")
notifier.Release(waiter) release()
assert.NoError(t, waiter.Wait(context.Background())) assert.NoError(t, waiter.Wait(context.Background()))
} }
@ -109,10 +109,10 @@ func TestNotifierWaitClosedMulti(t *testing.T) {
t.Parallel() t.Parallel()
var notifier Notifier var notifier Notifier
waiter1 := notifier.NewWaiter("foo") waiter1, release1 := notifier.NewWaiter("foo")
waiter2 := notifier.NewWaiter("foo") waiter2, release2 := notifier.NewWaiter("foo")
notifier.Release(waiter1) release1()
notifier.Release(waiter2) release2()
assert.NoError(t, waiter1.Wait(context.Background())) assert.NoError(t, waiter1.Wait(context.Background()))
assert.NoError(t, waiter2.Wait(context.Background())) assert.NoError(t, waiter2.Wait(context.Background()))
@ -122,8 +122,8 @@ func TestNotifierResetWillNotify(t *testing.T) {
t.Parallel() t.Parallel()
var notifier Notifier var notifier Notifier
waiter := notifier.NewWaiter("foo") waiter, release := notifier.NewWaiter("foo")
defer notifier.Release(waiter) defer release()
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Go(func() { wg.Go(func() {
@ -144,8 +144,8 @@ func TestNotifierDuplicate(t *testing.T) {
for range 2 { for range 2 {
done.Go(func() { done.Go(func() {
waiter := notifier.NewWaiter("foo") waiter, release := notifier.NewWaiter("foo")
defer notifier.Release(waiter) defer release()
ctx, cancel := context.WithTimeout(context.Background(), time.Second) ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel() defer cancel()

9
sfu/janus/janus.go
View file

@ -888,15 +888,10 @@ func (m *janusSFU) notifyPublisherConnected(id api.PublicSessionId, streamType s
m.publisherConnected.Notify(string(key)) m.publisherConnected.Notify(string(key))
} }
func (m *janusSFU) newPublisherConnectedWaiter(id api.PublicSessionId, streamType sfu.StreamType) (*async.Waiter, func()) { func (m *janusSFU) newPublisherConnectedWaiter(id api.PublicSessionId, streamType sfu.StreamType) (*async.Waiter, async.ReleaseFunc) {
key := sfu.GetStreamId(id, streamType) key := sfu.GetStreamId(id, streamType)
waiter := m.publisherConnected.NewWaiter(string(key)) return m.publisherConnected.NewWaiter(string(key))
stopFunc := func() {
m.publisherConnected.Release(waiter)
}
return waiter, stopFunc
} }
func (m *janusSFU) getPublisher(ctx context.Context, publisher api.PublicSessionId, streamType sfu.StreamType) (*janusPublisher, error) { func (m *janusSFU) getPublisher(ctx context.Context, publisher api.PublicSessionId, streamType sfu.StreamType) (*janusPublisher, error) {