修复集群传输bug

集群采集信息功能完善
2025-10-02 01:02:08 +08:00 · 2020-02-05 17:29:55 +08:00 · 2020-02-04 16:09:55 +08:00
21 changed files with 179 additions and 135 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/css/app.ce470878.css
+++ b/dashboard/dist/css/app.ce470878.css
@@ -1 +1 @@
-#app,body,html{height:100%}#app{font-family:Avenir,Helvetica,Arial,sans-serif;-webkit-font-smoothing:antialiased;-moz-osx-font-smoothing:grayscale;text-align:center;color:#184c18;position:relative}#app>div:first-child{position:absolute;top:10px;left:30px;font-size:x-large}.content{padding-top:60px}.feature-title[data-v-54efad41]{color:#eb5e46;font-weight:700;font-size:larger}p[data-v-54efad41]{margin:30px;font-size:20px}img[data-v-54efad41]{margin:20px}.root[data-v-e34eab40]{background:#d3d3d3}.root>img[data-v-e34eab40]{width:300px;margin:30px}.records[data-v-4eee1624]{display:-webkit-box;display:-ms-flexbox;display:flex;-ms-flex-wrap:wrap;flex-wrap:wrap;padding:0 15px}.records>[data-v-4eee1624]{width:200px}.log-container{overflow-y:auto;max-height:500px}@-webkit-keyframes recording-data-v-f6113870{0%{opacity:.2}50%{opacity:1}to{opacity:.2}}@keyframes recording-data-v-f6113870{0%{opacity:.2}50%{opacity:1}to{opacity:.2}}.recording[data-v-f6113870]{-webkit-animation:recording-data-v-f6113870 1s infinite;animation:recording-data-v-f6113870 1s infinite}.layout[data-v-f6113870]{padding-bottom:30px;display:-webkit-box;display:-ms-flexbox;display:flex;-ms-flex-wrap:wrap;flex-wrap:wrap}.room[data-v-f6113870]{width:250px;margin:10px;text-align:left}.empty[data-v-f6113870]{color:#eb5e46;width:100%;min-height:500px;-webkit-box-pack:center;-ms-flex-pack:center;justify-content:center;-webkit-box-align:center;-ms-flex-align:center;align-items:center}.empty[data-v-f6113870],.status[data-v-f6113870]{display:-webkit-box;display:-ms-flexbox;display:flex}.status[data-v-f6113870]{position:fixed;left:5px;bottom:10px}.status>div[data-v-f6113870]{margin:0 5px}
+#app,body,html{height:100%}#app{font-family:Avenir,Helvetica,Arial,sans-serif;-webkit-font-smoothing:antialiased;-moz-osx-font-smoothing:grayscale;text-align:center;color:#184c18;position:relative}#app>div:first-child{position:absolute;top:10px;left:30px;font-size:x-large}.content{padding-top:60px}.feature-title[data-v-54efad41]{color:#eb5e46;font-weight:700;font-size:larger}p[data-v-54efad41]{margin:30px;font-size:20px}img[data-v-54efad41]{margin:20px}.root[data-v-e34eab40]{background:#d3d3d3}.root>img[data-v-e34eab40]{width:300px;margin:30px}.records[data-v-7d5ab110]{display:-webkit-box;display:-ms-flexbox;display:flex;-ms-flex-wrap:wrap;flex-wrap:wrap;padding:0 15px}.records>[data-v-7d5ab110]{width:200px}.log-container{overflow-y:auto;max-height:500px}@-webkit-keyframes recording-data-v-65ac4b48{0%{opacity:.2}50%{opacity:1}to{opacity:.2}}@keyframes recording-data-v-65ac4b48{0%{opacity:.2}50%{opacity:1}to{opacity:.2}}.recording[data-v-65ac4b48]{-webkit-animation:recording-data-v-65ac4b48 1s infinite;animation:recording-data-v-65ac4b48 1s infinite}.layout[data-v-65ac4b48]{padding-bottom:30px;display:-webkit-box;display:-ms-flexbox;display:flex;-ms-flex-wrap:wrap;flex-wrap:wrap}.room[data-v-65ac4b48]{width:250px;margin:10px;text-align:left}.empty[data-v-65ac4b48]{color:#eb5e46;width:100%;min-height:500px;-webkit-box-pack:center;-ms-flex-pack:center;justify-content:center;-webkit-box-align:center;-ms-flex-align:center;align-items:center}.empty[data-v-65ac4b48],.status[data-v-65ac4b48]{display:-webkit-box;display:-ms-flexbox;display:flex}.status[data-v-65ac4b48]{position:fixed;left:5px;bottom:10px}.status>div[data-v-65ac4b48]{margin:0 5px}
--- 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.ce470878.css rel=preload as=style><link href=/css/chunk-vendors.22ebf426.css rel=preload as=style><link href=/js/app.017fb959.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.ce470878.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.017fb959.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.017fb959.js
+++ b/dashboard/dist/js/app.017fb959.js
--- a/dashboard/dist/js/app.017fb959.js.map
+++ b/dashboard/dist/js/app.017fb959.js.map
--- a/dashboard/dist/js/app.af5e5ef3.js
+++ b/dashboard/dist/js/app.af5e5ef3.js
--- 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
@@ -1,5 +1,9 @@
 <template>
