dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-09 18:52:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

use typed ops queue in pctransport (#2656)

Browse Source
This commit is contained in:
Paul Wells
2024-04-16 03:23:08 -07:00
committed by GitHub
parent 1ab0879d28
commit b77f0256c7
2 changed files with 32 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+9 -7
View File
@@ -164,7 +164,7 @@ type ParticipantImpl struct {
disconnectTimer *time.Timer
migrationTimer *time.Timer
pubRTCPQueue *sutils.OpsQueue
pubRTCPQueue *sutils.TypedOpsQueue[[]rtcp.Packet]
// hold reference for MediaTrack
twcc *twcc.Responder
@@ -244,7 +244,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
p := &ParticipantImpl{
params: params,
disconnected: make(chan struct{}),
pubRTCPQueue: sutils.NewOpsQueue(sutils.OpsQueueParams{
pubRTCPQueue: sutils.NewTypedOpsQueue[[]rtcp.Packet](sutils.OpsQueueParams{
Name: "pub-rtcp",
MinSize: 64,
Logger: params.Logger,
@@ -2379,11 +2379,13 @@ func (p *ParticipantImpl) postRtcp(pkts []rtcp.Packet) {
return
}
p.pubRTCPQueue.Enqueue(func() {
if err := p.TransportManager.WritePublisherRTCP(pkts); err != nil && !IsEOF(err) {
p.pubLogger.Errorw("could not write RTCP to participant", err)
}
})
p.pubRTCPQueue.Enqueue(p.writePublisherRTCP, pkts)
}
func (p *ParticipantImpl) writePublisherRTCP(pkts []rtcp.Packet) {
if err := p.TransportManager.WritePublisherRTCP(pkts); err != nil && !IsEOF(err) {
p.pubLogger.Errorw("could not write RTCP to participant", err)
}
}
func (p *ParticipantImpl) setDowntracksConnected() {
pkg/rtc/transport.go
+23 -26
View File
@@ -183,7 +183,7 @@ type PCTransport struct {
preferTCP atomic.Bool
isClosed atomic.Bool
eventsQueue *sutils.OpsQueue
eventsQueue *sutils.TypedOpsQueue[event]
// the following should be accessed only in event processing go routine
cacheLocalCandidates bool
@@ -388,7 +388,7 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) {
params: params,
debouncedNegotiate: debounce.New(negotiationFrequency),
negotiationState: transport.NegotiationStateNone,
eventsQueue: sutils.NewOpsQueue(utils.OpsQueueParams{
eventsQueue: sutils.NewTypedOpsQueue[event](utils.OpsQueueParams{
Name: "transport",
MinSize: 64,
Logger: params.Logger,
@@ -1241,37 +1241,34 @@ func (t *PCTransport) parseTrackMid(offer webrtc.SessionDescription, senders map
}
func (t *PCTransport) postEvent(event event) {
t.eventsQueue.Enqueue(func() {
err := t.handleEvent(&event)
if err != nil {
if !t.isClosed.Load() {
t.params.Logger.Warnw("error handling event", err, "event", event.String())
t.params.Handler.OnNegotiationFailed()
}
}
})
t.eventsQueue.Enqueue(t.handleEvent, event)
}
func (t *PCTransport) handleEvent(e *event) error {
func (t *PCTransport) handleEvent(e event) {
var err error
switch e.signal {
case signalICEGatheringComplete:
return t.handleICEGatheringComplete(e)
err = t.handleICEGatheringComplete(e)
case signalLocalICECandidate:
return t.handleLocalICECandidate(e)
err = t.handleLocalICECandidate(e)
case signalRemoteICECandidate:
return t.handleRemoteICECandidate(e)
err = t.handleRemoteICECandidate(e)
case signalSendOffer:
return t.handleSendOffer(e)
err = t.handleSendOffer(e)
case signalRemoteDescriptionReceived:
return t.handleRemoteDescriptionReceived(e)
err = t.handleRemoteDescriptionReceived(e)
case signalICERestart:
return t.handleICERestart(e)
err = t.handleICERestart(e)
}
if err != nil {
if !t.isClosed.Load() {
t.params.Logger.Warnw("error handling event", err, "event", e.String())
t.params.Handler.OnNegotiationFailed()
}
}
return nil
}
func (t *PCTransport) handleICEGatheringComplete(_ *event) error {
func (t *PCTransport) handleICEGatheringComplete(_ event) error {
if t.params.IsOfferer {
return t.handleICEGatheringCompleteOfferer()
} else {
@@ -1329,7 +1326,7 @@ func (t *PCTransport) clearLocalDescriptionSent() {
t.connectionDetails.Clear()
}
func (t *PCTransport) handleLocalICECandidate(e *event) error {
func (t *PCTransport) handleLocalICECandidate(e event) error {
c := e.data.(*webrtc.ICECandidate)
filtered := false
@@ -1353,7 +1350,7 @@ func (t *PCTransport) handleLocalICECandidate(e *event) error {
return t.params.Handler.OnICECandidate(c, t.params.Transport)
}
func (t *PCTransport) handleRemoteICECandidate(e *event) error {
func (t *PCTransport) handleRemoteICECandidate(e event) error {
c := e.data.(*webrtc.ICECandidateInit)
filtered := false
@@ -1549,11 +1546,11 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
return t.localDescriptionSent()
}
func (t *PCTransport) handleSendOffer(_ *event) error {
func (t *PCTransport) handleSendOffer(_ event) error {
return t.createAndSendOffer(nil)
}
func (t *PCTransport) handleRemoteDescriptionReceived(e *event) error {
func (t *PCTransport) handleRemoteDescriptionReceived(e event) error {
sd := e.data.(*webrtc.SessionDescription)
if sd.Type == webrtc.SDPTypeOffer {
return t.handleRemoteOfferReceived(sd)
@@ -1778,7 +1775,7 @@ func (t *PCTransport) doICERestart() error {
}
}
func (t *PCTransport) handleICERestart(_ *event) error {
func (t *PCTransport) handleICERestart(_ event) error {
return t.doICERestart()
}