First commit

config.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"crypto/rand"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+
+	"github.com/deepch/vdk/av"
+)
+
+var Config = loadConfig()
+
+type ConfigST struct {
+	Server  ServerST            `json:"server"`
+	Streams map[string]StreamST `json:"streams"`
+}
+
+type ServerST struct {
+	HTTPPort string `json:"http_port"`
+}
+
+type StreamST struct {
+	URL    string `json:"url"`
+	Status bool   `json:"status"`
+	Codecs []av.CodecData
+	Cl     map[string]viwer
+}
+type viwer struct {
+	c chan av.Packet
+}
+
+func loadConfig() *ConfigST {
+	var tmp ConfigST
+	data, err := ioutil.ReadFile("config.json")
+	if err != nil {
+		log.Fatalln(err)
+	}
+	err = json.Unmarshal(data, &tmp)
+	if err != nil {
+		log.Fatalln(err)
+	}
+	for i, v := range tmp.Streams {
+		v.Cl = make(map[string]viwer)
+		tmp.Streams[i] = v
+	}
+	return &tmp
+}
+
+func (element *ConfigST) cast(uuid string, pck av.Packet) {
+	for _, v := range element.Streams[uuid].Cl {
+		if len(v.c) < cap(v.c) {
+			v.c <- pck
+		}
+	}
+}
+
+func (element *ConfigST) ext(suuid string) bool {
+	_, ok := element.Streams[suuid]
+	return ok
+}
+
+func (element *ConfigST) coAd(suuid string, codecs []av.CodecData) {
+	t := element.Streams[suuid]
+	t.Codecs = codecs
+	element.Streams[suuid] = t
+}
+
+func (element *ConfigST) coGe(suuid string) []av.CodecData {
+	return element.Streams[suuid].Codecs
+}
+
+func (element *ConfigST) clAd(suuid string) (string, chan av.Packet) {
+	cuuid := pseudoUUID()
+	ch := make(chan av.Packet, 100)
+	element.Streams[suuid].Cl[cuuid] = viwer{c: ch}
+	return cuuid, ch
+}
+
+func (element *ConfigST) list() (string, []string) {
+	var res []string
+	var fist string
+	for k := range element.Streams {
+		if fist == "" {
+			fist = k
+		}
+		res = append(res, k)
+	}
+	return fist, res
+}
+func (element *ConfigST) clDe(uuid string) {
+	delete(element.Streams, uuid)
+}
+
+func pseudoUUID() (uuid string) {
+	b := make([]byte, 16)
+	_, err := rand.Read(b)
+	if err != nil {
+		fmt.Println("Error: ", err)
+		return
+	}
+	uuid = fmt.Sprintf("%X-%X-%X-%X-%X", b[0:4], b[4:6], b[6:8], b[8:10], b[10:])
+	return
+}

config.json

@@ -0,0 +1,10 @@
+{
+  "server": {
+    "http_port": ":8083"
+  },
+  "streams": {
+    "demo1": {
+      "url": "rtsp://170.93.143.139/rtplive/470011e600ef003a004ee33696235daa"
+    }
+  }
+}

http.go

@@ -0,0 +1,91 @@
+package main
+
+import (
+	"log"
+	"net/http"
+	"sort"
+	"time"
+
+	"github.com/deepch/vdk/format/mp4f"
+	"github.com/gin-gonic/gin"
+	"golang.org/x/net/websocket"
+)
+
+func serveHTTP() {
+	router := gin.Default()
+	router.LoadHTMLGlob("web/templates/*")
+	router.GET("/", func(c *gin.Context) {
+		fi, all := Config.list()
+		sort.Strings(all)
+		c.HTML(http.StatusOK, "index.tmpl", gin.H{
+			"port":     Config.Server.HTTPPort,
+			"suuid":    fi,
+			"suuidMap": all,
+		})
+	})
+	router.GET("/player/:suuid", func(c *gin.Context) {
+		_, all := Config.list()
+		sort.Strings(all)
+		c.HTML(http.StatusOK, "index.tmpl", gin.H{
+			"port":     Config.Server.HTTPPort,
+			"suuid":    c.Param("suuid"),
+			"suuidMap": all,
+		})
+	})
+	router.GET("/ws/:suuid", func(c *gin.Context) {
+		handler := websocket.Handler(ws)
+		handler.ServeHTTP(c.Writer, c.Request)
+	})
+	router.StaticFS("/static", http.Dir("web/static"))
+	err := router.Run(Config.Server.HTTPPort)
+	if err != nil {
+		log.Fatalln(err)
+	}
+}
+func ws(ws *websocket.Conn) {
+	defer ws.Close()
+	suuid := ws.Request().FormValue("suuid")
+	log.Println("Request", suuid)
+	if Config.ext(suuid) {
+		ws.SetWriteDeadline(time.Now().Add(5 * time.Second))
+		suuid := ws.Request().FormValue("suuid")
+		cuuid, ch := Config.clAd(suuid)
+		defer Config.clDe(cuuid)
+		codecs := Config.coGe(suuid)
+		if codecs == nil {
+			log.Println("No Codec Info")
+			return
+		}
+		muxer := mp4f.NewMuxer(nil)
+		muxer.WriteHeader(codecs)
+		meta, init := muxer.GetInit(codecs)
+		err := websocket.Message.Send(ws, append([]byte{9}, meta...))
+		if err != nil {
+			return
+		}
+		err = websocket.Message.Send(ws, init)
+		if err != nil {
+			return
+		}
+		var start bool
+		for {
+			select {
+			case pck := <-ch:
+				if pck.IsKeyFrame {
+					start = true
+				}
+				if !start {
+					continue
+				}
+				ready, buf, _ := muxer.WritePacket(pck, false)
+				if ready {
+					ws.SetWriteDeadline(time.Now().Add(10 * time.Second))
+					err := websocket.Message.Send(ws, buf)
+					if err != nil {
+						return
+					}
+				}
+			}
+		}
+	}
+}

main.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	go serveHTTP()
+	go serveStreams()
+	sigs := make(chan os.Signal, 1)
+	done := make(chan bool, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		sig := <-sigs
+		log.Println(sig)
+		done <- true
+	}()
+	log.Println("Server Start Awaiting Signal")
+	<-done
+	log.Println("Exiting")
+}

stream.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"log"
+	"time"
+
+	"github.com/deepch/vdk/format/rtsp"
+)
+
+func serveStreams() {
+	for k, v := range Config.Streams {
+		go func(name, url string) {
+			for {
+				log.Println(name, "connect", url)
+				rtsp.DebugRtsp = true
+				session, err := rtsp.Dial(url)
+				if err != nil {
+					log.Println(name, err)
+					time.Sleep(5 * time.Second)
+					continue
+				}
+				session.RtpKeepAliveTimeout = time.Duration(10 * time.Second)
+				if err != nil {
+					log.Println(name, err)
+					time.Sleep(5 * time.Second)
+					continue
+				}
+				codec, err := session.Streams()
+				if err != nil {
+					log.Println(name, err)
+					time.Sleep(5 * time.Second)
+					continue
+				}
+				Config.coAd(name, codec)
+				for {
+					pkt, err := session.ReadPacket()
+					if err != nil {
+						log.Println(name, err)
+						break
+					}
+					Config.cast(name, pkt)
+				}
+				session.Close()
+				log.Println(name, "reconnect wait 5s")
+				time.Sleep(5 * time.Second)
+			}
+		}(k, v.URL)
+	}
+}