Merge pull request #5 from LdDl/app-conf

App conf

README.md

@@ -80,8 +80,7 @@ Paste link to the browser and check if video loaded successfully.
 ## Dependencies
 GIN web-framework - [https://github.com/gin-gonic/gin](https://github.com/gin-gonic/gin). License is [MIT](https://github.com/gin-gonic/gin/blob/master/LICENSE)
 
-Media library - [github.com/morozka/vdk](https://github.com/morozka/vdk). License is [MIT](https://github.com/morozka/vdk/blob/master/LICENSE)
-
+Media library - [http://github.com/deepch/vdk](https://github.com/deepch/vdk). License is [MIT](https://github.com/deepch/vdk/blob/master/LICENSE). We are using fork actually - https://github.com/morozka/vdk
 
 UUID generation and parsing - [https://github.com/google/uuid](https://github.com/google/uuid). License is [BSD 3-Clause](https://github.com/google/uuid/blob/master/LICENSE)
 

application.go

@@ -0,0 +1,175 @@
+package videoserver
+
+import (
+	"log"
+	"sync"
+
+	"github.com/google/uuid"
+	"github.com/morozka/vdk/av"
+)
+
+// Application Configuration parameters for application
+type Application struct {
+	Server          *ServerInfo `json:"server"`
+	Streams         StreamsMap  `json:"streams"`
+	HlsMsPerSegment int64       `json:"hls_ms_per_segment"`
+	HlsDirectory    string      `json:"hls_directory"`
+	HlsWindowSize   uint        `json:"hls_window_size"`
+	HlsCapacity     uint        `json:"hls_window_capacity"`
+}
+
+// ServerInfo Information about server
+type ServerInfo struct {
+	HTTPAddr string `json:"http_addr"`
+	HTTPPort int    `json:"http_port"`
+}
+
+// StreamsMap Map wrapper for map[uuid.UUID]*StreamConfiguration with mutex for concurrent usage
+type StreamsMap struct {
+	sync.Mutex
+	Streams map[uuid.UUID]*StreamConfiguration
+}
+
+func (sm *StreamsMap) getKeys() []uuid.UUID {
+	sm.Lock()
+	defer sm.Unlock()
+	keys := make([]uuid.UUID, 0, len(sm.Streams))
+	for k := range sm.Streams {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+// StreamConfiguration Configuration parameters for stream
+type StreamConfiguration struct {
+	URL       string `json:"url"`
+	Status    bool   `json:"status"`
+	Codecs    []av.CodecData
+	Clients   map[uuid.UUID]viewer
+	hlsChanel chan av.Packet
+}
+
+type viewer struct {
+	c chan av.Packet
+}
+
+// NewApplication Prepare configuration for application
+func NewApplication(cfg *ConfigurationArgs) (*Application, error) {
+	tmp := Application{
+		Server: &ServerInfo{
+			HTTPAddr: cfg.Server.HTTPAddr,
+			HTTPPort: cfg.Server.HTTPPort,
+		},
+		Streams:         StreamsMap{Streams: make(map[uuid.UUID]*StreamConfiguration)},
+		HlsMsPerSegment: cfg.HlsMsPerSegment,
+		HlsDirectory:    cfg.HlsDirectory,
+		HlsWindowSize:   cfg.HlsWindowSize,
+		HlsCapacity:     cfg.HlsCapacity,
+	}
+	for i := range cfg.Streams {
+		validUUID, err := uuid.Parse(cfg.Streams[i].GUID)
+		if err != nil {
+			log.Printf("Not valid UUID: %s\n", cfg.Streams[i].GUID)
+			continue
+		}
+		tmp.Streams.Streams[validUUID] = &StreamConfiguration{
+			URL:       cfg.Streams[i].URL,
+			Clients:   make(map[uuid.UUID]viewer),
+			hlsChanel: make(chan av.Packet, 100),
+		}
+	}
+	return &tmp, nil
+}
+
+func (app *Application) cast(streamID uuid.UUID, pck av.Packet) error {
+	app.Streams.Lock()
+	defer app.Streams.Unlock()
+	curStream, ok := app.Streams.Streams[streamID]
+	if !ok {
+		return ErrStreamNotFound
+	}
+	curStream.hlsChanel <- pck
+	for _, v := range curStream.Clients {
+		if len(v.c) < cap(v.c) {
+			v.c <- pck
+		}
+	}
+	return nil
+}
+
+func (app *Application) ext(streamID uuid.UUID) bool {
+	app.Streams.Lock()
+	defer app.Streams.Unlock()
+	_, ok := app.Streams.Streams[streamID]
+	return ok
+}
+
+func (app *Application) codecAdd(streamID uuid.UUID, codecs []av.CodecData) {
+	app.Streams.Lock()
+	defer app.Streams.Unlock()
+	app.Streams.Streams[streamID].Codecs = codecs
+}
+
+func (app *Application) codecGet(streamID uuid.UUID) ([]av.CodecData, error) {
+	app.Streams.Lock()
+	defer app.Streams.Unlock()
+	curStream, ok := app.Streams.Streams[streamID]
+	if !ok {
+		return nil, ErrStreamNotFound
+	}
+	return curStream.Codecs, nil
+}
+
+func (app *Application) updateStatus(streamID uuid.UUID, status bool) error {
+	app.Streams.Lock()
+	defer app.Streams.Unlock()
+	t, ok := app.Streams.Streams[streamID]
+	if !ok {
+		return ErrStreamNotFound
+	}
+	t.Status = status
+	app.Streams.Streams[streamID] = t
+	return nil
+}
+
+func (app *Application) clientAdd(streamID uuid.UUID) (uuid.UUID, chan av.Packet, error) {
+	app.Streams.Lock()
+	defer app.Streams.Unlock()
+	clientID, err := uuid.NewUUID()
+	if err != nil {
+		return uuid.UUID{}, nil, err
+	}
+	ch := make(chan av.Packet, 100)
+	curStream, ok := app.Streams.Streams[streamID]
+	if !ok {
+		return uuid.UUID{}, nil, ErrStreamNotFound
+	}
+	curStream.Clients[clientID] = viewer{c: ch}
+	return clientID, ch, nil
+}
+
+func (app *Application) clientDelete(streamID, clientID uuid.UUID) {
+	defer app.Streams.Unlock()
+	app.Streams.Lock()
+	delete(app.Streams.Streams[streamID].Clients, clientID)
+}
+
+func (app *Application) startHlsCast(streamID uuid.UUID, stopCast chan bool) {
+	defer app.Streams.Unlock()
+	app.Streams.Lock()
+	go app.startHls(streamID, app.Streams.Streams[streamID].hlsChanel, stopCast)
+}
+
+func (app *Application) list() (uuid.UUID, []uuid.UUID) {
+	defer app.Streams.Unlock()
+	app.Streams.Lock()
+	res := []uuid.UUID{}
+	first := uuid.UUID{}
+	for k := range app.Streams.Streams {
+		if first.String() == "" {
+			first = k
+		}
+		res = append(res, k)
+	}
+	return first, res
+}

cmd/video_server/conf.json

@@ -1,15 +1,17 @@
 {
     "server": {
-        "http_port": ":8090"
+        "http_addr": "localhost",
+        "http_port": 8090
     },
-    "hlsMsPerSegment": 10000,
-    "hlsDirectory": "./hls",
-    "hlsWindowSize": 5,
-    "hlsWindowCapacity" : 10,
+    "hls_ms_per_segment": 10000,
+    "hls_directory": "./hls",
+    "hls_window_size": 5,
+    "hls_window_capacity" : 10,
     "streams": [
         {
             "guid": "c2680f08-07cc-4dee-9231-685893e2661f",
-            "url": "rtsp://170.93.143.139/rtplive/470011e600ef003a004ee33696235daa"
+            "url": "rtsp://170.93.143.139/rtplive/470011e600ef003a004ee33696235daa",
+            "stream_types": ["hls", "mse"]
         }
     ]
 }

cmd/video_server/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"flag"
+	"fmt"
 	"log"
 	"os"
 	"os/signal"
@@ -34,13 +35,19 @@ func main() {
 		defer pprof.StopCPUProfile()
 	}
 
-	cfg, err := videoserver.NewAppConfiguration(*conf)
+	settings, err := videoserver.NewConfiguration(*conf)
 	if err != nil {
-		log.Fatalln(err)
+		fmt.Printf("Can't prepare setting due the error: %s", err.Error())
+		return
+	}
+	app, err := videoserver.NewApplication(settings)
+	if err != nil {
+		fmt.Printf("Can't prepare application due the error: %s", err.Error())
+		return
 	}
 
-	go videoserver.StartHTTPServer(cfg)
-	go videoserver.StartStreams(cfg)
+	go app.StartHTTPServer()
+	go app.StartStreams()
 
 	sigOUT := make(chan os.Signal, 1)
 	exit := make(chan bool, 1)