Skip to content

Commit f3a7503

Browse files
langhuihuilanghuihui
committed
feat: add whip client
1 parent 29e2142 commit f3a7503

File tree

5 files changed

+175
-98
lines changed

5 files changed

+175
-98
lines changed

‎plugin/webrtc/pkg/client.go‎

Lines changed: 0 additions & 90 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,11 @@ package webrtc
22

33
import (
44
"errors"
5-
"io"
6-
"net/http"
75
"strings"
86

97
. "github.com/pion/webrtc/v4"
108
"m7s.live/v5"
11-
"m7s.live/v5/pkg"
129
"m7s.live/v5/pkg/config"
13-
"m7s.live/v5/pkg/util"
1410
)
1511

1612
const (
@@ -43,92 +39,6 @@ func (c *Client) Start() (err error) {
4339
return c.MultipleConnection.Start()
4440
}
4541

46-
// WHIPClient is a client that pushes media to the server
47-
type WHIPClient struct {
48-
Client
49-
pushCtx m7s.PushJob
50-
}
51-
52-
func (c *WHIPClient) GetPushJob() *m7s.PushJob {
53-
return &c.pushCtx
54-
}
55-
56-
// WHEPClient is a client that pulls media from the server
57-
type WHEPClient struct {
58-
Client
59-
pullCtx m7s.PullJob
60-
}
61-
62-
func (c *WHEPClient) GetPullJob() *m7s.PullJob {
63-
return &c.pullCtx
64-
}
65-
66-
func (c *WHEPClient) Start() (err error) {
67-
err = c.pullCtx.Publish()
68-
if err != nil {
69-
return
70-
}
71-
c.Publisher = c.pullCtx.Publisher
72-
c.pullCtx.GoToStepConst(StepWebRTCInit)
73-
err = c.Client.Start()
74-
if err != nil {
75-
return
76-
}
77-
// u, _ := url.Parse(c.pullCtx.RemoteURL)
78-
// c.ApiBase, _, _ = strings.Cut(c.pullCtx.RemoteURL, "?")
79-
c.Receive()
80-
if c.pullCtx.PublishConfig.PubVideo {
81-
var transeiver *RTPTransceiver
82-
transeiver, err = c.AddTransceiverFromKind(RTPCodecTypeVideo, RTPTransceiverInit{
83-
Direction: RTPTransceiverDirectionRecvonly,
84-
})
85-
if err != nil {
86-
return
87-
}
88-
c.Info("webrtc add video transceiver", "transceiver", transeiver.Mid())
89-
}
90-
91-
if c.pullCtx.PublishConfig.PubAudio {
92-
var transeiver *RTPTransceiver
93-
transeiver, err = c.AddTransceiverFromKind(RTPCodecTypeAudio, RTPTransceiverInit{
94-
Direction: RTPTransceiverDirectionRecvonly,
95-
})
96-
if err != nil {
97-
return
98-
}
99-
c.Info("webrtc add audio transceiver", "transceiver", transeiver.Mid())
100-
}
101-
102-
c.pullCtx.GoToStepConst(StepOfferCreate)
103-
var sdpBody SDPBody
104-
sdpBody.SessionDescription, err = c.GetOffer()
105-
if err != nil {
106-
return
107-
}
108-
109-
c.pullCtx.GoToStepConst(StepSessionCreate)
110-
var res *http.Response
111-
res, err = http.DefaultClient.Post(c.pullCtx.RemoteURL, "application/sdp", strings.NewReader(sdpBody.SessionDescription.SDP))
112-
if err != nil {
113-
return
114-
}
115-
c.pullCtx.GoToStepConst(StepNegotiation)
116-
if res.StatusCode != http.StatusCreated && res.StatusCode != http.StatusOK {
117-
err = errors.New(res.Status)
118-
return
119-
}
120-
var sd SessionDescription
121-
sd.Type = SDPTypeAnswer
122-
var body util.Buffer
123-
io.Copy(&body, res.Body)
124-
sd.SDP = string(body)
125-
if err = c.SetRemoteDescription(sd); err != nil {
126-
return
127-
}
128-
c.pullCtx.GoToStepConst(pkg.StepStreaming)
129-
return
130-
}
131-
13242
func NewPuller(conf config.Pull) m7s.IPuller {
13343
if strings.HasPrefix(conf.URL, "https://rtc.live.cloudflare.com") {
13444
return NewCFClient(DIRECTION_PULL)

‎plugin/webrtc/pkg/whep_client.go‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
package webrtc
2+
3+
import (
4+
"errors"
5+
"io"
6+
"net/http"
7+
"strings"
8+
9+
. "github.com/pion/webrtc/v4"
10+
"m7s.live/v5"
11+
"m7s.live/v5/pkg"
12+
"m7s.live/v5/pkg/util"
13+
)
14+
15+
// WHEPClient is a client that pulls media from the server
16+
type WHEPClient struct {
17+
Client
18+
pullCtx m7s.PullJob
19+
}
20+
21+
func (c *WHEPClient) GetPullJob() *m7s.PullJob {
22+
return &c.pullCtx
23+
}
24+
25+
func (c *WHEPClient) Start() (err error) {
26+
err = c.pullCtx.Publish()
27+
if err != nil {
28+
return
29+
}
30+
c.Publisher = c.pullCtx.Publisher
31+
c.pullCtx.GoToStepConst(StepWebRTCInit)
32+
err = c.Client.Start()
33+
if err != nil {
34+
return
35+
}
36+
// u, _ := url.Parse(c.pullCtx.RemoteURL)
37+
// c.ApiBase, _, _ = strings.Cut(c.pullCtx.RemoteURL, "?")
38+
if c.pullCtx.PublishConfig.PubVideo {
39+
var transeiver *RTPTransceiver
40+
transeiver, err = c.AddTransceiverFromKind(RTPCodecTypeVideo, RTPTransceiverInit{
41+
Direction: RTPTransceiverDirectionRecvonly,
42+
})
43+
if err != nil {
44+
return
45+
}
46+
c.Info("webrtc add video transceiver", "transceiver", transeiver.Mid())
47+
}
48+
49+
if c.pullCtx.PublishConfig.PubAudio {
50+
var transeiver *RTPTransceiver
51+
transeiver, err = c.AddTransceiverFromKind(RTPCodecTypeAudio, RTPTransceiverInit{
52+
Direction: RTPTransceiverDirectionRecvonly,
53+
})
54+
if err != nil {
55+
return
56+
}
57+
c.Info("webrtc add audio transceiver", "transceiver", transeiver.Mid())
58+
}
59+
60+
c.pullCtx.GoToStepConst(StepOfferCreate)
61+
var sdpBody SDPBody
62+
sdpBody.SessionDescription, err = c.GetOffer()
63+
if err != nil {
64+
return
65+
}
66+
67+
c.pullCtx.GoToStepConst(StepSessionCreate)
68+
var res *http.Response
69+
res, err = http.DefaultClient.Post(c.pullCtx.RemoteURL, "application/sdp", strings.NewReader(sdpBody.SessionDescription.SDP))
70+
if err != nil {
71+
return
72+
}
73+
c.pullCtx.GoToStepConst(StepNegotiation)
74+
if res.StatusCode != http.StatusCreated && res.StatusCode != http.StatusOK {
75+
err = errors.New(res.Status)
76+
return
77+
}
78+
var sd SessionDescription
79+
sd.Type = SDPTypeAnswer
80+
var body util.Buffer
81+
io.Copy(&body, res.Body)
82+
sd.SDP = string(body)
83+
if err = c.SetRemoteDescription(sd); err != nil {
84+
return
85+
}
86+
c.pullCtx.GoToStepConst(pkg.StepStreaming)
87+
return
88+
}

‎plugin/webrtc/pkg/whip_client.go‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
package webrtc
2+
3+
import (
4+
"errors"
5+
"io"
6+
"net/http"
7+
"strings"
8+
9+
. "github.com/pion/webrtc/v4"
10+
"m7s.live/v5"
11+
"m7s.live/v5/pkg"
12+
"m7s.live/v5/pkg/util"
13+
)
14+
15+
// WHIP push steps definition
16+
var webrtcPushSteps = []pkg.StepDef{
17+
{Name: pkg.StepPublish, Description: "Publishing stream"},
18+
{Name: StepWebRTCInit, Description: "Initializing WebRTC connection"},
19+
{Name: StepOfferCreate, Description: "Creating WebRTC offer"},
20+
{Name: StepSessionCreate, Description: "Creating session with server"},
21+
{Name: StepTrackSetup, Description: "Setting up media tracks"},
22+
{Name: StepNegotiation, Description: "Completing WebRTC negotiation"},
23+
{Name: pkg.StepStreaming, Description: "Pushing media stream"},
24+
}
25+
26+
// WHIPClient is a client that pushes media to the server
27+
type WHIPClient struct {
28+
Client
29+
pushCtx m7s.PushJob
30+
}
31+
32+
func (c *WHIPClient) GetPushJob() *m7s.PushJob {
33+
return &c.pushCtx
34+
}
35+
36+
func (c *WHIPClient) Start() (err error) {
37+
err = c.pushCtx.Subscribe()
38+
if err != nil {
39+
return
40+
}
41+
c.Subscriber = c.pushCtx.Subscriber
42+
c.Info("Initializing WHIP WebRTC connection")
43+
err = c.Client.Start()
44+
if err != nil {
45+
return
46+
}
47+
48+
c.Info("Creating WebRTC offer")
49+
var sdpBody SDPBody
50+
sdpBody.SessionDescription, err = c.GetOffer()
51+
if err != nil {
52+
return
53+
}
54+
55+
// Send offer to WHIP endpoint
56+
c.Info("Sending offer to WHIP endpoint", "url", c.pushCtx.RemoteURL)
57+
c.Debug("sdp", sdpBody.SessionDescription.SDP)
58+
var res *http.Response
59+
res, err = http.DefaultClient.Post(c.pushCtx.RemoteURL, "application/sdp", strings.NewReader(sdpBody.SessionDescription.SDP))
60+
if err != nil {
61+
return
62+
}
63+
var body util.Buffer
64+
io.Copy(&body, res.Body)
65+
66+
if res.StatusCode != http.StatusCreated && res.StatusCode != http.StatusOK {
67+
err = errors.New(res.Status + string(body))
68+
return
69+
}
70+
71+
// Parse answer from server
72+
c.Info("Processing WHIP answer from server")
73+
var sd SessionDescription
74+
sd.Type = SDPTypeAnswer
75+
76+
sd.SDP = string(body)
77+
if err = c.SetRemoteDescription(sd); err != nil {
78+
return
79+
}
80+
81+
c.Info("WHIP negotiation completed, ready to push media")
82+
return
83+
}

‎pull_proxy.go‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -314,10 +314,8 @@ func (s *Server) AddPullProxy(ctx context.Context, req *pb.PullProxyInfo) (res *
314314
return
315315
}
316316
switch u.Scheme {
317-
case "srt", "rtsp", "rtmp":
317+
case "srt", "rtsp", "rtmp", "webrtc":
318318
pullProxyConfig.Type = u.Scheme
319-
case "whep":
320-
pullProxyConfig.Type = "webrtc"
321319
default:
322320
ext := filepath.Ext(u.Path)
323321
switch ext {
@@ -422,10 +420,8 @@ func (s *Server) UpdatePullProxy(ctx context.Context, req *pb.UpdatePullProxyReq
422420
return
423421
}
424422
switch u.Scheme {
425-
case "srt", "rtsp", "rtmp":
423+
case "srt", "rtsp", "rtmp", "webrtc":
426424
target.Type = u.Scheme
427-
case "whep":
428-
target.Type = "webrtc"
429425
default:
430426
ext := filepath.Ext(u.Path)
431427
switch ext {

‎push_proxy.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -196,7 +196,7 @@ func (d *PushProxyConfig) InitializeWithServer(s *Server) {
196196
return
197197
}
198198
switch u.Scheme {
199-
case "srt", "rtsp", "rtmp":
199+
case "srt", "rtsp", "rtmp", "webrtc":
200200
d.Type = u.Scheme
201201
default:
202202
ext := filepath.Ext(u.Path)
@@ -272,7 +272,7 @@ func (s *Server) AddPushProxy(ctx context.Context, req *pb.PushProxyInfo) (res *
272272
return
273273
}
274274
switch u.Scheme {
275-
case "srt", "rtsp", "rtmp":
275+
case "srt", "rtsp", "rtmp", "webrtc":
276276
device.Type = u.Scheme
277277
default:
278278
ext := filepath.Ext(u.Path)

0 commit comments

Comments
 (0)