Merge pull request #230 from jaesung9507/rtmps_listen

[feat] rtmp: server support rtmps

conf/lalserver.conf.json

@@ -4,6 +4,10 @@
   "rtmp": {
     "enable": true,
     "addr": ":1935",
+    "rtmps_enable": true,
+    "rtmps_addr": ":4935",
+    "rtmps_cert_file": "./conf/cert.pem",
+    "rtmps_key_file": "./conf/key.pem",
     "gop_num": 0,
     "merge_write_size": 0,
     "add_dummy_audio_enable": false,

pkg/logic/config.go

@@ -53,6 +53,10 @@ type Config struct {
 type RtmpConfig struct {
 	Enable                   bool   `json:"enable"`
 	Addr                     string `json:"addr"`
+	RtmpsEnable              bool   `json:"rtmps_enable"`
+	RtmpsAddr                string `json:"rtmps_addr"`
+	RtmpsCertFile            string `json:"rtmps_cert_file"`
+	RtmpsKeyFile             string `json:"rtmps_key_file"`
 	GopNum                   int    `json:"gop_num"` // TODO(chef): refactor 更名为gop_cache_num
 	MergeWriteSize           int    `json:"merge_write_size"`
 	AddDummyAudioEnable      bool   `json:"add_dummy_audio_enable"`

pkg/logic/group__core_streaming.go

@@ -9,9 +9,10 @@
 package logic
 
 import (
+	"net"
+
 	"github.com/q191201771/lal/pkg/rtmp"
 	"github.com/q191201771/naza/pkg/nazalog"
-	"net"
 
 	"github.com/q191201771/lal/pkg/mpegts"
 
@@ -370,7 +371,7 @@ func (group *Group) broadcastByRtmpMsg(msg base.RtmpMsg) {
 	}
 
 	// # 缓存关键信息，以及gop
-	if group.config.RtmpConfig.Enable {
+	if group.config.RtmpConfig.Enable || group.config.RtmpConfig.RtmpsEnable {
 		group.rtmpGopCache.Feed(msg, lazyRtmpChunkDivider.GetEnsureWithoutSdf())
 		if msg.Header.MsgTypeId == base.RtmpTypeIdMetadata {
 			group.rtmpGopCache.SetMetadata(lazyRtmpChunkDivider.GetEnsureWithSdf(), lazyRtmpChunkDivider.GetEnsureWithoutSdf())

pkg/logic/server_manager__.go

@@ -46,6 +46,7 @@ type ServerManager struct {
 	hlsServerHandler  *hls.ServerHandler
 
 	rtmpServer    *rtmp.Server
+	rtmpsServer   *rtmp.Server
 	rtspServer    *rtsp.Server
 	httpApiServer *HttpApiServer
 	pprofServer   *http.Server
@@ -123,6 +124,9 @@ Doc: %s
 	if sm.config.RtmpConfig.Enable {
 		sm.rtmpServer = rtmp.NewServer(sm.config.RtmpConfig.Addr, sm)
 	}
+	if sm.config.RtmpConfig.RtmpsEnable {
+		sm.rtmpsServer = rtmp.NewServer(sm.config.RtmpConfig.RtmpsAddr, sm)
+	}
 	if sm.config.RtspConfig.Enable {
 		sm.rtspServer = rtsp.NewServer(sm.config.RtspConfig.Addr, sm, sm.config.RtspConfig.ServerAuthConfig)
 	}
@@ -220,6 +224,17 @@ func (sm *ServerManager) RunLoop() error {
 		}()
 	}
 
+	if sm.rtmpsServer != nil {
+		if err := sm.rtmpsServer.ListenWithTLS(sm.config.RtmpConfig.RtmpsCertFile, sm.config.RtmpConfig.RtmpsKeyFile); err != nil {
+			return err
+		}
+		go func() {
+			if err := sm.rtmpsServer.RunLoop(); err != nil {
+				Log.Error(err)
+			}
+		}()
+	}
+
 	if sm.rtspServer != nil {
 		if err := sm.rtspServer.Listen(); err != nil {
 			return err

pkg/rtmp/server.go

@@ -9,8 +9,10 @@
 package rtmp
 
 import (
-	"github.com/q191201771/lal/pkg/base"
+	"crypto/tls"
 	"net"
+
+	"github.com/q191201771/lal/pkg/base"
 )
 
 type IServerObserver interface {
@@ -55,6 +57,19 @@ func (server *Server) Listen() (err error) {
 	return
 }
 
+func (server *Server) ListenWithTLS(certFile, keyFile string) (err error) {
+	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+	if err != nil {
+		return
+	}
+	tlsConfig := &tls.Config{Certificates: []tls.Certificate{cert}}
+	if server.ln, err = tls.Listen("tcp", server.addr, tlsConfig); err != nil {
+		return
+	}
+	Log.Infof("start rtmps server listen. addr=%s", server.addr)
+	return
+}
+
 func (server *Server) RunLoop() error {
 	for {
 		conn, err := server.ln.Accept()