init

README.md

@@ -1,5 +1,5 @@
 # rtmp2ps
-rtmp推流转推rtp+ps，用于测试gb28181的媒体服务器，支持Http API触发转推TCP/UDP
+rtmp推流转推rtp+ps，用于测试gb28181的媒体服务器，支持tcp/udp
 
 Input:rtmp
 

config.json

@@ -0,0 +1,8 @@
+{
+    "rtmpsvr_config": {
+        "listen_addr": ":1935"
+    },
+    "mediasvr_config": {
+        "forward_addr": "100.100.100.100:6000"
+    }
+}

config/config.go

@@ -0,0 +1,17 @@
+package config
+
+type Config struct {
+	RtmpSvrConfig RtmpSvrConfig `json:"rtmpsvr_config"`
+	MediaSvrConfig MediaSvrConfig `json:"mediasvr_config"`
+}
+
+// rtmp服务配置
+type RtmpSvrConfig struct {
+	ListenAddr string `json:"listen_addr"`
+}
+
+// 转推媒体服务配置
+type MediaSvrConfig struct {
+	ForwardAddr string `json:"forward_addr"` // 转推媒体服务器地址
+	Network string `json:"network"` // tcp/udp
+}

forward/conn.go

@@ -0,0 +1,19 @@
+package forward
+
+import "rtmp2ps/config"
+
+type Conn struct {
+	conf config.MediaSvrConfig
+}
+
+func NewConn(conf config.MediaSvrConfig) *Conn {
+	return &Conn{}
+}
+
+func (c *Conn) StartForward() error {
+	for {
+
+	}
+
+	return nil
+}

go.mod

@@ -0,0 +1,9 @@
+module rtmp2ps
+
+go 1.20
+
+require (
+	github.com/deepch/vdk v0.0.27 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	golang.org/x/sys v0.9.0 // indirect
+)

go.sum

@@ -0,0 +1,15 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/deepch/vdk v0.0.27 h1:j/SHaTiZhA47wRpaue8NRp7P9xwOOO/lunxrDJBwcao=
+github.com/deepch/vdk v0.0.27/go.mod h1:JlgGyR2ld6+xOIHa7XAxJh+stSDBAkdNvIPkUIdIywk=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.9.0 h1:KS/R3tvhPqvJvwcKfnBHJwwthS11LRhmM5D59eEXa0s=
+golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

logger/logger.go

@@ -0,0 +1,43 @@
+package logger
+
+import (
+	"github.com/sirupsen/logrus"
+	"os"
+)
+var log *logrus.Logger
+
+func LogInit(logLevel string) {
+	log = logrus.New()
+
+	level := logrus.DebugLevel
+	log.SetLevel(logrus.ErrorLevel)
+	switch {
+	case logLevel == "debug":
+		level = logrus.DebugLevel
+		case logLevel == "info":
+			level = logrus.InfoLevel
+			case logLevel == "error":
+				level = logrus.ErrorLevel
+				default:
+					level = logrus.DebugLevel
+	}
+	log.Formatter = &logrus.JSONFormatter{
+		TimestampFormat: "2006-01-02 15:04:05",
+	}
+	log.SetOutput(os.Stdout)
+	log.SetLevel(level)
+}
+
+func Info(v ...interface{}) {
+	log.Info(v)
+}
+
+func Error(v ...interface{}) {
+	log.Error(v)
+}
+
+func Debug(v ...interface{}) {
+	log.Debug(v)
+}
+
+

main.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"rtmp2ps/config"
+	"rtmp2ps/forward"
+	log "rtmp2ps/logger"
+	"rtmp2ps/rtmp"
+)
+
+func main() {
+	log.LogInit("debug")
+
+	configFile, err := os.Open("config.json")
+	if err != nil {
+		panic(err)
+		return
+	}
+
+	defer configFile.Close()
+
+	buf, _ := ioutil.ReadAll(configFile)
+
+	var conf config.Config
+	err = json.Unmarshal(buf, &conf)
+	if err != nil {
+		log.Error("json Unmarshal failed, err:", err)
+		return 
+	}
+
+	if conf.RtmpSvrConfig.ListenAddr != "" {
+		svr := rtmp.NewRtmpSvr(conf.RtmpSvrConfig.ListenAddr)
+		go func() {
+			err = svr.Start()
+			if err != nil {
+				log.Error("rtmp svr start failed, err:", err)
+			}
+		}()
+	}
+
+	if conf.MediaSvrConfig.ForwardAddr != "" {
+		conn := forward.NewConn(conf.MediaSvrConfig)
+		err = conn.StartForward()
+		if err != nil {
+			log.Error("forward failed, err:", err)
+		}
+	}
+
+}

rtmp/server.go

@@ -0,0 +1,26 @@
+package rtmp
+
+import (
+	log "rtmp2ps/logger"
+	vdkrtmp "github.com/deepch/vdk/format/rtmp"
+)
+type RtmpSvr struct {
+	listenAddr string
+	svr *vdkrtmp.Server
+}
+
+func NewRtmpSvr(listenAddr string) *RtmpSvr{
+	return &RtmpSvr{
+		listenAddr: listenAddr,
+		svr: &vdkrtmp.Server{
+			Addr: listenAddr,
+		},
+	}
+}
+
+func (s *RtmpSvr) Start() error {
+	log.Info("start rtmp svr, listenaddr:", s.listenAddr)
+
+	s.svr.ListenAndServe()
+	return nil
+}