集群采集信息功能完善

2025-09-27 05:35:57 +08:00 · 2020-02-04 13:46:19 +08:00
13 changed files with 101 additions and 84 deletions
--- a/config.toml
+++ b/config.toml
@@ -7,8 +7,8 @@ ListenAddr = ":1935"
 [Plugins.GateWay]
 ListenAddr = ":8081"
 [Plugins.Cluster]
-Master = "203.60.1.23:2019"
-#ListenAddr = ":2019"
+#Master = "localhost:2019"
+ListenAddr = ":2019"
 #
 #[Plugins.Auth]
 #Key="www.monibuca.com"
--- a/dashboard/dist/index.html
+++ b/dashboard/dist/index.html
@@ -1 +1 @@
-<!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta http-equiv=X-UA-Compatible content="IE=edge"><meta name=viewport content="width=device-width,initial-scale=1"><link rel=icon href=/favicon.ico><title>Monibuca</title><script src=jessibuca/ajax.js></script><script src=jessibuca/renderer.js></script><link href=/css/app.ea4656d8.css rel=preload as=style><link href=/css/chunk-vendors.22ebf426.css rel=preload as=style><link href=/js/app.af5e5ef3.js rel=preload as=script><link href=/js/chunk-vendors.ebc28a73.js rel=preload as=script><link href=/css/chunk-vendors.22ebf426.css rel=stylesheet><link href=/css/app.ea4656d8.css rel=stylesheet></head><body><noscript><strong>We're sorry but dashboard doesn't work properly without JavaScript enabled. Please enable it to continue.</strong></noscript><div id=app></div><script src=/js/chunk-vendors.ebc28a73.js></script><script src=/js/app.af5e5ef3.js></script></body></html>
+<!DOCTYPE html><html lang=en><head><meta charset=utf-8><meta http-equiv=X-UA-Compatible content="IE=edge"><meta name=viewport content="width=device-width,initial-scale=1"><link rel=icon href=/favicon.ico><title>Monibuca</title><script src=jessibuca/ajax.js></script><script src=jessibuca/renderer.js></script><link href=/css/app.ea4656d8.css rel=preload as=style><link href=/css/chunk-vendors.22ebf426.css rel=preload as=style><link href=/js/app.0267da57.js rel=preload as=script><link href=/js/chunk-vendors.ebc28a73.js rel=preload as=script><link href=/css/chunk-vendors.22ebf426.css rel=stylesheet><link href=/css/app.ea4656d8.css rel=stylesheet></head><body><noscript><strong>We're sorry but dashboard doesn't work properly without JavaScript enabled. Please enable it to continue.</strong></noscript><div id=app></div><script src=/js/chunk-vendors.ebc28a73.js></script><script src=/js/app.0267da57.js></script></body></html>
--- a/dashboard/dist/jessibuca/renderer.js
+++ b/dashboard/dist/jessibuca/renderer.js
@@ -9,7 +9,7 @@ function Jessibuca(opt) {
        this.initBuffers();
        this.initTextures();
    };
-    this.decoderWorker = new Worker(opt.decoder || '264_mp3.js')
+    this.decoderWorker = new Worker(opt.decoder || 'ff.js')
    var _this = this
    function draw(output) {
        _this.drawNextOutputPicture(_this.width, _this.height, null, output)
@@ -118,12 +118,15 @@ Jessibuca.prototype.playAudio = function (data) {
        }
        // setTimeout(playNextBuffer, buffer.duration * 1000)
    }
