add rtmp broadcast

.m1k1o/base/Dockerfile

@@ -8,7 +8,7 @@ WORKDIR /src
 # install dependencies
 RUN set -eux; apt-get update; \
     apt-get install -y --no-install-recommends git cmake make libx11-dev libxrandr-dev libxtst-dev \
-    libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev gstreamer1.0-plugins-base gstreamer1.0-plugins-good; \
+    libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev gstreamer1.0-plugins-base gstreamer1.0-plugins-good gstreamer1.0-plugins-bad gstreamer1.0-plugins-ugly; \
     #
     # install libclipboard
     set -eux; \
@@ -69,7 +69,7 @@ RUN set -eux; apt-get update; \
     #
     # gst
     apt-get install -y --no-install-recommends libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev \
-                    gstreamer1.0-plugins-base gstreamer1.0-plugins-good gstreamer1.0-pulseaudio; \
+                    gstreamer1.0-plugins-base gstreamer1.0-plugins-good gstreamer1.0-plugins-bad gstreamer1.0-plugins-ugly gstreamer1.0-pulseaudio; \
     #
     # create a non-root user
     groupadd --gid $USER_GID $USERNAME; \

docker-compose.dev.yaml

@@ -12,9 +12,12 @@ services:
       - SYS_ADMIN
     environment:
       DISPLAY: :99.0
-      NEKO_SCREEN: '1280x720@30'
+      NEKO_SCREEN: '1920x1080@30'
       NEKO_PASSWORD: neko
       NEKO_PASSWORD_ADMIN: admin
       NEKO_BIND: :8080
       NEKO_EPR: 52000-52010
       NEKO_NAT1TO1: 192.168.1.20
+      NEKO_BROADCAST: 'true'
+      NEKO_RTMP: 'rtmp://192.168.1.20/live/neko'
+

server/cmd/serve.go

@@ -20,6 +20,7 @@ func init() {
 		neko.Service.Server,
 		neko.Service.WebRTC,
 		neko.Service.Remote,
+		neko.Service.Broadcast,
 		neko.Service.WebSocket,
 	}
 

server/internal/gst/gst.c

@@ -84,6 +84,10 @@ void gstreamer_send_start_pipeline(GstElement *pipeline, int pipelineId) {
   gst_element_set_state(pipeline, GST_STATE_PLAYING);
 }
 
+void gstreamer_send_play_pipeline(GstElement *pipeline) {
+  gst_element_set_state(pipeline, GST_STATE_PLAYING);
+}
+
 void gstreamer_send_stop_pipeline(GstElement *pipeline) {
   gst_element_set_state(pipeline, GST_STATE_NULL);
 }

server/internal/gst/gst.go

