gilles/go2rtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/master' into refactr-syscall-more-generic

Browse Source
This commit is contained in:
Sergey Krashevich
2024-05-07 14:45:48 +03:00
parent abe617a346 a9f2b5158c
commit f158ffb33e
32 changed files with 721 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/core/helpers.go
-9
View File
@@ -38,15 +38,6 @@ func RandString(size, base byte) string {
return string(b)
}
func Any(errs ...error) error {
for _, err := range errs {
if err != nil {
return err
}
}
return nil
}
func Between(s, sub1, sub2 string) string {
i := strings.Index(s, sub1)
if i < 0 {
pkg/flv/amf/amf_test.go
+1 -1
View File
@@ -141,7 +141,7 @@ func TestNewReader(t *testing.T) {
name: "obs-connect",
actual: "020007636f6e6e656374003ff000000000000003000361707002000c617070312f73747265616d3100047479706502000a6e6f6e70726976617465000e737570706f727473476f4177617901010008666c61736856657202001f464d4c452f332e302028636f6d70617469626c653b20464d53632f312e3029000673776655726c02002272746d703a2f2f3139322e3136382e31302e3130312f617070312f73747265616d310005746355726c02002272746d703a2f2f3139322e3136382e31302e3130312f617070312f73747265616d31000009",
expect: []any{
"connect", 1,
"connect", float64(1),
map[string]any{
"app": "app1/stream1",
"flashVer": "FMLE/3.0 (compatible; FMSc/1.0)",
pkg/h264/payloader.go
+7 -3
View File
@@ -67,11 +67,15 @@ func EmitNalus(nals []byte, isAVC bool, emit func([]byte)) {
}
} else {
for {
end := 4 + binary.BigEndian.Uint32(nals)
emit(nals[4:end])
if int(end) >= len(nals) {
n := uint32(len(nals))
if n < 4 {
break
}
end := 4 + binary.BigEndian.Uint32(nals)
if n < end {
break
}
emit(nals[4:end])
nals = nals[end:]
}
}
pkg/hls/reader.go
+1 -1
View File
@@ -88,7 +88,7 @@ func (r *reader) RoundTrip(_ *http.Request) (*http.Response, error) {
}
func (r *reader) getSegment() ([]byte, error) {
for i := 0; i < 5; i++ {
for i := 0; i < 10; i++ {
if r.playlist == nil {
if wait := time.Second - time.Since(r.lastTime); wait > 0 {
time.Sleep(wait)
pkg/magic/producer.go
+3 -3
View File
@@ -34,12 +34,12 @@ func Open(r io.Reader) (core.Producer, error) {
case bytes.HasPrefix(b, []byte(flv.Signature)):
return flv.Open(rd)
case bytes.HasPrefix(b, []byte{0xFF, 0xF1}):
return aac.Open(rd)
case bytes.HasPrefix(b, []byte("--")):
return multipart.Open(rd)
case b[0] == 0xFF && b[1]&0xF7 == 0xF1:
return aac.Open(rd)
case b[0] == mpegts.SyncByte:
return mpegts.Open(rd)
}
pkg/mjpeg/helpers.go
+35
View File
@@ -0,0 +1,35 @@
package mjpeg
import (
"bytes"
"image/jpeg"
)
// FixJPEG - reencode JPEG if it has wrong header
//
// for example, this app produce "bad" images:
// https://github.com/jacksonliam/mjpg-streamer
//
// and they can't be uploaded to the Telegram servers:
// {"ok":false,"error_code":400,"description":"Bad Request: IMAGE_PROCESS_FAILED"}
func FixJPEG(b []byte) []byte {
// skip non-JPEG
if len(b) < 10 || b[0] != 0xFF || b[1] != 0xD8 {
return b
}
// skip if header OK for imghdr library
// https://docs.python.org/3/library/imghdr.html
if string(b[2:4]) == "\xFF\xDB" || string(b[6:10]) == "JFIF" || string(b[6:10]) == "Exif" {
return b
}
img, err := jpeg.Decode(bytes.NewReader(b))
if err != nil {
return b
}
buf := bytes.NewBuffer(nil)
if err = jpeg.Encode(buf, img, nil); err != nil {
return b
}
return buf.Bytes()
}
pkg/rtsp/server.go
+37 -1
View File
@@ -146,7 +146,19 @@ func (c *Conn) Accept() error {
if strings.HasPrefix(tr, transport) {
c.session = core.RandString(8, 10)
c.state = StateSetup
res.Header.Set("Transport", tr[:len(transport)+3])
if c.mode == core.ModePassiveConsumer {
if i := reqTrackID(req); i >= 0 && i < len(c.senders) {
// mark sender as SETUP
c.senders[i].Media.ID = MethodSetup
tr = fmt.Sprintf("RTP/AVP/TCP;unicast;interleaved=%d-%d", i*2, i*2+1)
res.Header.Set("Transport", tr)
} else {
res.Status = "400 Bad Request"
}
} else {
res.Header.Set("Transport", tr[:len(transport)+3])
}
} else {
res.Status = "461 Unsupported transport"
}
@@ -156,6 +168,15 @@ func (c *Conn) Accept() error {
}
case MethodRecord, MethodPlay:
if c.mode == core.ModePassiveConsumer {
// stop unconfigured senders
for _, track := range c.senders {
if track.Media.ID != MethodSetup {
track.Close()
}
}
}
res := &tcp.Response{Request: req}
err = c.WriteResponse(res)
c.playOK = true
@@ -172,3 +193,18 @@ func (c *Conn) Accept() error {
}
}
}
func reqTrackID(req *tcp.Request) int {
var s string
if req.URL.RawQuery != "" {
s = req.URL.RawQuery
} else {
s = req.URL.Path
}
if i := strings.LastIndexByte(s, '='); i > 0 {
if i, err := strconv.Atoi(s[i+1:]); err == nil {
return i
}
}
return -1
}
pkg/stdin/client.go
+33
View File
@@ -0,0 +1,33 @@
package stdin
import (
"os/exec"
"github.com/AlexxIT/go2rtc/pkg/core"
)
type Client struct {
cmd *exec.Cmd
medias []*core.Media
sender *core.Sender
send int
}
func NewClient(cmd *exec.Cmd) (*Client, error) {
c := &Client{
cmd: cmd,
medias: []*core.Media{
{
Kind: core.KindAudio,
Direction: core.DirectionSendonly,
Codecs: []*core.Codec{
{Name: core.CodecPCMA, ClockRate: 8000},
{Name: core.CodecPCM},
},
},
},
}
return c, nil
}
pkg/stdin/consumer.go
+61
View File
@@ -0,0 +1,61 @@
package stdin
import (
"encoding/json"
"errors"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/pion/rtp"
)
func (c *Client) GetMedias() []*core.Media {
return c.medias
}
func (c *Client) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver, error) {
return nil, core.ErrCantGetTrack
}
func (c *Client) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
if c.sender == nil {
stdin, err := c.cmd.StdinPipe()
if err != nil {
return err
}
c.sender = core.NewSender(media, track.Codec)
c.sender.Handler = func(packet *rtp.Packet) {
_, _ = stdin.Write(packet.Payload)
c.send += len(packet.Payload)
}
}
c.sender.HandleRTP(track)
return nil
}
func (c *Client) Start() (err error) {
return c.cmd.Run()
}
func (c *Client) Stop() (err error) {
if c.sender != nil {
c.sender.Close()
}
if c.cmd.Process == nil {
return nil
}
return errors.Join(c.cmd.Process.Kill(), c.cmd.Wait())
}
func (c *Client) MarshalJSON() ([]byte, error) {
info := &core.Info{
Type: "Exec active consumer",
Medias: c.medias,
Send: c.send,
}
if c.sender != nil {
info.Senders = []*core.Sender{c.sender}
}
return json.Marshal(info)
}
pkg/webrtc/helpers.go
+18
View File
@@ -47,6 +47,9 @@ func UnmarshalMedias(descriptions []*sdp.MediaDescription) (medias []*core.Media
continue
}
// skip non-media codecs to avoid confusing users in info and logs
media.Codecs = SkipNonMediaCodecs(media.Codecs)
medias = append(medias, media)
}
}
@@ -54,6 +57,21 @@ func UnmarshalMedias(descriptions []*sdp.MediaDescription) (medias []*core.Media
return
}
func SkipNonMediaCodecs(input []*core.Codec) (output []*core.Codec) {
for _, codec := range input {
switch codec.Name {
case "RTX", "RED", "ULPFEC", "FLEXFEC-03":
continue
case "CN", "TELEPHONE-EVENT":
continue // https://datatracker.ietf.org/doc/html/rfc7874
}
// VP8, VP9, H264, H265, AV1
// OPUS, G722, PCMU, PCMA
output = append(output, codec)
}
return
}
// WithResampling - will add for consumer: PCMA/0, PCMU/0, PCM/0, PCML/0
// so it can add resampling for PCMA/PCMU and repack for PCM/PCML
func WithResampling(medias []*core.Media) []*core.Media {
pkg/webrtc/track.go
+22 -12
View File
@@ -1,6 +1,8 @@
package webrtc
import (
"sync"
"github.com/pion/rtp"
"github.com/pion/webrtc/v3"
)
@@ -12,6 +14,7 @@ type Track struct {
sequence uint16
ssrc uint32
writer webrtc.TrackLocalWriter
mu sync.Mutex
}
func NewTrack(kind string) *Track {
@@ -23,8 +26,10 @@ func NewTrack(kind string) *Track {
}
func (t *Track) Bind(context webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) {
t.mu.Lock()
t.ssrc = uint32(context.SSRC())
t.writer = context.WriteStream()
t.mu.Unlock()
for _, parameters := range context.CodecParameters() {
// return first parameters
@@ -35,7 +40,9 @@ func (t *Track) Bind(context webrtc.TrackLocalContext) (webrtc.RTPCodecParameter
}
func (t *Track) Unbind(context webrtc.TrackLocalContext) error {
t.mu.Lock()
t.writer = nil
t.mu.Unlock()
return nil
}
@@ -55,19 +62,22 @@ func (t *Track) Kind() webrtc.RTPCodecType {
return webrtc.NewRTPCodecType(t.kind)
}
func (t *Track) WriteRTP(payloadType uint8, packet *rtp.Packet) error {
func (t *Track) WriteRTP(payloadType uint8, packet *rtp.Packet) (err error) {
// using mutex because Unbind https://github.com/AlexxIT/go2rtc/issues/994
t.mu.Lock()
// in case when we start WriteRTP before Track.Bind
if t.writer == nil {
return nil
if t.writer != nil {
// important to have internal counter if input packets from different sources
t.sequence++
header := packet.Header
header.SSRC = t.ssrc
header.PayloadType = payloadType
header.SequenceNumber = t.sequence
_, err = t.writer.WriteRTP(&header, packet.Payload)
}
// important to have internal counter if input packets from different sources
t.sequence++
header := packet.Header
header.SSRC = t.ssrc
header.PayloadType = payloadType
header.SequenceNumber = t.sequence
_, err := t.writer.WriteRTP(&header, packet.Payload)
return err
t.mu.Unlock()
return
}