Join room and start pipeline #2
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -23,6 +23,7 @@ import ( | |
|
|
||
| "github.com/go-gst/go-glib/glib" | ||
| "github.com/go-gst/go-gst/gst" | ||
| "golang.org/x/sync/errgroup" | ||
|
|
||
| "github.com/livekit/protocol/livekit" | ||
| "github.com/livekit/protocol/logger" | ||
|
|
@@ -74,47 +75,56 @@ func (p *Publisher) Start() error { | |
| return err | ||
| } | ||
|
|
||
| var g errgroup.Group | ||
|
|
||
| g.Go(func() error { | ||
| cb := lksdk.NewRoomCallback() | ||
| cb.OnDisconnected = func() { | ||
| // TODO: stop publishing and exit | ||
| } | ||
| p.room = lksdk.NewRoom(cb) | ||
| if err := p.room.JoinWithToken(p.params.URL, p.params.Token, | ||
| lksdk.WithAutoSubscribe(false), | ||
| ); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if p.videoTrack != nil { | ||
| pub, err := p.room.LocalParticipant.PublishTrack(p.videoTrack.track, &lksdk.TrackPublicationOptions{ | ||
| Source: livekit.TrackSource_CAMERA, | ||
| }) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| p.videoTrack.publication = pub | ||
| onEOS := func() { | ||
| _ = p.room.LocalParticipant.UnpublishTrack(pub.SID()) | ||
| } | ||
| p.videoTrack.onEOS.Store(&onEOS) | ||
| } | ||
|
|
||
| if p.audioTrack != nil { | ||
| pub, err := p.room.LocalParticipant.PublishTrack(p.audioTrack.track, &lksdk.TrackPublicationOptions{ | ||
| Source: livekit.TrackSource_MICROPHONE, | ||
| }) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| p.audioTrack.publication = pub | ||
| onEOS := func() { | ||
| _ = p.room.LocalParticipant.UnpublishTrack(pub.SID()) | ||
| } | ||
| p.audioTrack.onEOS.Store(&onEOS) | ||
| } | ||
| return nil | ||
| }) | ||
|
Comment on lines
+80
to
+120
There was a problem hiding this comment. 🟡 Race: pipeline EOS can fire before onEOS callback is registered, causing per-track unpublish to be skipped The parallelization of room connection and pipeline startup introduces a race where When Prompt for agents
Was this helpful? React with 👍 or 👎 to provide feedback. |
||
|
|
||
| g.Go(func() error { | ||
| return p.pipeline.Start() | ||
| }) | ||
|
Comment on lines
+80
to
+124
There was a problem hiding this comment. 🧩 Analysis chain🌐 Web query:
💡 Result: In golang.org/x/sync/errgroup, a plain errgroup.Group (created as a zero value, without calling WithContext) has no context cancellation capability. When using g.Go(f), all submitted functions run to completion in their goroutines regardless of errors returned by other functions. The first non-nil error is captured and returned by g.Wait after all goroutines finish, but no cancellation occurs and remaining goroutines continue executing without interruption. This differs from errgroup.WithContext(ctx), where the group has an associated context that gets canceled (via the internal cancel func) when the first error is encountered. Goroutines can then check ctx.Err or use context-aware operations to exit early. Key evidence from source code: - Group struct has cancel func(error), which is nil in plain Group. - In Go/TryGo: if err != nil, g.errOnce.Do sets g.err and calls g.cancel(g.err) only if g.cancel != nil. - Wait calls g.wg.Wait (blocks until all done), then if g.cancel != nil calls g.cancel(g.err), then returns g.err. Plain Group behaves like sync.WaitGroup + error collection: always waits for all, no cancellation on error. Use WithContext explicitly for cancellation behavior. Citations:
🏁 Script executed: #!/bin/bash
# First, locate the publish.go file
find . -name "publish.go" -type f | head -5Repository: MainSquare/gstreamer-publisher Length of output: 87 🏁 Script executed: #!/bin/bash
# Read the relevant section of publish.go
sed -n '80,124p' ./publish.goRepository: MainSquare/gstreamer-publisher Length of output: 1187 🏁 Script executed: #!/bin/bash
# Get more context around the lines - see the full function/context
sed -n '60,140p' ./publish.goRepository: MainSquare/gstreamer-publisher Length of output: 1750 🏁 Script executed: #!/bin/bash
# Search for the Stop() method
grep -n "func.*Stop" ./publish.go | head -10Repository: MainSquare/gstreamer-publisher Length of output: 107 🏁 Script executed: #!/bin/bash
# Read the Stop() method
sed -n '142,200p' ./publish.goRepository: MainSquare/gstreamer-publisher Length of output: 1408 🏁 Script executed: #!/bin/bash
# Search for where onEOS is used/called
rg "onEOS" ./publish.go -A 2 -B 2Repository: MainSquare/gstreamer-publisher Length of output: 417 🏁 Script executed: #!/bin/bash
# Search for where onEOS callbacks are loaded and called
rg "onEOS.*Load" ./publish.go -A 3 -B 1Repository: MainSquare/gstreamer-publisher Length of output: 56 🏁 Script executed: #!/bin/bash
# Search more broadly for Load calls on track fields
rg "\.Load\(" ./publish.go -B 3 -A 2Repository: MainSquare/gstreamer-publisher Length of output: 56 🏁 Script executed: #!/bin/bash
# Search for Track type definition
rg "type.*Track" ./publish.go -A 5 | head -40Repository: MainSquare/gstreamer-publisher Length of output: 56 🏁 Script executed: #!/bin/bash
# Find all Go files in the repository
find . -name "*.go" -type fRepository: MainSquare/gstreamer-publisher Length of output: 108 🏁 Script executed: #!/bin/bash
# Read track.go to understand track structure and onEOS
cat ./track.goRepository: MainSquare/gstreamer-publisher Length of output: 4466 Reconsider concurrent ordering of pipeline start vs. room join/publish.
Consider starting the pipeline only after the room is joined and tracks are published, and use ♻️ Sketch of a sequential ordering- var g errgroup.Group
-
- g.Go(func() error {
- cb := lksdk.NewRoomCallback()
- ...
- })
-
- g.Go(func() error {
- return p.pipeline.Start()
- })
-
- if err := g.Wait(); err != nil {
- p.Stop()
- return err
- }
+ if err := p.joinAndPublish(); err != nil {
+ p.Stop()
+ return err
+ }
+ if err := p.pipeline.Start(); err != nil {
+ p.Stop()
+ return err
+ }🤖 Prompt for AI Agents |
||
|
|
||
| if err := g.Wait(); err != nil { | ||
| p.Stop() | ||
| return err | ||
| } | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Possible nil-pointer race:
onEOSclosure dereferencesp.roomwhichStop()nils.The
onEOSclosure callsp.room.LocalParticipant.UnpublishTrack(pub.SID()), butStop()(lines 148–151) setsp.room = nilafter disconnecting. EOS can be triggered from the GStreamer thread (handleEOSintrack.go) concurrently withStop()being invoked frommessageWatch(onMessageEOS/MessageError) or the signal goroutine, leaving a window where the closure dereferences a nilp.room.Capture
p.room(or justp.room.LocalParticipant) in the closure to make the callback self-contained, or guard the access:🛡️ Proposed fix
(apply the same pattern to the audio block at lines 106–118.)
Also applies to: 114-117
🤖 Prompt for AI Agents