-  <div id="mountNode"></div>
+  <div>
+    自动更新
+    <i-switch v-model="autoUpdate"></i-switch>
+    <div id="mountNode"></div>
+  </div>
 </template>

 <script>
@@ -7,24 +11,22 @@ import { mapState } from "vuex";
 import G6 from "@antv/g6";
 var graph = null;
 export default {
+  data() {
+    return {
+      autoUpdate: true
+    };
+  },
  computed: {
    ...mapState({
      data(state) {
-        let summary = state.summary;
-        // 点集
-        let nodes = [];
-        // 边集
-        let edges = [];
-        this.addServer(summary, nodes, edges);
-        return {
-          nodes,
-          edges
-        };
+        let d = this.addServer(state.summary);
+        d.label = "🏠" + d.label;
+        return d;
      }
    })
  },
  methods: {
-    addServer(node, nodes, edges) {
+    addServer(node) {
      let result = {
        id: node.Address,
        label: node.Address,
@@ -33,38 +35,35 @@ export default {
        shape: "modelRect",
        logoIcon: {
          show: false
-        }
+        },
+        children: []
      };
-      nodes.push(result);
+
      if (node.Rooms) {
        for (let i = 0; i < node.Rooms.length; i++) {
          let room = node.Rooms[i];
-          let roomId = result.id + room.StreamPath;
-          nodes.push({
+          let roomId = room.StreamPath;
+          let roomData = {
            id: roomId,
            label: room.StreamPath,
-            shape: "rect"
-          });
-          edges.push({ source: result.id, target: roomId });
+            shape: "rect",
+            children: []
+          };
+          result.children.push(roomData);
          if (room.SubscriberInfo) {
            for (let j = 0; j < room.SubscriberInfo.length; j++) {
              let subId = roomId + room.SubscriberInfo[j].ID;
-              nodes.push({
+              roomData.children.push({
                id: subId,
                label: room.SubscriberInfo[j].ID
              });
-              edges.push({ source: roomId, target: subId });
            }
          }
        }
      }
-      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);
-          edges.push({
-            source: result.id,
-            target: child.id
-          });
+      if (node.Children) {
+        for (let childId in node.Children) {
+          result.children.push(this.addServer(node.Children[childId]));
        }
      }
      return result;
@@ -72,23 +71,33 @@ export default {
  },
  watch: {
    data(v) {
-      if (graph) {
-        graph.read(v); // 加载数据
+      if (graph && this.autoUpdate) {
+        //graph.updateChild(v, "");
+        graph.changeData(v); // 加载数据
+        graph.fitView();
+        //graph.read(v);
      }
    }
  },
  mounted() {
-    graph = new G6.Graph({
-      renderer: "svg",
+    graph = new G6.TreeGraph({
+      linkCenter: true,
+      // renderer: "svg",
      container: "mountNode", // 指定挂载容器
      width: 800, // 图的宽度
      height: 500, // 图的高度
-      layout: {
-        type: "radial"
+      modes: {
+        default: ["drag-canvas", "zoom-canvas", "click-select", "drag-node"]
      },
-      defaultNode: {}
+      animate: false,
+      layout: {
+        // type: "indeted",
+        direction: "H"
+      }
    });
+    //graph.addChild(this.data, "");
    graph.read(this.data); // 加载数据
+    graph.fitView();
  }
 };
 </script>
--- a/dashboard/src/components/Jessibuca.vue
+++ b/dashboard/src/components/Jessibuca.vue
@@ -9,6 +9,8 @@
  >
    <canvas id="canvas" width="488" height="275" style="background: black" />
    <div slot="footer">
+      音频缓冲：
+      <InputNumber v-model="audioBuffer" size="small"></InputNumber>
      <Button v-if="audioEnabled" @click="turnOff" icon="md-volume-off" />
      <Button v-else @click="turnOn" icon="md-volume-up"></Button>
    </div>
@@ -22,18 +24,23 @@ export default {
  data() {
    return {
      audioEnabled: false,
+      audioBuffer: 12,
      url: ""
    };
  },
  watch: {
    audioEnabled(value) {
      h5lc.audioEnabled(value);
+    },
+    audioBuffer(v) {
+      h5lc.audioBuffer = v;
    }
  },
  mounted() {
    h5lc = new window.Jessibuca({
      canvas: document.getElementById("canvas"),
-      decoder: "jessibuca/ff.js"
+      decoder: "jessibuca/ff.js",
+      audioBuffer: this.audioBuffer
    });
  },
  destroyed() {
--- a/dashboard/src/components/Records.vue
+++ b/dashboard/src/components/Records.vue
@@ -33,7 +33,7 @@ export default {
        x => {
          if (x == "success") {
            this.onVisible(true);
-            this.$Message.success("删除成功");
+            this.$Message.success("开始发布");
          } else {
            this.$Message.error(x);
          }
@@ -50,7 +50,7 @@ export default {
            { streamPath: item.Path.replace(".flv", "") },
            x => {
              if (x == "success") {
-                this.$Message.success("开始发布");
+                this.$Message.success("删除成功");
              } else {
                this.$Message.error(x);
              }
--- a/dashboard/src/views/Console.vue
+++ b/dashboard/src/views/Console.vue
@@ -118,6 +118,7 @@ export default {
      currentTab: "",
      currentStream: [],
      typeMap: {
+        Receiver: "📡",
        FlvFile: "🎥",
        TS: "🎬",
        HLS: "🍎",
--- 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/monica/room.go
+++ b/monica/room.go
@@ -2,11 +2,12 @@ package monica

 import (
 	"context"
-	"github.com/langhuihui/monibuca/monica/avformat"
-	"github.com/langhuihui/monibuca/monica/pool"
 	"log"
 	"sync"
 	"time"
+
+	"github.com/langhuihui/monibuca/monica/avformat"
+	"github.com/langhuihui/monibuca/monica/pool"
 )

 var (
@@ -154,11 +155,17 @@ func (r *Room) Run() {
 	}
 }
 func (r *Room) PushAudio(audio *pool.AVPacket) {
+	if len(audio.Payload) < 3 {
+		return
+	}
 	if audio.Payload[0] == 0xFF && (audio.Payload[1]&0xF0) == 0xF0 {
 		audio.IsADTS = true
 		r.AudioTag = audio
 	} else if r.AudioTag == nil {
 		audio.IsAACSequence = true
+		if len(audio.Payload) < 5 {
+			return
+		}
 		r.AudioTag = audio
 		tmp := audio.Payload[0]                                                // 第一个字节保存着音频的相关信息
 		if r.AudioInfo.SoundFormat = tmp >> 4; r.AudioInfo.SoundFormat == 10 { //真的是AAC的话，后面有一个字节的详细信息
@@ -200,6 +207,9 @@ func (r *Room) setH264Info(video *pool.AVPacket) {
 	}
 }
 func (r *Room) PushVideo(video *pool.AVPacket) {
+	if len(video.Payload) < 3 {
+		return
+	}
 	video.VideoFrameType = video.Payload[0] >> 4  // 帧类型 4Bit, H264一般为1或者2
 	r.VideoInfo.CodecID = video.Payload[0] & 0x0f // 编码类型ID 4Bit, JPEG, H263, AVC...
 	video.IsAVCSequence = video.VideoFrameType == 1 && video.Payload[1] == 0
--- 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/publisher.go
+++ b/plugins/cluster/publisher.go
@@ -3,12 +3,14 @@ package cluster
 import (
 	"bufio"
 	"encoding/binary"
+	"io"
+	"log"
+	"net"
+	"strings"
+
 	. "github.com/langhuihui/monibuca/monica"
 	"github.com/langhuihui/monibuca/monica/avformat"
 	"github.com/langhuihui/monibuca/monica/pool"
-	"io"
-	"net"
-	"strings"
 )

 type Receiver struct {
@@ -26,6 +28,7 @@ func (p *Receiver) Auth(authSub *OutputStream) {

 func (p *Receiver) readAVPacket(avType byte) (av *pool.AVPacket, err error) {
 	buf := pool.GetSlice(4)
+	defer pool.RecycleSlice(buf)
 	_, err = io.ReadFull(p, buf)
 	if err != nil {
 		println(err.Error())
@@ -39,10 +42,7 @@ func (p *Receiver) readAVPacket(avType byte) (av *pool.AVPacket, err error) {
 	}
 	av.Payload = pool.GetSlice(int(binary.BigEndian.Uint32(buf)))
 	_, err = io.ReadFull(p, av.Payload)
-	if MayBeError(err) {
-		return
-	}
-	pool.RecycleSlice(buf)
+	MayBeError(err)
 	return
 }

@@ -57,7 +57,7 @@ func PullUpStream(streamPath string) {
 	}
 	brw := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
 	p := &Receiver{
-		Reader: conn,
+		Reader: brw.Reader,
 		Writer: brw.Writer,
 	}
 	if p.Publish(streamPath, p) {
@@ -72,11 +72,7 @@ func PullUpStream(streamPath string) {
 		return
 	}
 	defer p.Cancel()
-	for {
-		cmd, err := brw.ReadByte()
-		if MayBeError(err) {
-			return
-		}
+	for cmd, err := brw.ReadByte(); !MayBeError(err); cmd, err = brw.ReadByte() {
 		switch cmd {
 		case MSG_AUDIO:
 			if audio, err := p.readAVPacket(avformat.FLV_TAG_TYPE_AUDIO); err == nil {
@@ -103,6 +99,8 @@ func PullUpStream(streamPath string) {
 					v.Cancel()
 				}
 			}
+		default:
+			log.Printf("unknown cmd:%v", cmd)
 		}
 	}
 }
--- 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/plugins/jessica/subscriber.go
+++ b/plugins/jessica/subscriber.go
@@ -2,12 +2,13 @@ package jessica

 import (
 	"encoding/binary"
+	"net/http"
+	"strings"
+
 	"github.com/gobwas/ws"
 	. "github.com/langhuihui/monibuca/monica"
 	"github.com/langhuihui/monibuca/monica/avformat"
 	"github.com/langhuihui/monibuca/monica/pool"
-	"net/http"
-	"strings"
 )

 func WsHandler(w http.ResponseWriter, r *http.Request) {
--- a/slave.toml
+++ b/slave.toml
@@ -0,0 +1,18 @@
+# [Plugins.HDL]
+# ListenAddr = ":2020"
+[Plugins.Jessica]
+ListenAddr = ":8082"
+[Plugins.RTMP]
+ListenAddr = ":1936"
+[Plugins.GateWay]
+ListenAddr = ":8083"
+[Plugins.Cluster]
+Master = "localhost:2019"
+#ListenAddr = ":2019"
+#
+#[Plugins.Auth]
+#Key="www.monibuca.com"
+# [Plugins.RecordFlv]
+# Path="./resource"
+# [Plugins.QoS]
+# Suffix = ["high","medium","low"]
Author	SHA1	Message	Date
langhuihui	b3c8d35fad	修复集群传输bug	2020-02-05 17:29:55 +08:00
langhuihui	ab745145d9	集群采集信息功能完善	2020-02-04 16:09:55 +08:00