golang rtsp to http

import ( "github.com/gorilla/mux" "github.com/nareix/srtcp" "github.com/nareix/udp" "github.com/nareix/webrtc" "github.com/nareix/xtcp" "github.com/nareix/joy4/format" "github.com/nareix/joy4/format/ts" "github.com/nareix/joy4/av" "github.com/nareix/joy4/container/rtp" "github.com/nareix/joy4/av/pubsub" "github.com/nareix/joy4/cgo/ffmpeg" "github.com/nareix/joy4/av/pktque" "net/http" "io" "fmt" "bytes" "strconv" ) ... // 使用gosip库解析RTSP请求 func processRTSP(rtspRequest io.ReadWriteCloser, pubsub1 *pubsub.PubSub, aacWrite io.WriteCloser, h264Write io.WriteCloser) { sessionHandle := func(s *rtsp.Session) { p, err := s.Streams() checkError(err) var vtrack av.CodecData var atracks []av.CodecData for _, vi := range p { switch vi.Type().(type) { case av.H264CodecData: vtrack = vi.(av.H264CodecData) break case av.AACCodecData: atracks = append(atracks, vi.(av.AACCodecData)) break } } var streamDialers []av.MuxCloser var streamWriters []av.MuxCloser // 创建H264的PubSub并添加到H264 Pub集 H264Pub := pubsub.NewSimpleMuxer(100) streamDialers = append(streamDialers, H264Pub) go func() { H264Out := <-H264Pub.Out H264Outs, err := rtp.Encode(H264Out, vtrack.(av.VideoCodecData).Sdp()) checkError(err) defer H264Outs.Close() n, err := s.WriteInterleaved(H264Outs) checkError(err) fmt.Println("Sent", n, "bytes. H264") }() // 创建AAC的PubSub并添加到AAC Pub集 AACPubs := make([]*pubsub.PubSub, len(atracks)) for i, atrack := range atracks { AACPubs[i] = pubsub.NewSimpleMuxer(100) streamDialers = append(streamDialers, AACPubs[i]) go func(atrack av.CodecData, AACPubs1 *pubsub.PubSub) { out := <-AACPubs1.Out aacOut, _ := atrack.NewMuxer(out) defer aacOut.Close() outs, err := rtp.Encode(aacOut, atrack.(av.AudioCodecData).Sdp()) checkError(err) defer outs.Close() n, err := s.WriteInterleaved(outs) checkError(err) fmt.Println("Sent", n, "bytes. Audio") }(atrack, AACPubs[i]) } // 打开相应的转换器 if aacWrite != nil { streamWriters = append(streamWriters, aacWrite) pubAACOut := make(chan []byte) AACPubs[0].Out <- pubAACOut go func() { // 把音频包推送到channel，再写到文件 for { samples := <-pubAACOut _, err := aacWrite.Write(samples) checkError(err) } }() } if h264Write != nil { streamWriters = append(streamWriters, h264Write) H264Pub.Out <- h264Write } // 等待停止 <-s.Done() for _, dialer := range streamDialers { fmt.Println("Closing dialer") dialer.Close() } for _, writer := range streamWriters { fmt.Println("Closing writer") writer.Close() } } for { req, err := rtsp.NewRequest() checkError(err) s, err := rtsp.NewSession(req, rtspRequest) if err != nil { fmt.Println(err) break } sessionHandle(s) } }

... // 使用HTTP协议请求推送的HLS流 func processHTTP(w http.ResponseWriter, r *http.Request) { ctx := &av.Context{Logger: logger} fmt.Println("New connection") defer fmt.Println("Closing write") v := mux.Vars(r) streamName := v["name"] if r.Method == "GET" { fmt.Println("HTTP GET request received...") segSeq := 0 for { writer := NewHTTPStreamer(streamName, segSeq, w) segSeq++ pubsub1 := pubsub.NewSimpleMuxer(100) // 创建http请求推送流着音视频流 go func() { defer writer.Close() fmt.Println("Connected HTTP Writer. Waiting for output.") for { Out := <-pubsub1.Out fmt.Println("Received output") ctx := &av.Context{Logger: logger, Write: writer} fmt.Println(ctx) defer ctx.Flush() stream := Out[0] fmt.Println(stream) _ = avutil.CopyPackets(ctx, stream) } }() aacWrite, h264Write := getHLS(path.Join(videoTempDir, streamName), segSeq) processRTSP(NewRTSPReader(streamName), pubsub1, aacWrite, h264Write) } } } // 实现HTTP音视频流 func NewHTTPStreamer(base string, sn int, w http.ResponseWriter) *HTTPStreamer { str := fmt.Sprintf("%v/%v-%d.ts", videoTempDir, base, sn) f, _ := os.Create(str) return &HTTPStreamer{Writer: f, ResponseWriter: w} } ...

... // 连接RTSP请求和HTTP响应 func streamInvoke(w http.ResponseWriter, r *http.Request) { fmt.Println(r.URL.Path) if strings.HasPrefix(r.URL.Path, "/stream/") { processHTTP(w, r) } else if strings.HasPrefix(r.URL.Path, "/hls/") { processHLS(w, r) } else { fmt.Println(r.URL.Path) w.WriteHeader(404) } } ...