mirror of https://github.com/q191201771/lal.git
messages:
- [feat] rtsp，实现PushSession - [feat] demo，新增app/demo/pullrtsp2pushrtsp，可拉取rtsp流，并使用rtsp转推出去 - [refactor] rtsp，重构部分逻辑，聚合至sdp.LogicContext中pull/44/head
q191201771
4 years ago
parent
7dcd4a6b73
commit
766573741e
@ -0,0 +1,101 @@
|
||||
// Copyright 2020, Chef. All rights reserved.
|
||||
// https://github.com/q191201771/lal
|
||||
//
|
||||
// Use of this source code is governed by a MIT-style license
|
||||
// that can be found in the License file.
|
||||
//
|
||||
// Author: Chef (191201771@qq.com)
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/q191201771/lal/pkg/base"
|
||||
"github.com/q191201771/lal/pkg/rtprtcp"
|
||||
"github.com/q191201771/lal/pkg/rtsp"
|
||||
"github.com/q191201771/naza/pkg/nazalog"
|
||||
)
|
||||
|
||||
var rtpPacketChan = make(chan rtprtcp.RTPPacket, 1024)
|
||||
|
||||
type Observer struct {
|
||||
}
|
||||
|
||||
func (o *Observer) OnRTPPacket(pkt rtprtcp.RTPPacket) {
|
||||
rtpPacketChan <- pkt
|
||||
}
|
||||
|
||||
func (o *Observer) OnAVConfig(asc, vps, sps, pps []byte) {
|
||||
// noop
|
||||
}
|
||||
|
||||
func (o *Observer) OnAVPacket(pkt base.AVPacket) {
|
||||
// noop
|
||||
}
|
||||
|
||||
func main() {
|
||||
_ = nazalog.Init(func(option *nazalog.Option) {
|
||||
option.AssertBehavior = nazalog.AssertFatal
|
||||
})
|
||||
|
||||
inURL, outURL := parseFlag()
|
||||
|
||||
o := &Observer{}
|
||||
rtspPullSession := rtsp.NewPullSession(o, func(option *rtsp.PullSessionOption) {
|
||||
option.PullTimeoutMS = 5000
|
||||
option.OverTCP = false
|
||||
})
|
||||
|
||||
rtspPushSession := rtsp.NewPushSession(func(option *rtsp.PushSessionOption) {
|
||||
option.PushTimeoutMS = 5000
|
||||
option.OverTCP = false
|
||||
})
|
||||
|
||||
go func() {
|
||||
time.Sleep(3 * time.Second)
|
||||
for {
|
||||
rtspPullSession.UpdateStat(1)
|
||||
rtspStat := rtspPullSession.GetStat()
|
||||
nazalog.Debugf("bitrate. rtsp pull=%dkbit/s, rtsp push=", rtspStat.Bitrate)
|
||||
time.Sleep(1 * time.Second)
|
||||
}
|
||||
}()
|
||||
|
||||
err := rtspPullSession.Pull(inURL)
|
||||
nazalog.Assert(nil, err)
|
||||
rawSDP, sdpLogicCtx := rtspPullSession.GetSDP()
|
||||
|
||||
err = rtspPushSession.Push(outURL, rawSDP, sdpLogicCtx)
|
||||
nazalog.Assert(nil, err)
|
||||
|
||||
for {
|
||||
select {
|
||||
case err = <-rtspPullSession.Wait():
|
||||
nazalog.Infof("pull rtsp done. err=%+v", err)
|
||||
return
|
||||
case err = <-rtspPushSession.Wait():
|
||||
nazalog.Infof("push rtsp done. err=%+v", err)
|
||||
return
|
||||
case pkt := <-rtpPacketChan:
|
||||
rtspPushSession.WriteRTPPacket(pkt)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func parseFlag() (inURL string, outURL string) {
|
||||
i := flag.String("i", "", "specify pull rtsp url")
|
||||
o := flag.String("o", "", "specify push rtmp url")
|
||||
flag.Parse()
|
||||
if *i == "" || *o == "" {
|
||||
flag.Usage()
|
||||
_, _ = fmt.Fprintf(os.Stderr, `Example:
|
||||
./bin/pullrtsp2pushrtsp -i rtsp://localhost:5544/live/test110 -o rtsp://localhost:5544/live/test220
|
||||
`)
|
||||
os.Exit(1)
|
||||
}
|
||||
return *i, *o
|
||||
}
|
||||
@ -0,0 +1,203 @@
|
||||
// Copyright 2020, Chef. All rights reserved.
|
||||
// https://github.com/q191201771/lal
|
||||
//
|
||||
// Use of this source code is governed by a MIT-style license
|
||||
// that can be found in the License file.
|
||||
//
|
||||
// Author: Chef (191201771@qq.com)
|
||||
|
||||
package sdp
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type RawContext struct {
|
||||
MediaDescList []MediaDesc
|
||||
}
|
||||
|
||||
type MediaDesc struct {
|
||||
M M
|
||||
ARTPMap ARTPMap
|
||||
AFmtPBase *AFmtPBase
|
||||
AControl AControl
|
||||
}
|
||||
|
||||
type M struct {
|
||||
Media string
|
||||
}
|
||||
|
||||
type ARTPMap struct {
|
||||
PayloadType int
|
||||
EncodingName string
|
||||
ClockRate int
|
||||
EncodingParameters string
|
||||
}
|
||||
|
||||
type AFmtPBase struct {
|
||||
Format int // same as PayloadType
|
||||
Parameters map[string]string // name -> value
|
||||
}
|
||||
|
||||
type AControl struct {
|
||||
Value string
|
||||
}
|
||||
|
||||
// 例子见单元测试
|
||||
func ParseSDP2RawContext(b []byte) (RawContext, error) {
|
||||
var (
|
||||
sdpCtx RawContext
|
||||
md *MediaDesc
|
||||
)
|
||||
|
||||
s := string(b)
|
||||
lines := strings.Split(s, "\r\n")
|
||||
for _, line := range lines {
|
||||
if strings.HasPrefix(line, "m=") {
|
||||
m, err := ParseM(line)
|
||||
if err != nil {
|
||||
return sdpCtx, err
|
||||
}
|
||||
if md != nil {
|
||||
sdpCtx.MediaDescList = append(sdpCtx.MediaDescList, *md)
|
||||
}
|
||||
md = &MediaDesc{
|
||||
M: m,
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(line, "a=rtpmap") {
|
||||
aRTPMap, err := ParseARTPMap(line)
|
||||
if err != nil {
|
||||
return sdpCtx, err
|
||||
}
|
||||
if md == nil {
|
||||
continue
|
||||
}
|
||||
md.ARTPMap = aRTPMap
|
||||
}
|
||||
if strings.HasPrefix(line, "a=fmtp") {
|
||||
aFmtPBase, err := ParseAFmtPBase(line)
|
||||
if err != nil {
|
||||
return sdpCtx, err
|
||||
}
|
||||
if md == nil {
|
||||
continue
|
||||
}
|
||||
md.AFmtPBase = &aFmtPBase
|
||||
}
|
||||
if strings.HasPrefix(line, "a=control") {
|
||||
aControl, err := ParseAControl(line)
|
||||
if err != nil {
|
||||
return sdpCtx, err
|
||||
}
|
||||
if md == nil {
|
||||
continue
|
||||
}
|
||||
md.AControl = aControl
|
||||
}
|
||||
}
|
||||
if md != nil {
|
||||
sdpCtx.MediaDescList = append(sdpCtx.MediaDescList, *md)
|
||||
}
|
||||
|
||||
return sdpCtx, nil
|
||||
}
|
||||
|
||||
func ParseM(s string) (ret M, err error) {
|
||||
ss := strings.TrimPrefix(s, "m=")
|
||||
items := strings.Split(ss, " ")
|
||||
if len(items) < 1 {
|
||||
return ret, ErrSDP
|
||||
}
|
||||
ret.Media = items[0]
|
||||
return
|
||||
}
|
||||
|
||||
// 例子见单元测试
|
||||
func ParseARTPMap(s string) (ret ARTPMap, err error) {
|
||||
// rfc 3640 3.3.1. General
|
||||
// rfc 3640 3.3.6. High Bit-rate AAC
|
||||
//
|
||||
// a=rtpmap:<payload type> <encoding name>/<clock rate>[/<encoding parameters>]
|
||||
//
|
||||
|
||||
items := strings.SplitN(s, ":", 2)
|
||||
if len(items) != 2 {
|
||||
err = ErrSDP
|
||||
return
|
||||
}
|
||||
items = strings.SplitN(items[1], " ", 2)
|
||||
if len(items) != 2 {
|
||||
err = ErrSDP
|
||||
return
|
||||
}
|
||||
ret.PayloadType, err = strconv.Atoi(items[0])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
items = strings.SplitN(items[1], "/", 3)
|
||||
switch len(items) {
|
||||
case 3:
|
||||
ret.EncodingParameters = items[2]
|
||||
fallthrough
|
||||
case 2:
|
||||
ret.EncodingName = items[0]
|
||||
ret.ClockRate, err = strconv.Atoi(items[1])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
default:
|
||||
err = ErrSDP
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// 例子见单元测试
|
||||
func ParseAFmtPBase(s string) (ret AFmtPBase, err error) {
|
||||
// rfc 3640 4.4.1. The a=fmtp Keyword
|
||||
//
|
||||
// a=fmtp:<format> <parameter name>=<value>[; <parameter name>=<value>]
|
||||
//
|
||||
|
||||
ret.Parameters = make(map[string]string)
|
||||
|
||||
items := strings.SplitN(s, ":", 2)
|
||||
if len(items) != 2 {
|
||||
err = ErrSDP
|
||||
return
|
||||
}
|
||||
|
||||
items = strings.SplitN(items[1], " ", 2)
|
||||
if len(items) != 2 {
|
||||
err = ErrSDP
|
||||
return
|
||||
}
|
||||
|
||||
ret.Format, err = strconv.Atoi(items[0])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
items = strings.Split(items[1], ";")
|
||||
for _, pp := range items {
|
||||
pp = strings.TrimSpace(pp)
|
||||
kv := strings.SplitN(pp, "=", 2)
|
||||
if len(kv) != 2 {
|
||||
err = ErrSDP
|
||||
return
|
||||
}
|
||||
ret.Parameters[kv[0]] = kv[1]
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func ParseAControl(s string) (ret AControl, err error) {
|
||||
if !strings.HasPrefix(s, "a=control:") {
|
||||
err = ErrSDP
|
||||
return
|
||||
}
|
||||
ret.Value = strings.TrimPrefix(s, "a=control:")
|
||||
return
|
||||
}
|
||||
Loading…
Reference in New Issue