完善http接口封装

api.go

@@ -19,71 +19,34 @@ import (
 	"time"
 )
 
-var upgrader *websocket.Upgrader
+type ApiServer struct {
+	upgrader *websocket.Upgrader
+	router   *mux.Router
+}
+
+var apiServer *ApiServer
 
 func init() {
-	upgrader = &websocket.Upgrader{
-		CheckOrigin: func(r *http.Request) bool {
-			return true
+	apiServer = &ApiServer{
+		upgrader: &websocket.Upgrader{
+			CheckOrigin: func(r *http.Request) bool {
+				return true
+			},
 		},
+
+		router: mux.NewRouter(),
 	}
 }
 
 func startApiServer(addr string) {
-	r := mux.NewRouter()
-	/**
-	  http://host:port/xxx.flv
-	  http://host:port/xxx.rtc
-	  http://host:port/xxx.m3u8
-	  http://host:port/xxx_0.ts
-	  ws://host:port/xxx.flv
-	*/
-	r.HandleFunc("/live/{source}", func(w http.ResponseWriter, r *http.Request) {
-		vars := mux.Vars(r)
-		source := vars["source"]
-		index := strings.LastIndex(source, ".")
-		if index < 0 || index == len(source)-1 {
-			log.Sugar.Errorf("bad request:%s. stream format must be passed at the end of the URL", r.URL.Path)
-			w.WriteHeader(http.StatusBadRequest)
-			return
-		}
-
-		sourceId := source[:index]
-		format := source[index+1:]
-
-		if "flv" == format {
-			//判断是否是websocket请求
-			ws := true
-			if !("upgrade" == strings.ToLower(r.Header.Get("Connection"))) {
-				ws = false
-			} else if !("websocket" == strings.ToLower(r.Header.Get("Upgrade"))) {
-				ws = false
-			} else if !("13" == r.Header.Get("Sec-Websocket-Version")) {
-				ws = false
-			}
-
-			if ws {
-				onWSFlv(sourceId, w, r)
-			} else {
-				onFLV(sourceId, w, r)
-			}
-
-		} else if "m3u8" == format {
-			onHLS(sourceId, w, r)
-		} else if "ts" == format {
-			onTS(sourceId, w, r)
-		} else if "rtc" == format {
-			onRtc(sourceId, w, r)
-		}
-	})
-
-	r.HandleFunc("/rtc.html", func(writer http.ResponseWriter, request *http.Request) {
+	apiServer.router.HandleFunc("/live/{source}", apiServer.filterLive)
+	apiServer.router.HandleFunc("/rtc.html", func(writer http.ResponseWriter, request *http.Request) {
 		http.ServeFile(writer, request, "./rtc.html")
 	})
-	http.Handle("/", r)
+	http.Handle("/", apiServer.router)
 
 	srv := &http.Server{
-		Handler: r,
+		Handler: apiServer.router,
 		Addr:    addr,
 		// Good practice: enforce timeouts for servers you create!
 		WriteTimeout: 30 * time.Second,
@@ -97,27 +60,89 @@ func startApiServer(addr string) {
 	}
 }
 
-func onWSFlv(sourceId string, w http.ResponseWriter, r *http.Request) {
-	conn, err := upgrader.Upgrade(w, r, nil)
+func (api *ApiServer) generateSinkId(remoteAddr string) stream.SinkId {
+	tcpAddr, err := net.ResolveTCPAddr("tcp", remoteAddr)
+	if err != nil {
+		panic(err)
+	}
+
+	return stream.GenerateSinkId(tcpAddr)
+}
+
+func (api *ApiServer) doPlay(sink stream.ISink) utils.HookState {
+	ok := utils.HookStateOK
+	sink.Play(func() {
+
+	}, func(state utils.HookState) {
+		ok = state
+	})
+
+	return ok
+}
+
+func (api *ApiServer) filterLive(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	source := vars["source"]
+	index := strings.LastIndex(source, ".")
+	if index < 0 || index == len(source)-1 {
+		log.Sugar.Errorf("bad request:%s. stream format must be passed at the end of the URL", r.URL.Path)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	sourceId := source[:index]
+	format := source[index+1:]
+
+	/**
+	  http://host:port/xxx.flv
+	  http://host:port/xxx.rtc
+	  http://host:port/xxx.m3u8
+	  http://host:port/xxx_0.ts
+	  ws://host:port/xxx.flv
+	*/
+	if "flv" == format {
+		//判断是否是websocket请求
+		ws := true
+		if !("upgrade" == strings.ToLower(r.Header.Get("Connection"))) {
+			ws = false
+		} else if !("websocket" == strings.ToLower(r.Header.Get("Upgrade"))) {
+			ws = false
+		} else if !("13" == r.Header.Get("Sec-Websocket-Version")) {
+			ws = false
+		}
+
+		if ws {
+			api.onWSFlv(sourceId, w, r)
+		} else {
+			api.onFLV(sourceId, w, r)
+		}
+
+	} else if "m3u8" == format {
+		api.onHLS(sourceId, w, r)
+	} else if "ts" == format {
+		api.onTS(sourceId, w, r)
+	} else if "rtc" == format {
+		api.onRtc(sourceId, w, r)
+	}
+}
+
+func (api *ApiServer) onWSFlv(sourceId string, w http.ResponseWriter, r *http.Request) {
+	conn, err := api.upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		log.Sugar.Errorf("websocket头检查失败 err:%s", err.Error())
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
-	tcpAddr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
-	sinkId := stream.GenerateSinkId(tcpAddr)
-	sink := flv.NewFLVSink(sinkId, sourceId, flv.NewWSConn(conn))
-
+	sink := flv.NewFLVSink(api.generateSinkId(r.RemoteAddr), sourceId, flv.NewWSConn(conn))
 	log.Sugar.Infof("ws-flv 连接 sink:%s", sink.PrintInfo())
 
-	sink.(*stream.SinkImpl).Play(sink, func() {
-
-	}, func(state utils.HookState) {
+	state := api.doPlay(sink)
+	if utils.HookStateOK != state {
+		log.Sugar.Warnf("ws-flv 播放失败 state:%d sink:%s", state, sink.PrintInfo())
 		w.WriteHeader(http.StatusForbidden)
-
-		conn.Close()
-	})
+		return
+	}
 
 	netConn := conn.NetConn()
 	bytes := make([]byte, 64)
@@ -130,7 +155,7 @@ func onWSFlv(sourceId string, w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func onFLV(sourceId string, w http.ResponseWriter, r *http.Request) {
+func (api *ApiServer) onFLV(sourceId string, w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "video/x-flv")
 	w.Header().Set("Connection", "Keep-Alive")
 	w.Header().Set("Transfer-Encoding", "chunked")
@@ -148,18 +173,16 @@ func onFLV(sourceId string, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	tcpAddr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
-	sinkId := stream.GenerateSinkId(tcpAddr)
-	sink := flv.NewFLVSink(sinkId, sourceId, conn)
-
+	sink := flv.NewFLVSink(api.generateSinkId(r.RemoteAddr), sourceId, conn)
 	log.Sugar.Infof("http-flv 连接 sink:%s", sink.PrintInfo())
-	sink.(*stream.SinkImpl).Play(sink, func() {
 
-	}, func(state utils.HookState) {
+	state := api.doPlay(sink)
+	if utils.HookStateOK != state {
+		log.Sugar.Warnf("http-flv 播放失败 state:%d sink:%s", state, sink.PrintInfo())
+
 		w.WriteHeader(http.StatusForbidden)
-
-		conn.Close()
-	})
+		return
+	}
 
 	bytes := make([]byte, 64)
 	for {
@@ -170,7 +193,7 @@ func onFLV(sourceId string, w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func onTS(source string, w http.ResponseWriter, r *http.Request) {
+func (api *ApiServer) onTS(source string, w http.ResponseWriter, r *http.Request) {
 	if !stream.AppConfig.Hls.Enable {
 		log.Sugar.Warnf("处理m3u8请求失败 server未开启hls request:%s", r.URL.Path)
 		http.Error(w, "hls disable", http.StatusInternalServerError)
@@ -196,17 +219,16 @@ func onTS(source string, w http.ResponseWriter, r *http.Request) {
 	http.ServeFile(w, r, tsPath)
 }
 
-func onHLS(sourceId string, w http.ResponseWriter, r *http.Request) {
+func (api *ApiServer) onHLS(sourceId string, w http.ResponseWriter, r *http.Request) {
 	if !stream.AppConfig.Hls.Enable {
-		log.Sugar.Warnf("处理hls请求失败 server未开启hls request:%s", r.URL.Path)
+		log.Sugar.Warnf("处理hls请求失败 server未开启hls")
 		http.Error(w, "hls disable", http.StatusInternalServerError)
 		return
 	}
 
 	w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
 	//m3u8和ts会一直刷新, 每个请求只hook一次.
-	tcpAddr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
-	sinkId := stream.GenerateSinkId(tcpAddr)
+	sinkId := api.generateSinkId(r.RemoteAddr)
 
 	//hook成功后, 如果还没有m3u8文件，等生成m3u8文件
 	//后续直接返回当前m3u8文件
@@ -221,31 +243,26 @@ func onHLS(sourceId string, w http.ResponseWriter, r *http.Request) {
 			done <- 0
 		})
 
-		hookState := utils.HookStateOK
-		sink.Play(sink, func() {
-			err := stream.SinkManager.Add(sink)
+		state := api.doPlay(sink)
+		if utils.HookStateOK != state {
+			log.Sugar.Warnf("m3u8 请求失败 state:%d sink:%s", state, sink.PrintInfo())
 
-			utils.Assert(err == nil)
-		}, func(state utils.HookState) {
-			log.Sugar.Warnf("hook播放事件失败 request:%s", r.URL.Path)
-			hookState = state
 			w.WriteHeader(http.StatusForbidden)
-		})
-
-		if utils.HookStateOK != hookState {
 			return
+		} else {
+			err := stream.SinkManager.Add(sink)
+			utils.Assert(err == nil)
 		}
 
 		select {
 		case <-done:
 		case <-context.Done():
-			log.Sugar.Infof("http m3u8连接断开")
 			break
 		}
 	}
 }
 
-func onRtc(sourceId string, w http.ResponseWriter, r *http.Request) {
+func (api *ApiServer) onRtc(sourceId string, w http.ResponseWriter, r *http.Request) {
 	v := struct {
 		Type string `json:"type"`
 		SDP  string `json:"sdp"`
@@ -253,19 +270,20 @@ func onRtc(sourceId string, w http.ResponseWriter, r *http.Request) {
 
 	data, err := io.ReadAll(r.Body)
 	if err != nil {
-		panic(err)
-	}
+		log.Sugar.Errorf("rtc请求错误 err:%s remote:%s", err.Error(), r.RemoteAddr)
 
-	if err := json.Unmarshal(data, &v); err != nil {
-		panic(err)
-	}
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	} else if err := json.Unmarshal(data, &v); err != nil {
+		log.Sugar.Errorf("rtc请求错误 err:%s remote:%s", err.Error(), r.RemoteAddr)
 
-	tcpAddr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
-	sinkId := stream.GenerateSinkId(tcpAddr)
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
 
 	group := sync.WaitGroup{}
 	group.Add(1)
-	sink := rtc.NewSink(sinkId, sourceId, v.SDP, func(sdp string) {
+	sink := rtc.NewSink(api.generateSinkId(r.RemoteAddr), sourceId, v.SDP, func(sdp string) {
 		response := struct {
 			Type string `json:"type"`
 			SDP  string `json:"sdp"`
@@ -285,13 +303,15 @@ func onRtc(sourceId string, w http.ResponseWriter, r *http.Request) {
 		group.Done()
 	})
 
-	sink.Play(sink, func() {
+	log.Sugar.Infof("rtc 请求 sink:%s sdp:%v", sink.PrintInfo(), v.SDP)
+
+	state := api.doPlay(sink)
+	if utils.HookStateOK != state {
+		log.Sugar.Warnf("rtc 播放失败 state:%d sink:%s", state, sink.PrintInfo())
 
-	}, func(state utils.HookState) {
 		w.WriteHeader(http.StatusForbidden)
-
 		group.Done()
-	})
+	}
 
 	group.Wait()
 }

rtmp/rtmp_session.go

@@ -61,7 +61,7 @@ func (s *sessionImpl) OnPlay(app, stream_ string, response chan utils.HookState)
 
 	log.Sugar.Infof("rtmp onplay app:%s stream:%s sink:%v conn:%s", app, stream_, sink.Id(), s.conn.RemoteAddr().String())
 
-	sink.(*stream.SinkImpl).Play(sink, func() {
+	sink.Play(func() {
 		s.handle = sink
 		response <- utils.HookStateOK
 	}, func(state utils.HookState) {

rtsp/rtsp_session.go

@@ -163,7 +163,7 @@ func (s *session) onDescribe(source string, headers textproto.MIMEHeader) error
 
 	code := utils.HookStateOK
 	s.sink_ = sink_.(*sink)
-	sink_.(*sink).Play(sink_, func() {
+	sink_.Play(func() {
 
 	}, func(state utils.HookState) {
 		code = state

stream/hook.go

@@ -38,9 +38,9 @@ func NewPublishHookEventInfo(stream, remoteAddr string, protocol SourceType) eve
 }
 
 type HookHandler interface {
-	Play(sink ISink, success func(), failure func(state utils.HookState))
+	Play(success func(), failure func(state utils.HookState))
 
-	PlayDone(sink ISink, success func(), failure func(state utils.HookState))
+	PlayDone(success func(), failure func(state utils.HookState))
 }
 
 type HookSession interface {

stream/sink.go

@@ -192,18 +192,18 @@ func (s *SinkImpl) PrintInfo() string {
 	return fmt.Sprintf("%s-%v source:%s", s.ProtocolStr(), s.Id_, s.SourceId_)
 }
 
-func (s *SinkImpl) Play(sink ISink, success func(), failure func(state utils.HookState)) {
+func (s *SinkImpl) Play(success func(), failure func(state utils.HookState)) {
 	f := func() {
-		source := SourceManager.Find(sink.SourceId())
+		source := SourceManager.Find(s.SourceId())
 		if source == nil {
-			log.Sugar.Infof("添加sink到等待队列 sink:%s-%v source:%s", sink.ProtocolStr(), sink.Id(), sink.SourceId())
+			log.Sugar.Infof("添加sink到等待队列 sink:%s-%v source:%s", s.ProtocolStr(), s.Id(), s.SourceId())
 
-			sink.SetState(SessionStateWait)
-			AddSinkToWaitingQueue(sink.SourceId(), sink)
+			s.SetState(SessionStateWait)
+			AddSinkToWaitingQueue(s.SourceId(), s)
 		} else {
-			log.Sugar.Debugf("发送播放事件 sink:%s-%v source:%s", sink.ProtocolStr(), sink.Id(), sink.SourceId())
+			log.Sugar.Debugf("发送播放事件 sink:%s-%v source:%s", s.ProtocolStr(), s.Id(), s.SourceId())
 
-			source.AddEvent(SourceEventPlay, sink)
+			source.AddEvent(SourceEventPlay, s)
 		}
 	}
 
@@ -213,23 +213,23 @@ func (s *SinkImpl) Play(sink ISink, success func(), failure func(state utils.Hoo
 		return
 	}
 
-	err := s.Hook(HookEventPlay, NewPlayHookEventInfo(sink.SourceId(), "", sink.Protocol()), func(response *http.Response) {
+	err := s.Hook(HookEventPlay, NewPlayHookEventInfo(s.SourceId(), "", s.Protocol()), func(response *http.Response) {
 		f()
 		success()
 	}, func(response *http.Response, err error) {
-		log.Sugar.Errorf("Hook播放事件响应失败 err:%s sink:%s-%v source:%s", err.Error(), sink.ProtocolStr(), sink.Id(), sink.SourceId())
+		log.Sugar.Errorf("Hook播放事件响应失败 err:%s sink:%s-%v source:%s", err.Error(), s.ProtocolStr(), s.Id(), s.SourceId())
 
 		failure(utils.HookStateFailure)
 	})
 
 	if err != nil {
-		log.Sugar.Errorf("Hook播放事件发送失败 err:%s sink:%s-%v source:%s", err.Error(), sink.ProtocolStr(), sink.Id(), sink.SourceId())
+		log.Sugar.Errorf("Hook播放事件发送失败 err:%s sink:%s-%v source:%s", err.Error(), s.ProtocolStr(), s.Id(), s.SourceId())
 
 		failure(utils.HookStateFailure)
 		return
 	}
 }
 
-func (s *SinkImpl) PlayDone(source ISink, success func(), failure func(state utils.HookState)) {
+func (s *SinkImpl) PlayDone(success func(), failure func(state utils.HookState)) {
 
 }