开发订阅者插件

所谓订阅者就是用来从流媒体服务器接收音视频流的程序，例如RTMP协议执行play命令后、http-flv请求响应程序、websocket响应程序。内置插件中录制flv程序也是一个特殊的订阅者。下面是http-flv插件的源码，供参考

  1. package HDL
  3. import (
  4.     . "github.com/langhuihui/monibuca/monica"
  5.     "github.com/langhuihui/monibuca/monica/avformat"
  6.     "github.com/langhuihui/monibuca/monica/pool"
  7.     "log"
  8.     "net/http"
  9.     "strings"
  10. )
  12. var config = new(ListenerConfig)
  14. func init() {
  15.     InstallPlugin(&PluginConfig{
  16.         Name:   "HDL",
  17.         Type:   PLUGIN_SUBSCRIBER,
  18.         Config: config,
  19.         Run:    run,
  20.     })
  21. }
  23. func run() {
  24.     log.Printf("HDL start at %s", config.ListenAddr)
  25.     log.Fatal(http.ListenAndServe(config.ListenAddr, http.HandlerFunc(HDLHandler)))
  26. }
  28. func HDLHandler(w http.ResponseWriter, r *http.Request) {
  29.     sign := r.URL.Query().Get("sign")
  30.     if err := AuthHooks.Trigger(sign); err != nil {
  31.         w.WriteHeader(403)
  32.         return
  33.     }
  34.     stringPath := strings.TrimLeft(r.RequestURI, "/")
  35.     if strings.HasSuffix(stringPath, ".flv") {
  36.         stringPath = strings.TrimRight(stringPath, ".flv")
  37.     }
  38.     if _, ok := AllRoom.Load(stringPath); ok {
  39.         //atomic.AddInt32(&hdlId, 1)
  40.         w.Header().Set("Transfer-Encoding", "chunked")
  41.         w.Header().Set("Content-Type", "video/x-flv")
  42.         w.Write(avformat.FLVHeader)
  43.         p := OutputStream{
  44.             Sign: sign,
  45.             SendHandler: func(packet *pool.SendPacket) error {
  46.                 return avformat.WriteFLVTag(w, packet)
  47.             },
  48.             SubscriberInfo: SubscriberInfo{
  49.                 ID: r.RemoteAddr, Type: "FLV",
  50.             },
  51.         }
  52.         p.Play(stringPath)
  53.     } else {
  54.         w.WriteHeader(404)
  55.     }
  56. }

其中，核心逻辑就是创建OutputStream对象，每一个订阅者需要提供SendHandler函数，用来接收来自发布者广播出来的音视频数据。最后调用该对象的Play函数进行播放。请注意：Play函数会阻塞当前goroutine。