-    var tryPlay = function (buffer) {
+    var decodeAudio = function () {
        if (decodeQueue.length) {
-            context.decodeAudioData(decodeQueue.shift(), tryPlay, console.error);
+            context.decodeAudioData(decodeQueue.shift(), tryPlay, decodeAudio);
        } else {
            isDecoding = false
        }
+    }
+    var tryPlay = function (buffer) {
+        decodeAudio()
        if (isPlaying) {
            audioBuffers.push(buffer);
        } else {
@@ -134,7 +137,7 @@ Jessibuca.prototype.playAudio = function (data) {
        decodeQueue.push(...data)
        if (!isDecoding) {
            isDecoding = true
-            context.decodeAudioData(decodeQueue.shift(), tryPlay, console.error);
+            decodeAudio()
        }
    }
    this.playAudio = playAudio
@@ -452,7 +455,7 @@ Jessibuca.prototype.close = function () {
    this.decoderWorker.postMessage({ cmd: "close" })
    this.contextGL.clear(this.contextGL.COLOR_BUFFER_BIT);
 }
-Jessibuca.prototype.destroy = function(){
+Jessibuca.prototype.destroy = function () {
    this.decoderWorker.terminate()
 }
 Jessibuca.prototype.play = function (url) {
--- a/dashboard/dist/js/app.0267da57.js
+++ b/dashboard/dist/js/app.0267da57.js
--- a/dashboard/dist/js/app.0267da57.js.map
+++ b/dashboard/dist/js/app.0267da57.js.map
--- a/dashboard/dist/js/app.af5e5ef3.js.map
+++ b/dashboard/dist/js/app.af5e5ef3.js.map
--- a/dashboard/public/jessibuca/renderer.js
+++ b/dashboard/public/jessibuca/renderer.js
@@ -9,7 +9,7 @@ function Jessibuca(opt) {
        this.initBuffers();
        this.initTextures();
    };
-    this.decoderWorker = new Worker(opt.decoder || '264_mp3.js')
+    this.decoderWorker = new Worker(opt.decoder || 'ff.js')
    var _this = this
    function draw(output) {
        _this.drawNextOutputPicture(_this.width, _this.height, null, output)
--- a/dashboard/src/components/Cluster.vue
+++ b/dashboard/src/components/Cluster.vue
@@ -58,16 +58,15 @@ export default {
          }
        }
      }
-      if (node.Children && node.Children.length > 0) {
-        for (let i = 0; i < node.Children.length; i++) {
-          let child = this.addServer(node.Children[i], nodes, edges);
+      if (node.Children) {
+        for (let childId in node.Children) {
+          this.addServer(node.Children[childId], nodes, edges);
          edges.push({
            source: result.id,
-            target: child.id
+            target: childId
          });
        }
      }
-      return result;
    }
  },
  watch: {
--- a/monibuca.exe~
+++ b/monibuca.exe~
--- a/monica/hook.go
+++ b/monica/hook.go
@@ -8,8 +8,8 @@ func (h AuthHook) AddHook(hook func(string) error) {
 	AuthHooks = append(h, hook)
 }
 func (h AuthHook) Trigger(sign string) error {
-	for _, h := range h {
-		if err := h(sign); err != nil {
+	for _, f := range h {
+		if err := f(sign); err != nil {
 			return err
 		}
 	}
@@ -24,8 +24,8 @@ func (h OnPublishHook) AddHook(hook func(r *Room)) {
 	OnPublishHooks = append(h, hook)
 }
 func (h OnPublishHook) Trigger(r *Room) {
-	for _, h := range h {
-		h(r)
+	for _, f := range h {
+		f(r)
 	}
 }

@@ -37,8 +37,8 @@ func (h OnSubscribeHook) AddHook(hook func(s *OutputStream)) {
 	OnSubscribeHooks = append(h, hook)
 }
 func (h OnSubscribeHook) Trigger(s *OutputStream) {
-	for _, h := range h {
-		h(s)
+	for _, f := range h {
+		f(s)
 	}
 }

@@ -50,8 +50,8 @@ func (h OnDropHook) AddHook(hook func(s *OutputStream)) {
 	OnDropHooks = append(h, hook)
 }
 func (h OnDropHook) Trigger(s *OutputStream) {
-	for _, h := range h {
-		h(s)
+	for _, f := range h {
+		f(s)
 	}
 }

@@ -63,7 +63,7 @@ func (h OnSummaryHook) AddHook(hook func(bool)) {
 	OnSummaryHooks = append(h, hook)
 }
 func (h OnSummaryHook) Trigger(v bool) {
-	for _, h := range h {
-		h(v)
+	for _, f := range h {
+		f(v)
 	}
 }
--- a/plugins/cluster/index.go
+++ b/plugins/cluster/index.go
@@ -3,6 +3,7 @@ package cluster
 import (
 	"bufio"
 	"encoding/json"
+	"io"
 	"log"
 	"math/rand"
 	"net"
@@ -46,56 +47,52 @@ func run() {
 		if MayBeError(err) {
 			return
 		}
-		masterConn, err = net.DialTCP("tcp", nil, addr)
-		if MayBeError(err) {
-			return
-		}
-		go readMaster()
+		go readMaster(addr)
 	}
 	if config.ListenAddr != "" {
+		Summary.Children = make(map[string]*ServerSummary)
 		OnSummaryHooks.AddHook(onSummary)
 		log.Printf("server bare start at %s", config.ListenAddr)
 		log.Fatal(ListenBare(config.ListenAddr))
 	}
 }
-func readMaster() {
+
+func readMaster(addr *net.TCPAddr) {
 	var err error
-	defer func() {
-		for {
-			t := 5 + rand.Int63n(5)
-			log.Printf("reconnect to master %s after %d seconds", config.Master, t)
-			time.Sleep(time.Duration(t) * time.Second)
-			addr, _ := net.ResolveTCPAddr("tcp", config.Master)
-			if masterConn, err = net.DialTCP("tcp", nil, addr); err == nil {
-				go readMaster()
-				return
-			}
-		}
-	}()
-	brw := bufio.NewReadWriter(bufio.NewReader(masterConn), bufio.NewWriter(masterConn))
-	log.Printf("connect to master %s reporting", config.Master)
-	//首次报告
-	if b, err := json.Marshal(Summary); err == nil {
-		_, err = masterConn.Write(b)
-	}
+	var cmd byte
 	for {
-		cmd, err := brw.ReadByte()
-		if err != nil {
-			return
-		}
-		switch cmd {
-		case MSG_SUMMARY: //收到主服务器指令，进行采集和上报
-			log.Println("receive summary request from master")
-			if cmd, err = brw.ReadByte(); err != nil {
-				return
-			}
-			if cmd == 1 {
-				Summary.Add()
-				go onReport()
-			} else {
-				Summary.Done()
+		if masterConn, err = net.DialTCP("tcp", nil, addr); !MayBeError(err) {
+			reader := bufio.NewReader(masterConn)
+			log.Printf("connect to master %s reporting", config.Master)
+			for report(); err == nil; {
+				if cmd, err = reader.ReadByte(); !MayBeError(err) {
+					switch cmd {
+					case MSG_SUMMARY: //收到主服务器指令，进行采集和上报
+						log.Println("receive summary request from master")
+						if cmd, err = reader.ReadByte(); !MayBeError(err) {
+							if cmd == 1 {
+								Summary.Add()
+								go onReport()
+							} else {
+								Summary.Done()
+							}
+						}
+					}
+				}
 			}
 		}
+		t := 5 + rand.Int63n(5)
+		log.Printf("reconnect to master %s after %d seconds", config.Master, t)
+		time.Sleep(time.Duration(t) * time.Second)
+	}
+}
+func report() {
+	if b, err := json.Marshal(Summary); err == nil {
+		data := make([]byte, len(b)+2)
+		data[0] = MSG_SUMMARY
+		copy(data[1:], b)
+		data[len(data)-1] = 0
+		_, err = masterConn.Write(data)
 	}
 }

@@ -103,28 +100,24 @@ func readMaster() {
 func onReport() {
 	for range time.NewTicker(time.Second).C {
 		if Summary.Running() {
-			if b, err := json.Marshal(Summary); err == nil {
-				data := make([]byte, len(b)+2)
-				data[0] = MSG_SUMMARY
-				copy(data[1:], b)
-				data[len(data)-1] = 0
-				_, err = masterConn.Write(data)
-			}
+			report()
 		} else {
 			return
 		}
 	}
 }
+func orderReport(conn io.Writer, start bool) {
+	b := []byte{MSG_SUMMARY, 0}
+	if start {
+		b[1] = 1
+	}
+	conn.Write(b)
+}

 //通知从服务器需要上报或者关闭上报
 func onSummary(start bool) {
 	slaves.Range(func(k, v interface{}) bool {
-		conn := v.(*net.TCPConn)
-		b := []byte{MSG_SUMMARY, 0}
-		if start {
-			b[1] = 1
-		}
-		conn.Write(b)
+		orderReport(v.(*net.TCPConn), start)
 		return true
 	})
 }
--- a/plugins/cluster/subscriber.go
+++ b/plugins/cluster/subscriber.go
@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net"
 	"strconv"
 	"strings"
@@ -80,29 +81,32 @@ func process(conn net.Conn) {
 		if err != nil {
 			return
 		}
+		bytes = bytes[0 : len(bytes)-1]
 		switch cmd {
 		case MSG_SUBSCRIBE:
 			if stream.Room != nil {
 				fmt.Printf("bare stream already exist from %s", conn.RemoteAddr())
 				return
 			}
-			streamName := string(bytes[0 : len(bytes)-1])
-			go stream.Play(streamName)
+			go stream.Play(string(bytes))
 		case MSG_AUTH:
-			sign := strings.Split(string(bytes[0:len(bytes)-1]), ",")
+			sign := strings.Split(string(bytes), ",")
 			head := []byte{MSG_AUTH, 2}
 			if len(sign) > 1 && AuthHooks.Trigger(sign[1]) == nil {
 				head[1] = 1
 			}
 			conn.Write(head)
-			conn.Write(bytes)
+			conn.Write(bytes[0 : len(bytes)+1])
 		case MSG_SUMMARY: //收到从服务器发来报告，加入摘要中
-			var summary *ServerSummary
+			summary := &ServerSummary{}
 			if err = json.Unmarshal(bytes, summary); err == nil {
 				summary.Address = connAddr
 				Summary.Report(summary)
 				if _, ok := slaves.Load(connAddr); !ok {
 					slaves.Store(connAddr, conn)
+					if Summary.Running() {
+						orderReport(io.Writer(conn), true)
+					}
 					defer slaves.Delete(connAddr)
 				}
 			}
--- a/slave.toml
+++ b/slave.toml
@@ -0,0 +1,18 @@
+# [Plugins.HDL]
+# ListenAddr = ":2020"
+# [Plugins.Jessica]
+# ListenAddr = ":8080"
+# [Plugins.RTMP]
+# ListenAddr = ":1935"
+# [Plugins.GateWay]
+# ListenAddr = ":8081"
+[Plugins.Cluster]
+Master = "localhost:2019"
+#ListenAddr = ":2019"
+#
+#[Plugins.Auth]
+#Key="www.monibuca.com"
+# [Plugins.RecordFlv]
+# Path="./resource"
+# [Plugins.QoS]
+# Suffix = ["high","medium","low"]