diff --git a/pkg/aac/adts.go b/pkg/aac/adts.go index 45d3167e..e38b4c42 100644 --- a/pkg/aac/adts.go +++ b/pkg/aac/adts.go @@ -40,7 +40,7 @@ func ADTSToCodec(b []byte) *core.Codec { //_ = rd.ReadBits(16) // CRC check // 3. Encode RTP config - wr := bits.NewWriter() + wr := bits.NewWriter(nil) wr.WriteBits8(objType, 5) wr.WriteBits8(sampleRateIdx, 4) wr.WriteBits16(channels, 4) diff --git a/pkg/core/core.go b/pkg/core/core.go index e663ac75..10a435d3 100644 --- a/pkg/core/core.go +++ b/pkg/core/core.go @@ -47,7 +47,10 @@ type Producer interface { // GetTrack - return Receiver, that can only produce rtp.Packet(s) GetTrack(media *Media, codec *Codec) (*Receiver, error) + // Deprecated: rename to Run() Start() error + + // Deprecated: rename to Close() Stop() error } @@ -59,6 +62,7 @@ type Consumer interface { AddTrack(media *Media, codec *Codec, track *Receiver) error + // Deprecated: rename to Close() Stop() error } @@ -102,3 +106,60 @@ const ( UnsupportedCodec = "unsupported codec" WrongMediaDirection = "wrong media direction" ) + +type SuperProducer struct { + Type string `json:"type,omitempty"` + URL string `json:"url,omitempty"` + Medias []*Media `json:"medias,omitempty"` + Receivers []*Receiver `json:"receivers,omitempty"` + Recv int `json:"recv,omitempty"` +} + +func (s *SuperProducer) GetMedias() []*Media { + return s.Medias +} + +func (s *SuperProducer) GetTrack(media *Media, codec *Codec) (*Receiver, error) { + for _, receiver := range s.Receivers { + if receiver.Codec == codec { + return receiver, nil + } + } + receiver := NewReceiver(media, codec) + s.Receivers = append(s.Receivers, receiver) + return receiver, nil +} + +func (s *SuperProducer) Close() error { + for _, receiver := range s.Receivers { + receiver.Close() + } + return nil +} + +type SuperConsumer struct { + Type string `json:"type,omitempty"` + URL string `json:"url,omitempty"` + Medias []*Media `json:"medias,omitempty"` + Senders []*Sender `json:"receivers,omitempty"` + Send int `json:"recv,omitempty"` +} + +func (s *SuperConsumer) GetMedias() []*Media { + return s.Medias +} + +func (s *SuperConsumer) AddTrack(media *Media, codec *Codec, track *Receiver) error { + return nil +} + +//func (b *SuperConsumer) WriteTo(w io.Writer) (n int64, err error) { +// return 0, nil +//} + +func (s *SuperConsumer) Close() error { + for _, sender := range s.Senders { + sender.Close() + } + return nil +} diff --git a/pkg/core/readseeker.go b/pkg/core/readbuffer.go similarity index 75% rename from pkg/core/readseeker.go rename to pkg/core/readbuffer.go index 5eff128f..59161c83 100644 --- a/pkg/core/readseeker.go +++ b/pkg/core/readbuffer.go @@ -12,13 +12,13 @@ const ( BufferDrainAndClear = -1 ) -// ReadSeeker support buffering and Seek over buffer +// ReadBuffer support buffering and Seek over buffer // positive BufferSize will enable buffering mode // Seek to negative offset will clear buffer // Seek with a positive BufferSize will continue buffering after the last read from the buffer // Seek with a negative BufferSize will clear buffer after the last read from the buffer // Read more than BufferSize will raise error -type ReadSeeker struct { +type ReadBuffer struct { io.Reader BufferSize int @@ -27,14 +27,14 @@ type ReadSeeker struct { pos int } -func NewReadSeeker(rd io.Reader) *ReadSeeker { - if rs, ok := rd.(*ReadSeeker); ok { +func NewReadBuffer(rd io.Reader) *ReadBuffer { + if rs, ok := rd.(*ReadBuffer); ok { return rs } - return &ReadSeeker{Reader: rd} + return &ReadBuffer{Reader: rd} } -func (r *ReadSeeker) Read(p []byte) (n int, err error) { +func (r *ReadBuffer) Read(p []byte) (n int, err error) { // with zero buffer - read as usual if r.BufferSize == BufferDisable { return r.Reader.Read(p) @@ -65,7 +65,14 @@ func (r *ReadSeeker) Read(p []byte) (n int, err error) { return } -func (r *ReadSeeker) Seek(offset int64, whence int) (int64, error) { +func (r *ReadBuffer) Close() error { + if closer, ok := r.Reader.(io.Closer); ok { + return closer.Close() + } + return nil +} + +func (r *ReadBuffer) Seek(offset int64, whence int) (int64, error) { var pos int switch whence { case io.SeekStart: @@ -89,17 +96,17 @@ func (r *ReadSeeker) Seek(offset int64, whence int) (int64, error) { return int64(r.pos), nil } -func (r *ReadSeeker) Peek(n int) ([]byte, error) { +func (r *ReadBuffer) Peek(n int) ([]byte, error) { r.BufferSize = n b := make([]byte, n) if _, err := io.ReadAtLeast(r, b, n); err != nil { return nil, err } - r.Rewind() + r.Reset() return b, nil } -func (r *ReadSeeker) Rewind() { +func (r *ReadBuffer) Reset() { r.BufferSize = BufferDrainAndClear r.pos = 0 } diff --git a/pkg/core/readseeker_test.go b/pkg/core/readbuffer_test.go similarity index 97% rename from pkg/core/readseeker_test.go rename to pkg/core/readbuffer_test.go index 69080c52..01472170 100644 --- a/pkg/core/readseeker_test.go +++ b/pkg/core/readbuffer_test.go @@ -12,7 +12,7 @@ func TestReadSeeker(t *testing.T) { b := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9} buf := bytes.NewReader(b) - rd := NewReadSeeker(buf) + rd := NewReadBuffer(buf) rd.BufferSize = ProbeSize // 1. Read to buffer diff --git a/pkg/flv/client.go b/pkg/flv/client.go index 9fe18430..bb2c783f 100644 --- a/pkg/flv/client.go +++ b/pkg/flv/client.go @@ -18,7 +18,7 @@ const Signature = "FLV" type Client struct { URL string - rd *core.ReadSeeker + rd *core.ReadBuffer medias []*core.Media receivers []*core.Receiver @@ -30,7 +30,7 @@ type Client struct { func Open(rd io.Reader) (*Client, error) { client := &Client{ - rd: core.NewReadSeeker(rd), + rd: core.NewReadBuffer(rd), } if err := client.describe(); err != nil { return nil, err @@ -53,7 +53,7 @@ func (c *Client) describe() error { } c.rd.BufferSize = core.ProbeSize - defer c.rd.Rewind() + defer c.rd.Reset() // Normal software sends: // 1. Video/audio flag in header diff --git a/pkg/h264/annexb/annexb.go b/pkg/h264/annexb/annexb.go index b7e6bd5d..fcae18a4 100644 --- a/pkg/h264/annexb/annexb.go +++ b/pkg/h264/annexb/annexb.go @@ -70,8 +70,10 @@ func EncodeToAVCC(b []byte, safeAppend bool) []byte { return b } -func DecodeAVCC(b []byte) []byte { - b = bytes.Clone(b) +func DecodeAVCC(b []byte, safeClone bool) []byte { + if safeClone { + b = bytes.Clone(b) + } for i := 0; i < len(b); { size := int(binary.BigEndian.Uint32(b[i:])) b[i] = 0 diff --git a/pkg/magic/bitstream/client.go b/pkg/magic/bitstream/producer.go similarity index 54% rename from pkg/magic/bitstream/client.go rename to pkg/magic/bitstream/producer.go index 97bf7088..d6f7e72e 100644 --- a/pkg/magic/bitstream/client.go +++ b/pkg/magic/bitstream/producer.go @@ -2,7 +2,6 @@ package bitstream import ( "encoding/hex" - "encoding/json" "errors" "io" @@ -13,17 +12,13 @@ import ( "github.com/pion/rtp" ) -type Client struct { - rd *core.ReadSeeker - - media *core.Media - receiver *core.Receiver - - recv int +type Producer struct { + core.SuperProducer + rd *core.ReadBuffer } -func Open(r io.Reader) (*Client, error) { - rd := core.NewReadSeeker(r) +func Open(r io.Reader) (*Producer, error) { + rd := core.NewReadBuffer(r) buf, err := rd.Peek(256) if err != nil { @@ -43,30 +38,19 @@ func Open(r io.Reader) (*Client, error) { return nil, errors.New("bitstream: unsupported header: " + hex.EncodeToString(buf[:8])) } - client := &Client{ - rd: rd, - media: &core.Media{ + prod := &Producer{rd: rd} + prod.Type = "Bitstream producer" + prod.Medias = []*core.Media{ + { Kind: core.KindVideo, Direction: core.DirectionRecvonly, Codecs: []*core.Codec{codec}, }, } - - return client, nil + return prod, nil } -func (c *Client) GetMedias() []*core.Media { - return []*core.Media{c.media} -} - -func (c *Client) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver, error) { - if c.receiver == nil { - c.receiver = core.NewReceiver(media, codec) - } - return c.receiver, nil -} - -func (c *Client) Start() error { +func (c *Producer) Start() error { var buf []byte b := make([]byte, core.BufferSize) @@ -76,7 +60,7 @@ func (c *Client) Start() error { return err } - c.recv += n + c.Recv += n buf = append(buf, b[:n]...) @@ -89,7 +73,7 @@ func (c *Client) Start() error { Header: rtp.Header{Timestamp: core.Now90000()}, Payload: annexb.EncodeToAVCC(buf[:i], true), } - c.receiver.WriteRTP(pkt) + c.Receivers[0].WriteRTP(pkt) //log.Printf("[AVC] %v, len: %d", h264.Types(pkt.Payload), len(pkt.Payload)) @@ -97,22 +81,7 @@ func (c *Client) Start() error { } } -func (c *Client) Stop() error { - if c.receiver != nil { - c.receiver.Close() - } - if closer, ok := c.rd.Reader.(io.Closer); ok { - return closer.Close() - } - return nil -} - -func (c *Client) MarshalJSON() ([]byte, error) { - info := &core.Info{ - Type: "Bitstream active producer", - Medias: []*core.Media{c.media}, - Receivers: []*core.Receiver{c.receiver}, - Recv: c.recv, - } - return json.Marshal(info) +func (c *Producer) Stop() error { + _ = c.SuperProducer.Close() + return c.rd.Close() } diff --git a/pkg/magic/keyframe.go b/pkg/magic/keyframe.go index fb7e25aa..220f3a3d 100644 --- a/pkg/magic/keyframe.go +++ b/pkg/magic/keyframe.go @@ -43,7 +43,7 @@ func (k *Keyframe) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiv if !h264.IsKeyframe(packet.Payload) { return } - b := annexb.DecodeAVCC(packet.Payload) + b := annexb.DecodeAVCC(packet.Payload, true) k.Fire(b) } diff --git a/pkg/magic/mjpeg/client.go b/pkg/magic/mjpeg/producer.go similarity index 50% rename from pkg/magic/mjpeg/client.go rename to pkg/magic/mjpeg/producer.go index 4a5732a3..e5627fd7 100644 --- a/pkg/magic/mjpeg/client.go +++ b/pkg/magic/mjpeg/producer.go @@ -2,26 +2,22 @@ package mjpeg import ( "bytes" - "encoding/json" "io" "github.com/AlexxIT/go2rtc/pkg/core" "github.com/pion/rtp" ) -type Client struct { - rd *core.ReadSeeker - - media *core.Media - receiver *core.Receiver - - recv int +type Producer struct { + core.SuperProducer + rd *core.ReadBuffer } -func NewClient(rd io.Reader) *Client { - return &Client{ - rd: core.NewReadSeeker(rd), - media: &core.Media{ +func Open(rd io.Reader) (*Producer, error) { + prod := &Producer{rd: core.NewReadBuffer(rd)} + prod.Type = "MJPEG producer" + prod.Medias = []*core.Media{ + { Kind: core.KindVideo, Direction: core.DirectionRecvonly, Codecs: []*core.Codec{ @@ -33,20 +29,10 @@ func NewClient(rd io.Reader) *Client { }, }, } + return prod, nil } -func (c *Client) GetMedias() []*core.Media { - return []*core.Media{c.media} -} - -func (c *Client) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver, error) { - if c.receiver == nil { - c.receiver = core.NewReceiver(media, codec) - } - return c.receiver, nil -} - -func (c *Client) Start() error { +func (c *Producer) Start() error { var buf []byte // total bufer b := make([]byte, core.BufferSize) // reading buffer @@ -59,7 +45,7 @@ func (c *Client) Start() error { return err } - c.recv += n + c.Recv += n buf = append(buf, b[:n]...) @@ -77,7 +63,7 @@ func (c *Client) Start() error { Header: rtp.Header{Timestamp: core.Now90000()}, Payload: buf[:i], } - c.receiver.WriteRTP(pkt) + c.Receivers[0].WriteRTP(pkt) //log.Printf("[mjpeg] ts=%d size=%d", pkt.Header.Timestamp, len(pkt.Payload)) @@ -85,22 +71,7 @@ func (c *Client) Start() error { } } -func (c *Client) Stop() error { - if c.receiver != nil { - c.receiver.Close() - } - if closer, ok := c.rd.Reader.(io.Closer); ok { - return closer.Close() - } - return nil -} - -func (c *Client) MarshalJSON() ([]byte, error) { - info := &core.Info{ - Type: "MJPEG active producer", - Medias: []*core.Media{c.media}, - Receivers: []*core.Receiver{c.receiver}, - Recv: c.recv, - } - return json.Marshal(info) +func (c *Producer) Stop() error { + _ = c.SuperProducer.Close() + return c.rd.Close() } diff --git a/pkg/magic/magic.go b/pkg/magic/producer.go similarity index 92% rename from pkg/magic/magic.go rename to pkg/magic/producer.go index 4224171d..90a4b2a0 100644 --- a/pkg/magic/magic.go +++ b/pkg/magic/producer.go @@ -15,7 +15,7 @@ import ( ) func Open(r io.Reader) (core.Producer, error) { - rd := core.NewReadSeeker(r) + rd := core.NewReadBuffer(r) b, err := rd.Peek(4) if err != nil { @@ -27,7 +27,7 @@ func Open(r io.Reader) (core.Producer, error) { return bitstream.Open(rd) case bytes.HasPrefix(b, []byte{0xFF, 0xD8}): - return mjpeg.NewClient(rd), nil + return mjpeg.Open(rd) case bytes.HasPrefix(b, []byte(flv.Signature)): return flv.Open(rd) diff --git a/pkg/mpegts/client.go b/pkg/mpegts/client.go index 520c69ca..6d925276 100644 --- a/pkg/mpegts/client.go +++ b/pkg/mpegts/client.go @@ -14,7 +14,7 @@ import ( type Client struct { URL string - rd *core.ReadSeeker + rd *core.ReadBuffer medias []*core.Media receivers []*core.Receiver @@ -23,7 +23,7 @@ type Client struct { } func Open(rd io.Reader) (*Client, error) { - client := &Client{rd: core.NewReadSeeker(rd)} + client := &Client{rd: core.NewReadBuffer(rd)} if err := client.describe(); err != nil { return nil, err } @@ -32,7 +32,7 @@ func Open(rd io.Reader) (*Client, error) { func (c *Client) describe() error { c.rd.BufferSize = core.ProbeSize - defer c.rd.Rewind() + defer c.rd.Reset() rd := NewReader()