@@ -68,7 +68,8 @@ func init() {
 func CreateRTMPPipeline(pipelineDevice string, pipelineDisplay string, pipelineRTMP string) (*Pipeline, error) {
 	video := fmt.Sprintf(videoSrc, pipelineDisplay)
 	audio := fmt.Sprintf(audioSrc, pipelineDevice)
-	return CreatePipeline(fmt.Sprintf("%s ! x264enc ! flv. ! %s ! faac ! flv. ! flvmux name='flv' ! rtmpsink location='%s'", video, audio, pipelineRTMP), "", 0)
+
+	return CreatePipeline(fmt.Sprintf("flvmux name=mux ! rtmpsink location='%s' live=1 %s voaacenc ! mux. %s x264enc bframes=0 key-int-max=60 byte-stream=true tune=zerolatency speed-preset=veryfast ! mux.", pipelineRTMP, audio, video), "", 0)
 }
 
 // CreateAppPipeline creates a GStreamer Pipeline
@@ -228,6 +229,11 @@ func (p *Pipeline) Start() {
 	C.gstreamer_send_start_pipeline(p.Pipeline, C.int(p.id))
 }
 
+// Play starts the GStreamer Pipeline
+func (p *Pipeline) Play() {
+	C.gstreamer_send_play_pipeline(p.Pipeline)
+}
+
 // Stop stops the GStreamer Pipeline
 func (p *Pipeline) Stop() {
 	C.gstreamer_send_stop_pipeline(p.Pipeline)

server/internal/gst/gst.h

@@ -11,6 +11,7 @@ extern void goHandlePipelineBuffer(void *buffer, int bufferLen, int samples, int
 GstElement *gstreamer_send_create_pipeline(char *pipeline);
 
 void gstreamer_send_start_pipeline(GstElement *pipeline, int pipelineId);
+void gstreamer_send_play_pipeline(GstElement *pipeline);
 void gstreamer_send_stop_pipeline(GstElement *pipeline);
 void gstreamer_send_start_mainloop(void);
 void gstreamer_init(void);

server/internal/remote/manager.go

@@ -17,20 +17,23 @@ type RemoteManager struct {
 	logger    zerolog.Logger
 	video     *gst.Pipeline
 	audio     *gst.Pipeline
+	rtmp      *gst.Pipeline
 	config    *config.Remote
+	broadcast *config.Broadcast
 	cleanup   *time.Ticker
 	shutdown  chan bool
 	emmiter   events.EventEmmiter
 	streaming bool
 }
 
-func New(config *config.Remote) *RemoteManager {
+func New(config *config.Remote, broadcast *config.Broadcast) *RemoteManager {
 	return &RemoteManager{
 		logger:    log.With().Str("module", "remote").Logger(),
 		cleanup:   time.NewTicker(1 * time.Second),
 		shutdown:  make(chan bool),
 		emmiter:   events.New(),
 		config:    config,
+		broadcast: broadcast,
 		streaming: false,
 	}
 }
@@ -70,6 +73,11 @@ func (manager *RemoteManager) Shutdown() error {
 	manager.logger.Info().Msgf("remote shutting down")
 	manager.video.Stop()
 	manager.audio.Stop()
+
+	if manager.broadcast.Enabled {
+		manager.rtmp.Stop()
+	}
+
 	manager.cleanup.Stop()
 	manager.shutdown <- true
 	return nil
@@ -98,6 +106,12 @@ func (manager *RemoteManager) StartStream() {
 		Str("screen_resolution", fmt.Sprintf("%dx%d@%d", manager.config.ScreenWidth, manager.config.ScreenHeight, manager.config.ScreenRate)).
 		Msgf("Pipelines starting...")
 
+	if manager.broadcast.Enabled {
+		manager.logger.Info().
+			Str("rtmp_pipeline_src", manager.rtmp.Src).
+			Msgf("Prtmp pipeline is starting...")
+	}
+
 	xorg.Display(manager.config.Display)
 
 	if !xorg.ValidScreenSize(manager.config.ScreenWidth, manager.config.ScreenHeight, manager.config.ScreenRate) {
@@ -109,6 +123,11 @@ func (manager *RemoteManager) StartStream() {
 	manager.createPipelines()
 	manager.video.Start()
 	manager.audio.Start()
+
+	if manager.broadcast.Enabled {
+		manager.rtmp.Play()
+	}
+
 	manager.streaming = true
 }
 
@@ -116,6 +135,11 @@ func (manager *RemoteManager) StopStream() {
 	manager.logger.Info().Msgf("Pipelines shutting down...")
 	manager.video.Stop()
 	manager.audio.Stop()
+
+	if manager.broadcast.Enabled {
+		manager.rtmp.Stop()
+	}
+
 	manager.streaming = false
 }
 
@@ -140,7 +164,18 @@ func (manager *RemoteManager) createPipelines() {
 		manager.config.AudioParams,
 	)
 	if err != nil {
-		manager.logger.Panic().Err(err).Msg("unable to screate audio pipeline")
+		manager.logger.Panic().Err(err).Msg("unable to create audio pipeline")
+	}
+
+	if manager.broadcast.Enabled {
+		manager.rtmp, err = gst.CreateRTMPPipeline(
+			manager.config.Device,
+			manager.config.Display,
+			manager.broadcast.RTMP,
+		)
+		if err != nil {
+			manager.logger.Panic().Err(err).Msg("unable to create rtmp pipeline")
+		}
 	}
 }
 
@@ -150,8 +185,18 @@ func (manager *RemoteManager) ChangeResolution(width int, height int, rate int)
 	}
 
 	manager.video.Stop()
+
+	if manager.broadcast.Enabled {
+		manager.rtmp.Stop()
+	}
+
 	defer func() {
 		manager.video.Start()
+
+		if manager.broadcast.Enabled {
+			manager.rtmp.Play()
+		}
+
 		manager.logger.Info().Msg("starting video pipeline...")
 	}()
 
@@ -159,17 +204,26 @@ func (manager *RemoteManager) ChangeResolution(width int, height int, rate int)
 		return err
 	}
 
-	video, err := gst.CreateAppPipeline(
+	manager.video, err := gst.CreateAppPipeline(
 		manager.config.VideoCodec,
 		manager.config.Display,
 		manager.config.VideoParams,
 	)
-
 	if err != nil {
 		manager.logger.Panic().Err(err).Msg("unable to create new video pipeline")
 	}
 
-	manager.video = video
+	if manager.broadcast.Enabled {
+		manager.rtmp, err = gst.CreateRTMPPipeline(
+			manager.config.Device,
+			manager.config.Display,
+			manager.broadcast.RTMP,
+		)
+		if err != nil {
+			manager.logger.Panic().Err(err).Msg("unable to create new rtmp pipeline")
+		}
+	}
+
 	return nil
 }
 

server/internal/types/config/broadcast.go

@@ -7,10 +7,10 @@ import (
 
 type Broadcast struct {
 	Enabled     bool
-	Display     string
+	// Display     string
-	Device      string
+	// Device      string
-	AudioParams string
+	// AudioParams string
-	VideoParams string
+	// VideoParams string
 	RTMP        string
 }
 
@@ -40,9 +40,9 @@ func (Broadcast) Init(cmd *cobra.Command) error {
 
 func (s *Broadcast) Set() {
 	s.Enabled = viper.GetBool("broadcast")
-	s.Display = viper.GetString("display")
+	// s.Display = viper.GetString("display")
-	s.Device = viper.GetString("device")
+	// s.Device = viper.GetString("device")
-	s.AudioParams = viper.GetString("cast_audio")
+	// s.AudioParams = viper.GetString("cast_audio")
-	s.VideoParams = viper.GetString("cast_video")
+	// s.VideoParams = viper.GetString("cast_video")
 	s.RTMP = viper.GetString("rtmp")
 }

server/neko.go

@@ -61,6 +61,7 @@ func init() {
 		Root:      &config.Root{},
 		Server:    &config.Server{},
 		Remote:    &config.Remote{},
+		Broadcast: &config.Broadcast{},
 		WebRTC:    &config.WebRTC{},
 		WebSocket: &config.WebSocket{},
 	}
@@ -99,6 +100,7 @@ type Neko struct {
 	Version   *Version
 	Root      *config.Root
 	Remote    *config.Remote
+	Broadcast *config.Broadcast
 	Server    *config.Server
 	WebRTC    *config.WebRTC
 	WebSocket *config.WebSocket
@@ -117,7 +119,7 @@ func (neko *Neko) Preflight() {
 
 func (neko *Neko) Start() {
 
-	remoteManager := remote.New(neko.Remote)
+	remoteManager := remote.New(neko.Remote, neko.Broadcast)
 	remoteManager.Start()
 
 	sessionManager := session.New(remoteManager)