example.go

package main

import "C"

import (
	"fmt"
	"net/http"
	"os"

	"github.com/gin-contrib/static"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"github.com/joho/godotenv"
	gstreamer "github.com/notedit/gstreamer-go"
	mediaserver "github.com/notedit/media-server-go"
	"github.com/notedit/sdp"
)

var pipelineStr = "appsrc do-timestamp=true is-live=true  name=appsrc ! h264parse" +
	" ! mpegtsmux name=muxer ! hlssink max-files=10 target-duration=5"

var pipelineStrBak = "appsrc format=time is-live=true do-timestamp=true name=appsrc " +
	"! application/x-rtp,encoding-name=VP8-DRAFT-IETF-01 ! rtpvp8depay ! decodebin " +
	" ! videoconvert ! videoscale !  videorate ! video/x-raw,framerate=30/1,format=I420 " +
	" ! vp8enc lag-in-frames=0 cpu-used=5 threads=4 static-threshold=2000 end-usage=vbr target-bitrate=500000 min-quantizer=4 max-quantizer=56 " +
	" ! appsink name=appsink sync=false"
	//" ! mpegtsmux name=muxer ! hlssink max-files=10 target-duration=5"
	////"! queue ! x264enc tune=zerolatency "+
	//"! fakesink"

type Message struct {
	Cmd string `json:"cmd,omitempty"`
	Sdp string `json:"sdp,omitempty"`
}

var upGrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

var Capabilities = map[string]*sdp.Capability{
	"audio": &sdp.Capability{
		Codecs: []string{"opus"},
	},
	"video": &sdp.Capability{
		Codecs: []string{"vp8"},
		Rtx:    true,
		Rtcpfbs: []*sdp.RtcpFeedback{
			&sdp.RtcpFeedback{
				ID: "goog-remb",
			},
			&sdp.RtcpFeedback{
				ID: "transport-cc",
			},
			&sdp.RtcpFeedback{
				ID:     "ccm",
				Params: []string{"fir"},
			},
			&sdp.RtcpFeedback{
				ID:     "nack",
				Params: []string{"pli"},
			},
		},
		Extensions: []string{
			"urn:3gpp:video-orientation",
			"http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01",
			"http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time",
			"urn:ietf:params:rtp-hdrext:toffse",
			"urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id",
			"urn:ietf:params:rtp-hdrext:sdes:mid",
		},
	},
}

func channel(c *gin.Context) {

	ws, err := upGrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		return
	}
	defer ws.Close()

	var transport *mediaserver.Transport
	endpoint := mediaserver.NewEndpoint("127.0.0.1")

	for {
		// read json
		var msg Message
		err = ws.ReadJSON(&msg)
		if err != nil {
			fmt.Println("error: ", err)
			break
		}

		if msg.Cmd == "offer" {
			offer, err := sdp.Parse(msg.Sdp)
			if err != nil {
				panic(err)
			}
			transport = endpoint.CreateTransport(offer, nil)
			transport.SetRemoteProperties(offer.GetMedia("audio"), offer.GetMedia("video"))

			// answer := offer.Answer(transport.GetLocalICEInfo(),
			// 	transport.GetLocalDTLSInfo(),
			// 	endpoint.GetLocalCandidates(),
			// 	Capabilities)
			ice := transport.GetLocalICEInfo()
			dtls := transport.GetLocalDTLSInfo()
			candidates := endpoint.GetLocalCandidates()

			answer := sdp.NewSDPInfo()
			answer.SetICE(ice)
			answer.SetDTLS(dtls)
			answer.AddCandidates(candidates)

			if offer.GetMedia("audio") != nil {
				audioMedia := offer.GetMedia("audio").AnswerCapability(Capabilities["audio"])
				answer.AddMedia(audioMedia)
			}

			if offer.GetMedia("video") != nil {
				videoMedia := offer.GetMedia("video").AnswerCapability(Capabilities["video"])
				answer.AddMedia(videoMedia)
			}

			transport.SetLocalProperties(answer.GetMedia("audio"), answer.GetMedia("video"))

			for _, stream := range offer.GetStreams() {
				incomingStream := transport.CreateIncomingStream(stream)

				refresher := mediaserver.NewRefresher(2000)
				refresher.AddStream(incomingStream)

				// Response stream
				outgoingStream := transport.CreateOutgoingStream(stream.Clone())
				outgoingStream.AttachTo(incomingStream)
				answer.AddStream(outgoingStream.GetStreamInfo())

				if len(incomingStream.GetVideoTracks()) > 0 {
					videoTrack := incomingStream.GetVideoTracks()[0]

					pipeline, err := gstreamer.New(pipelineStr)
					if err != nil {
						panic(err)
					}

					appsrc := pipeline.FindElement("appsrc")
					pipeline.Start()

					continue
					videoTrack.OnMediaFrame(func(frame []byte, timestamp uint) {
						//fmt.Println("media frame ===========")
						if len(frame) <= 4 {
							fmt.Println("media frame short ===========", len(frame))
							fmt.Println(string(frame))
							return
						}
						appsrc.Push(frame)
					})
				}
			}

			ws.WriteJSON(Message{
				Cmd: "answer",
				Sdp: answer.String(),
			})
		}
	}
}

func index(c *gin.Context) {
	fmt.Println("helloworld")
	c.HTML(http.StatusOK, "index.html", gin.H{})
}

func main() {
	godotenv.Load()
	//mediaserver.EnableDebug(true)
	//mediaserver.EnableLog(true)

	address := ":8000"
	if os.Getenv("PORT") != "" {
		address = ":" + os.Getenv("PORT")
	}
	r := gin.Default()
	r.Use(static.Serve("/", static.LocalFile("./", false)))
	r.LoadHTMLFiles("./index.html")
	r.GET("/channel", channel)
	r.GET("/", index)
	r.Run(address)
}