Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix context propagation and work cancellation #20

Merged
merged 6 commits into from
Mar 15, 2023
Merged

Fix context propagation and work cancellation #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
16aa2b3
Move types to its own pkg
nassah221 Mar 15, 2023
10c1473
Move orchestrator public interface to its own pkg
nassah221 Mar 15, 2023
461c287
Move service public interface to its own pkg
nassah221 Mar 15, 2023
f9332ef
Enhance project structure
nassah221 Mar 15, 2023
14ebd2f
Fix import ordering and var-naming
nassah221 Mar 15, 2023
19f9d3c
Fix context propagation and sub-task cancellation
nassah221 Mar 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 8 additions & 7 deletions internal/inputs/fs/fs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,14 @@ import (
"time"

"github.com/Glimesh/waveguide/pkg/control"

"github.com/pion/webrtc/v3"
"github.com/pion/webrtc/v3/pkg/media"
"github.com/pion/webrtc/v3/pkg/media/h264reader"
"github.com/sirupsen/logrus"
)

type FSSource struct {
type Source struct {
log logrus.FieldLogger
control *control.Control

Expand All @@ -23,23 +24,23 @@ type FSSource struct {
AudioFile string `mapstructure:"audio_file"`
}

func New(address, videoFile, audioFile string) *FSSource {
return &FSSource{
func New(address, videoFile, audioFile string) *Source {
return &Source{
Address: address,
VideoFile: videoFile,
AudioFile: audioFile,
}
}

func (s *FSSource) SetControl(ctrl *control.Control) {
func (s *Source) SetControl(ctrl *control.Control) {
s.control = ctrl
}

func (s *FSSource) SetLogger(log logrus.FieldLogger) {
func (s *Source) SetLogger(log logrus.FieldLogger) {
s.log = log
}

func (s *FSSource) Listen(ctx context.Context) {
func (s *Source) Listen(ctx context.Context) {
s.log.Infof("Reading from FS for video=%s and audio=%s", s.VideoFile, s.AudioFile)

// Assert that we have an audio or video file
Expand All @@ -58,7 +59,7 @@ func (s *FSSource) Listen(ctx context.Context) {
panic(videoTrackErr)
}

stream, ctx, err := s.control.StartStream(1234)
stream, err := s.control.StartStream(1234)
if err != nil {
panic(err)
}
Expand Down
40 changes: 20 additions & 20 deletions internal/inputs/ftl/ftl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,35 +4,37 @@ import (
"context"
"net"

"github.com/Glimesh/waveguide/pkg/control"
control "github.com/Glimesh/waveguide/pkg/control"
ftlproto "github.com/Glimesh/waveguide/pkg/protocols/ftl"
types "github.com/Glimesh/waveguide/pkg/types"

"github.com/pion/rtp"
"github.com/pion/webrtc/v3"
"github.com/sirupsen/logrus"
)

type FTLSource struct {
type Source struct {
log logrus.FieldLogger
control *control.Control

Address string
}

func New(address string) *FTLSource {
return &FTLSource{
func New(address string) *Source {
return &Source{
Address: address,
}
}

func (s *FTLSource) SetControl(ctrl *control.Control) {
func (s *Source) SetControl(ctrl *control.Control) {
s.control = ctrl
}

func (s *FTLSource) SetLogger(log logrus.FieldLogger) {
func (s *Source) SetLogger(log logrus.FieldLogger) {
s.log = log
}

func (s *FTLSource) Listen(ctx context.Context) {
func (s *Source) Listen(ctx context.Context) {
tcpAddr, err := net.ResolveTCPAddr("tcp", s.Address)
if err != nil {
s.log.Errorf("Failed: %+v", err)
Expand Down Expand Up @@ -65,11 +67,10 @@ func (s *FTLSource) Listen(ctx context.Context) {
}

type connHandler struct {
control *control.Control
log logrus.FieldLogger
controlCtx context.Context
control *control.Control
log logrus.FieldLogger

channelID control.ChannelID
channelID types.ChannelID

stream *control.Stream
videoTrack *webrtc.TrackLocalStaticRTP
Expand All @@ -79,13 +80,13 @@ type connHandler struct {
}

func (c *connHandler) OnConnect(channelID ftlproto.ChannelID) error {
c.channelID = control.ChannelID(channelID)
c.channelID = types.ChannelID(channelID)

var err error
c.stream, c.controlCtx, err = c.control.StartStream(c.channelID)
stream, err := c.control.StartStream(c.channelID)
if err != nil {
return err
}
c.stream = stream

// Create a video track
c.videoTrack, err = webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{MimeType: "video/h264"}, "video", "pion")
Expand Down Expand Up @@ -124,10 +125,9 @@ func (c *connHandler) OnPlay(metadata ftlproto.FtlConnectionMetadata) error {
}

func (c *connHandler) OnAudio(packet *rtp.Packet) error {
if c.controlCtx.Err() != nil {
return c.controlCtx.Err()
if err := c.control.ContextErr(); err != nil {
return err
}

err := c.audioTrack.WriteRTP(packet)

c.stream.ReportMetadata(control.AudioPacketsMetadata(len(packet.Payload)))
Expand All @@ -136,8 +136,8 @@ func (c *connHandler) OnAudio(packet *rtp.Packet) error {
}

func (c *connHandler) OnVideo(packet *rtp.Packet) error {
if c.controlCtx.Err() != nil {
return c.controlCtx.Err()
if err := c.control.ContextErr(); err != nil {
return err
}

// Write the RTP packet immediately, log after
Expand All @@ -149,7 +149,7 @@ func (c *connHandler) OnVideo(packet *rtp.Packet) error {
}

func (c *connHandler) OnClose() {
if c.controlCtx.Err() == nil {
if c.control.ContextErr() == nil {
// This is the FTL => Control cancellation
// Only since if we're not the canceller.
c.control.StopStream(c.channelID)
Expand Down
24 changes: 13 additions & 11 deletions internal/inputs/janus/janus.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,33 +11,35 @@ import (
"time"

"github.com/Glimesh/waveguide/pkg/control"
"github.com/Glimesh/waveguide/pkg/types"

"github.com/pion/webrtc/v3"
"github.com/sirupsen/logrus"
)

type JanusSource struct {
type Source struct {
log logrus.FieldLogger
control *control.Control

channelID control.ChannelID
channelID types.ChannelID

// Address to connect to for Janus
Address string
ChannelID int `mapstructure:"channel_id"`
}

func New(address string, channelID int) control.Input {
return &JanusSource{
return &Source{
Address: address,
ChannelID: channelID,
}
}

func (s *JanusSource) SetControl(ctrl *control.Control) {
func (s *Source) SetControl(ctrl *control.Control) {
s.control = ctrl
}

func (s *JanusSource) SetLogger(log logrus.FieldLogger) {
func (s *Source) SetLogger(log logrus.FieldLogger) {
s.log = log
}

Expand All @@ -60,10 +62,10 @@ type janusFtlOfferResponse struct {
Jsep JSEP `json:"jsep"`
}

func (s *JanusSource) Listen(ctx context.Context) {
func (s *Source) Listen(ctx context.Context) {
s.log.Infof("Connecting to janus=%s for channel_id=%d", s.Address, s.ChannelID)

s.channelID = control.ChannelID(s.ChannelID)
s.channelID = types.ChannelID(s.ChannelID)

values := map[string]string{"janus": "create", "transaction": randString()}

Expand Down Expand Up @@ -165,8 +167,8 @@ func (s *JanusSource) Listen(ctx context.Context) {
}()
}

func (s *JanusSource) negotiate(sdpString string, pluginUrl string) {
stream, ctx, err := s.control.StartStream(control.ChannelID(s.ChannelID))
func (s *Source) negotiate(sdpString string, pluginUrl string) {
stream, err := s.control.StartStream(types.ChannelID(s.ChannelID))
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -227,7 +229,7 @@ func (s *JanusSource) negotiate(sdpString string, pluginUrl string) {
if codec.MimeType == "audio/opus" {
s.log.Info("Got Opus track, sending to audio track")
for {
if ctx.Err() != nil {
if err := s.control.ContextErr(); err != nil {
return
}

Expand All @@ -241,7 +243,7 @@ func (s *JanusSource) negotiate(sdpString string, pluginUrl string) {
} else if codec.MimeType == "video/H264" {
s.log.Info("Got H264 track, sending to video track")
for {
if ctx.Err() != nil {
if err := s.control.ContextErr(); err != nil {
return
}

Expand Down
Loading