Simplify MJPEG consumer

internal/mjpeg/init.go

@@ -96,10 +96,9 @@ func outputMjpeg(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	cons := &mjpeg.Consumer{
+	cons := mjpeg.NewConsumer()
-		RemoteAddr: tcp.RemoteAddr(r),
+	cons.RemoteAddr = tcp.RemoteAddr(r)
-		UserAgent:  r.UserAgent(),
+	cons.UserAgent = r.UserAgent()
-	}
 
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Error().Err(err).Msg("[api.mjpeg] add consumer")
@@ -168,10 +167,9 @@ func handlerWS(tr *ws.Transport, _ *ws.Message) error {
 		return errors.New(api.StreamNotFound)
 	}
 
-	cons := &mjpeg.Consumer{
+	cons := mjpeg.NewConsumer()
-		RemoteAddr: tcp.RemoteAddr(tr.Request),
+	cons.RemoteAddr = tcp.RemoteAddr(tr.Request)
-		UserAgent:  tr.Request.UserAgent(),
+	cons.UserAgent = tr.Request.UserAgent()
-	}
 
 	if err := stream.AddConsumer(cons); err != nil {
 		log.Error().Err(err).Caller().Send()

pkg/mjpeg/consumer.go

@@ -1,7 +1,6 @@
 package mjpeg
 
 import (
-	"encoding/json"
 	"io"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
@@ -9,51 +8,42 @@ import (
 )
 
 type Consumer struct {
-	UserAgent  string
+	core.SuperConsumer
-	RemoteAddr string
-
-	medias []*core.Media
-	sender *core.Sender
-
 	wr *core.WriteBuffer
-
-	send int
 }
 
-func (c *Consumer) GetMedias() []*core.Media {
+func NewConsumer() *Consumer {
-	if c.medias == nil {
+	return &Consumer{
-		c.medias = []*core.Media{
+		core.SuperConsumer{
-			{
+			Type: "MJPEG passive consumer",
-				Kind:      core.KindVideo,
+			Medias: []*core.Media{
-				Direction: core.DirectionSendonly,
+				{
-				Codecs: []*core.Codec{
+					Kind:      core.KindVideo,
-					{Name: core.CodecJPEG},
+					Direction: core.DirectionSendonly,
+					Codecs: []*core.Codec{
+						{Name: core.CodecJPEG},
+					},
 				},
 			},
-		}
+		},
+		core.NewWriteBuffer(nil),
 	}
-	return c.medias
 }
 
 func (c *Consumer) AddTrack(media *core.Media, _ *core.Codec, track *core.Receiver) error {
-	if c.wr == nil {
+	sender := core.NewSender(media, track.Codec)
-		c.wr = core.NewWriteBuffer(nil)
+	sender.Handler = func(packet *rtp.Packet) {
-	}
+		if n, err := c.wr.Write(packet.Payload); err == nil {
-
+			c.Send += n
-	if c.sender == nil {
-		c.sender = core.NewSender(media, track.Codec)
-		c.sender.Handler = func(packet *rtp.Packet) {
-			if n, err := c.wr.Write(packet.Payload); err == nil {
-				c.send += n
-			}
-		}
-
-		if track.Codec.IsRTP() {
-			c.sender.Handler = RTPDepay(c.sender.Handler)
 		}
 	}
 
-	c.sender.HandleRTP(track)
+	if track.Codec.IsRTP() {
+		sender.Handler = RTPDepay(sender.Handler)
+	}
+
+	sender.HandleRTP(track)
+	c.Senders = append(c.Senders, sender)
 	return nil
 }
 
@@ -62,25 +52,6 @@ func (c *Consumer) WriteTo(wr io.Writer) (int64, error) {
 }
 
 func (c *Consumer) Stop() error {
-	if c.sender != nil {
+	_ = c.SuperConsumer.Close()
-		c.sender.Close()
+	return c.wr.Close()
-	}
-	if c.wr != nil {
-		_ = c.wr.Close()
-	}
-	return nil
-}
-
-func (c *Consumer) MarshalJSON() ([]byte, error) {
-	info := &core.Info{
-		Type:       "MJPEG passive consumer",
-		RemoteAddr: c.RemoteAddr,
-		UserAgent:  c.UserAgent,
-		Medias:     c.medias,
-		Send:       c.send,
-	}
-	if c.sender != nil {
-		info.Senders = []*core.Sender{c.sender}
-	}
-	return json.Marshal(info)
 }