zishuo/rtsp-simple-server
1
0
Fork 0
mirror of https://github.com/aler9/rtsp-simple-server synced 2025-10-05 07:36:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

api: decode durations from strings instead of numbers

Browse Source 
This allows to use human-readable durations with the API,
for instance: "5s" instead of 5000000000
This commit is contained in:
aler9
2021-09-26 23:06:40 +02:00
parent 64808863ed
commit ca9645c2f9
19 changed files with 242 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
apidocs/openapi.yaml
View File

@@ -26,9 +26,9 @@ components:
logFile:
type: string
readTimeout:
type: integer
type: string
writeTimeout:
type: integer
type: string
readBufferCount:
type: integer
api:
@@ -98,7 +98,7 @@ components:
hlsSegmentCount:
type: integer
hlsSegmentDuration:
type: integer
type: string
hlsAllowOrigin:
type: string
@@ -122,9 +122,9 @@ components:
sourceOnDemand:
type: boolean
sourceOnDemandStartTimeout:
type: integer
type: string
sourceOnDemandCloseAfter:
type: integer
type: string
sourceRedirect:
type: string
disablePublisherOverride:
@@ -160,9 +160,9 @@ components:
runOnDemandRestart:
type: boolean
runOnDemandStartTimeout:
type: integer
type: string
runOnDemandCloseAfter:
type: integer
type: string
runOnPublish:
type: string
runOnPublishRestart:

12
internal/conf/conf.go
View File

@@ -61,8 +61,8 @@ type Conf struct {
LogDestinations []string `yaml:"logDestinations" json:"logDestinations"`
LogDestinationsParsed map[logger.Destination]struct{} `yaml:"-" json:"-"`
LogFile string `yaml:"logFile" json:"logFile"`
ReadTimeout time.Duration `yaml:"readTimeout" json:"readTimeout"`
WriteTimeout time.Duration `yaml:"writeTimeout" json:"writeTimeout"`
ReadTimeout StringDuration `yaml:"readTimeout" json:"readTimeout"`
WriteTimeout StringDuration `yaml:"writeTimeout" json:"writeTimeout"`
ReadBufferCount int `yaml:"readBufferCount" json:"readBufferCount"`
API bool `yaml:"api" json:"api"`
APIAddress string `yaml:"apiAddress" json:"apiAddress"`
@@ -101,7 +101,7 @@ type Conf struct {
HLSAddress string `yaml:"hlsAddress" json:"hlsAddress"`
HLSAlwaysRemux bool `yaml:"hlsAlwaysRemux" json:"hlsAlwaysRemux"`
HLSSegmentCount int `yaml:"hlsSegmentCount" json:"hlsSegmentCount"`
HLSSegmentDuration time.Duration `yaml:"hlsSegmentDuration" json:"hlsSegmentDuration"`
HLSSegmentDuration StringDuration `yaml:"hlsSegmentDuration" json:"hlsSegmentDuration"`
HLSAllowOrigin string `yaml:"hlsAllowOrigin" json:"hlsAllowOrigin"`
// paths
@@ -201,10 +201,10 @@ func (conf *Conf) CheckAndFillMissing() error {
conf.LogFile = "rtsp-simple-server.log"
}
if conf.ReadTimeout == 0 {
conf.ReadTimeout = 10 * time.Second
conf.ReadTimeout = 10 * StringDuration(time.Second)
}
if conf.WriteTimeout == 0 {
conf.WriteTimeout = 10 * time.Second
conf.WriteTimeout = 10 * StringDuration(time.Second)
}
if conf.ReadBufferCount == 0 {
conf.ReadBufferCount = 512
@@ -322,7 +322,7 @@ func (conf *Conf) CheckAndFillMissing() error {
conf.HLSSegmentCount = 3
}
if conf.HLSSegmentDuration == 0 {
conf.HLSSegmentDuration = 1 * time.Second
conf.HLSSegmentDuration = 1 * StringDuration(time.Second)
}
if conf.HLSAllowOrigin == "" {
conf.HLSAllowOrigin = "*"

47
internal/conf/conf_test.go
View File

@@ -28,7 +28,32 @@ func writeTempFile(byts []byte) (string, error) {
return tmpf.Name(), nil
}
func TestWithFileAndEnv(t *testing.T) {
func TestConfFromFile(t *testing.T) {
tmpf, err := writeTempFile([]byte(`
paths:
cam1:
runOnDemandStartTimeout: 5s
`))
require.NoError(t, err)
defer os.Remove(tmpf)
conf, hasFile, err := Load(tmpf)
require.NoError(t, err)
require.Equal(t, true, hasFile)
pa, ok := conf.Paths["cam1"]
require.Equal(t, true, ok)
require.Equal(t, &PathConf{
Source: "publisher",
SourceProtocol: "",
SourceOnDemandStartTimeout: 10 * StringDuration(time.Second),
SourceOnDemandCloseAfter: 10 * StringDuration(time.Second),
RunOnDemandStartTimeout: 5 * StringDuration(time.Second),
RunOnDemandCloseAfter: 10 * StringDuration(time.Second),
}, pa)
}
func TestConfFromFileAndEnv(t *testing.T) {
os.Setenv("RTSP_PATHS_CAM1_SOURCE", "rtsp://testing")
defer os.Unsetenv("RTSP_PATHS_CAM1_SOURCE")
@@ -45,14 +70,14 @@ func TestWithFileAndEnv(t *testing.T) {
require.Equal(t, &PathConf{
Source: "rtsp://testing",
SourceProtocol: "automatic",
SourceOnDemandStartTimeout: 10 * time.Second,
SourceOnDemandCloseAfter: 10 * time.Second,
RunOnDemandStartTimeout: 10 * time.Second,
RunOnDemandCloseAfter: 10 * time.Second,
SourceOnDemandStartTimeout: 10 * StringDuration(time.Second),
SourceOnDemandCloseAfter: 10 * StringDuration(time.Second),
RunOnDemandStartTimeout: 10 * StringDuration(time.Second),
RunOnDemandCloseAfter: 10 * StringDuration(time.Second),
}, pa)
}
func TestWithEnvOnly(t *testing.T) {
func TestConfFromEnvOnly(t *testing.T) {
os.Setenv("RTSP_PATHS_CAM1_SOURCE", "rtsp://testing")
defer os.Unsetenv("RTSP_PATHS_CAM1_SOURCE")
@@ -65,14 +90,14 @@ func TestWithEnvOnly(t *testing.T) {
require.Equal(t, &PathConf{
Source: "rtsp://testing",
SourceProtocol: "automatic",
SourceOnDemandStartTimeout: 10 * time.Second,
SourceOnDemandCloseAfter: 10 * time.Second,
RunOnDemandStartTimeout: 10 * time.Second,
RunOnDemandCloseAfter: 10 * time.Second,
SourceOnDemandStartTimeout: 10 * StringDuration(time.Second),
SourceOnDemandCloseAfter: 10 * StringDuration(time.Second),
RunOnDemandStartTimeout: 10 * StringDuration(time.Second),
RunOnDemandCloseAfter: 10 * StringDuration(time.Second),
}, pa)
}
func TestEncryption(t *testing.T) {
func TestConfEncryption(t *testing.T) {
key := "testing123testin"
plaintext := `
paths:

4
internal/conf/env.go
View File

@@ -12,13 +12,13 @@ import (
func loadEnvInternal(env map[string]string, prefix string, rv reflect.Value) error {
rt := rv.Type()
if rt == reflect.TypeOf(time.Duration(0)) {
if rt == reflect.TypeOf(StringDuration(0)) {
if ev, ok := env[prefix]; ok {
d, err := time.ParseDuration(ev)
if err != nil {
return fmt.Errorf("%s: %s", prefix, err)
}
rv.Set(reflect.ValueOf(d))
rv.Set(reflect.ValueOf(StringDuration(d)))
}
return nil
}

4
internal/conf/env_test.go
View File

@@ -23,7 +23,7 @@ type testStruct struct {
MyBool bool
// duration
MyDuration time.Duration
MyDuration StringDuration
// slice
MySlice []string
@@ -61,7 +61,7 @@ func TestEnvironment(t *testing.T) {
require.Equal(t, "testcontent", s.MyString)
require.Equal(t, 123, s.MyInt)
require.Equal(t, true, s.MyBool)
require.Equal(t, 22*time.Second, s.MyDuration)
require.Equal(t, 22*StringDuration(time.Second), s.MyDuration)
require.Equal(t, []string{"el1", "el2"}, s.MySlice)
_, ok := s.MyMap["mykey"]

16
internal/conf/path.go
View File

@@ -75,8 +75,8 @@ type PathConf struct {
SourceAnyPortEnable bool `yaml:"sourceAnyPortEnable" json:"sourceAnyPortEnable"`
SourceFingerprint string `yaml:"sourceFingerprint" json:"sourceFingerprint"`
SourceOnDemand bool `yaml:"sourceOnDemand" json:"sourceOnDemand"`
SourceOnDemandStartTimeout time.Duration `yaml:"sourceOnDemandStartTimeout" json:"sourceOnDemandStartTimeout"` //nolint:lll
SourceOnDemandCloseAfter time.Duration `yaml:"sourceOnDemandCloseAfter" json:"sourceOnDemandCloseAfter"`
SourceOnDemandStartTimeout StringDuration `yaml:"sourceOnDemandStartTimeout" json:"sourceOnDemandStartTimeout"` //nolint:lll
SourceOnDemandCloseAfter StringDuration `yaml:"sourceOnDemandCloseAfter" json:"sourceOnDemandCloseAfter"`
SourceRedirect string `yaml:"sourceRedirect" json:"sourceRedirect"`
DisablePublisherOverride bool `yaml:"disablePublisherOverride" json:"disablePublisherOverride"`
Fallback string `yaml:"fallback" json:"fallback"`
@@ -96,8 +96,8 @@ type PathConf struct {
RunOnInitRestart bool `yaml:"runOnInitRestart" json:"runOnInitRestart"`
RunOnDemand string `yaml:"runOnDemand" json:"runOnDemand"`
RunOnDemandRestart bool `yaml:"runOnDemandRestart" json:"runOnDemandRestart"`
RunOnDemandStartTimeout time.Duration `yaml:"runOnDemandStartTimeout" json:"runOnDemandStartTimeout"`
RunOnDemandCloseAfter time.Duration `yaml:"runOnDemandCloseAfter" json:"runOnDemandCloseAfter"`
RunOnDemandStartTimeout StringDuration `yaml:"runOnDemandStartTimeout" json:"runOnDemandStartTimeout"`
RunOnDemandCloseAfter StringDuration `yaml:"runOnDemandCloseAfter" json:"runOnDemandCloseAfter"`
RunOnPublish string `yaml:"runOnPublish" json:"runOnPublish"`
RunOnPublishRestart bool `yaml:"runOnPublishRestart" json:"runOnPublishRestart"`
RunOnRead string `yaml:"runOnRead" json:"runOnRead"`
@@ -236,11 +236,11 @@ func (pconf *PathConf) checkAndFillMissing(name string) error {
}
if pconf.SourceOnDemandStartTimeout == 0 {
pconf.SourceOnDemandStartTimeout = 10 * time.Second
pconf.SourceOnDemandStartTimeout = 10 * StringDuration(time.Second)
}
if pconf.SourceOnDemandCloseAfter == 0 {
pconf.SourceOnDemandCloseAfter = 10 * time.Second
pconf.SourceOnDemandCloseAfter = 10 * StringDuration(time.Second)
}
if pconf.Fallback != "" {
@@ -337,11 +337,11 @@ func (pconf *PathConf) checkAndFillMissing(name string) error {
}
if pconf.RunOnDemandStartTimeout == 0 {
pconf.RunOnDemandStartTimeout = 10 * time.Second
pconf.RunOnDemandStartTimeout = 10 * StringDuration(time.Second)
}
if pconf.RunOnDemandCloseAfter == 0 {
pconf.RunOnDemandCloseAfter = 10 * time.Second
pconf.RunOnDemandCloseAfter = 10 * StringDuration(time.Second)
}
return nil

65
internal/conf/stringduration.go Normal file
View File

@@ -0,0 +1,65 @@
package conf
import (
"encoding/json"
"errors"
"time"
"gopkg.in/yaml.v2"
)
// StringDuration is a duration that is unmarshaled from a string.
// Durations are normally unmarshaled from numbers.
type StringDuration time.Duration
// MarshalJSON marshals a StringDuration into a string.
func (d StringDuration) MarshalJSON() ([]byte, error) {
return json.Marshal(time.Duration(d).String())
}
// UnmarshalJSON unmarshals a StringDuration from a string.
func (d *StringDuration) UnmarshalJSON(b []byte) error {
var v interface{}
if err := json.Unmarshal(b, &v); err != nil {
return err
}
value, ok := v.(string)
if !ok {
return errors.New("invalid duration")
}
du, err := time.ParseDuration(value)
if err != nil {
return err
}
*d = StringDuration(du)
return nil
}
// MarshalYAML marshals a StringDuration into a string.
func (d StringDuration) MarshalYAML() ([]byte, error) {
return yaml.Marshal(time.Duration(d).String())
}
// UnmarshalYAML unmarshals a StringDuration from a string.
func (d *StringDuration) UnmarshalYAML(unmarshal func(interface{}) error) error {
var v interface{}
if err := unmarshal(&v); err != nil {
return err
}
value, ok := v.(string)
if !ok {
return errors.New("invalid duration")
}
du, err := time.ParseDuration(value)
if err != nil {
return err
}
*d = StringDuration(du)
return nil
}

15
internal/core/api.go
View File

@@ -10,7 +10,6 @@ import (
"net/http/httputil"
"reflect"
"sync"
"time"
"github.com/gin-gonic/gin"
@@ -50,8 +49,8 @@ func loadConfData(ctx *gin.Context) (interface{}, error) {
LogLevel *string `json:"logLevel"`
LogDestinations *[]string `json:"logDestinations"`
LogFile *string `json:"logFile"`
ReadTimeout *time.Duration `json:"readTimeout"`
WriteTimeout *time.Duration `json:"writeTimeout"`
ReadTimeout *conf.StringDuration `json:"readTimeout"`
WriteTimeout *conf.StringDuration `json:"writeTimeout"`
ReadBufferCount *int `json:"readBufferCount"`
API *bool `json:"api"`
APIAddress *string `json:"apiAddress"`
@@ -87,7 +86,7 @@ func loadConfData(ctx *gin.Context) (interface{}, error) {
HLSAddress *string `json:"hlsAddress"`
HLSAlwaysRemux *bool `json:"hlsAlwaysRemux"`
HLSSegmentCount *int `json:"hlsSegmentCount"`
HLSSegmentDuration *time.Duration `json:"hlsSegmentDuration"`
HLSSegmentDuration *conf.StringDuration `json:"hlsSegmentDuration"`
HLSAllowOrigin *string `json:"hlsAllowOrigin"`
}
err := json.NewDecoder(ctx.Request.Body).Decode(&in)
@@ -106,8 +105,8 @@ func loadConfPathData(ctx *gin.Context) (interface{}, error) {
SourceAnyPortEnable *bool `json:"sourceAnyPortEnable"`
SourceFingerprint *string `json:"sourceFingerprint"`
SourceOnDemand *bool `json:"sourceOnDemand"`
SourceOnDemandStartTimeout *time.Duration `json:"sourceOnDemandStartTimeout"`
SourceOnDemandCloseAfter *time.Duration `json:"sourceOnDemandCloseAfter"`
SourceOnDemandStartTimeout *conf.StringDuration `json:"sourceOnDemandStartTimeout"`
SourceOnDemandCloseAfter *conf.StringDuration `json:"sourceOnDemandCloseAfter"`
SourceRedirect *string `json:"sourceRedirect"`
DisablePublisherOverride *bool `json:"disablePublisherOverride"`
Fallback *string `json:"fallback"`
@@ -125,8 +124,8 @@ func loadConfPathData(ctx *gin.Context) (interface{}, error) {
RunOnInitRestart *bool `json:"runOnInitRestart"`
RunOnDemand *string `json:"runOnDemand"`
RunOnDemandRestart *bool `json:"runOnDemandRestart"`
RunOnDemandStartTimeout *time.Duration `json:"runOnDemandStartTimeout"`
RunOnDemandCloseAfter *time.Duration `json:"runOnDemandCloseAfter"`
RunOnDemandStartTimeout *conf.StringDuration `json:"runOnDemandStartTimeout"`
RunOnDemandCloseAfter *conf.StringDuration `json:"runOnDemandCloseAfter"`
RunOnPublish *string `json:"runOnPublish"`
RunOnPublishRestart *bool `json:"runOnPublishRestart"`
RunOnRead *string `json:"runOnRead"`

2
internal/core/api_test.go
View File

@@ -71,6 +71,7 @@ func TestAPIConfigSet(t *testing.T) {
err := httpRequest(http.MethodPost, "http://localhost:9997/v1/config/set", map[string]interface{}{
"rtmpDisable": true,
"readTimeout": "7s",
}, nil)
require.NoError(t, err)
@@ -80,6 +81,7 @@ func TestAPIConfigSet(t *testing.T) {
err = httpRequest(http.MethodGet, "http://localhost:9997/v1/config/get", nil, &out)
require.NoError(t, err)
require.Equal(t, true, out["rtmpDisable"])
require.Equal(t, "7s", out["readTimeout"])
}
func TestAPIConfigPathsAdd(t *testing.T) {

7
internal/core/hls_muxer.go
View File

@@ -18,6 +18,7 @@ import (
"github.com/aler9/gortsplib/pkg/rtph264"
"github.com/pion/rtp"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/hls"
"github.com/aler9/rtsp-simple-server/internal/logger"
)
@@ -117,7 +118,7 @@ type hlsMuxerParent interface {
type hlsMuxer struct {
hlsAlwaysRemux bool
hlsSegmentCount int
hlsSegmentDuration time.Duration
hlsSegmentDuration conf.StringDuration
readBufferCount int
wg *sync.WaitGroup
pathName string
@@ -140,7 +141,7 @@ func newHLSMuxer(
parentCtx context.Context,
hlsAlwaysRemux bool,
hlsSegmentCount int,
hlsSegmentDuration time.Duration,
hlsSegmentDuration conf.StringDuration,
readBufferCount int,
wg *sync.WaitGroup,
pathName string,
@@ -300,7 +301,7 @@ func (r *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{})
var err error
r.muxer, err = hls.NewMuxer(
r.hlsSegmentCount,
r.hlsSegmentDuration,
time.Duration(r.hlsSegmentDuration),
videoTrack,
audioTrack,
)

6
internal/core/hls_server.go
View File

@@ -8,8 +8,8 @@ import (
gopath "path"
"strings"
"sync"
"time"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/logger"
)
@@ -20,7 +20,7 @@ type hlsServerParent interface {
type hlsServer struct {
hlsAlwaysRemux bool
hlsSegmentCount int
hlsSegmentDuration time.Duration
hlsSegmentDuration conf.StringDuration
hlsAllowOrigin string
readBufferCount int
pathManager *pathManager
@@ -43,7 +43,7 @@ func newHLSServer(
address string,
hlsAlwaysRemux bool,
hlsSegmentCount int,
hlsSegmentDuration time.Duration,
hlsSegmentDuration conf.StringDuration,
hlsAllowOrigin string,
readBufferCount int,
pathManager *pathManager,

16
internal/core/path.go
View File

@@ -182,8 +182,8 @@ type pathPublisherPauseReq struct {
type path struct {
rtspAddress string
readTimeout time.Duration
writeTimeout time.Duration
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
readBufferSize int
confName string
@@ -226,8 +226,8 @@ type path struct {
func newPath(
parentCtx context.Context,
rtspAddress string,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
readBufferSize int,
confName string,
@@ -478,7 +478,7 @@ func (pa *path) onDemandStartSource() {
pa.onDemandReadyTimer.Stop()
if pa.hasStaticSource() {
pa.staticSourceCreate()
pa.onDemandReadyTimer = time.NewTimer(pa.conf.SourceOnDemandStartTimeout)
pa.onDemandReadyTimer = time.NewTimer(time.Duration(pa.conf.SourceOnDemandStartTimeout))
} else {
pa.Log(logger.Info, "on demand command started")
@@ -487,7 +487,7 @@ func (pa *path) onDemandStartSource() {
Path: pa.name,
Port: port,
})
pa.onDemandReadyTimer = time.NewTimer(pa.conf.RunOnDemandStartTimeout)
pa.onDemandReadyTimer = time.NewTimer(time.Duration(pa.conf.RunOnDemandStartTimeout))
}
pa.onDemandState = pathOnDemandStateWaitingReady
@@ -496,9 +496,9 @@ func (pa *path) onDemandStartSource() {
func (pa *path) onDemandScheduleClose() {
pa.onDemandCloseTimer.Stop()
if pa.hasStaticSource() {
pa.onDemandCloseTimer = time.NewTimer(pa.conf.SourceOnDemandCloseAfter)
pa.onDemandCloseTimer = time.NewTimer(time.Duration(pa.conf.SourceOnDemandCloseAfter))
} else {
pa.onDemandCloseTimer = time.NewTimer(pa.conf.RunOnDemandCloseAfter)
pa.onDemandCloseTimer = time.NewTimer(time.Duration(pa.conf.RunOnDemandCloseAfter))
}
pa.onDemandState = pathOnDemandStateClosing

9
internal/core/path_manager.go
View File

@@ -5,7 +5,6 @@ import (
"fmt"
"net"
"sync"
"time"
"github.com/aler9/gortsplib/pkg/base"
@@ -23,8 +22,8 @@ type pathManagerParent interface {
type pathManager struct {
rtspAddress string
readTimeout time.Duration
writeTimeout time.Duration
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
readBufferSize int
pathConfs map[string]*conf.PathConf
@@ -52,8 +51,8 @@ type pathManager struct {
func newPathManager(
parentCtx context.Context,
rtspAddress string,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
readBufferSize int,
pathConfs map[string]*conf.PathConf,

23
internal/core/rtmp_conn.go
View File

@@ -19,6 +19,7 @@ import (
"github.com/notedit/rtmp/av"
"github.com/pion/rtp"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/externalcmd"
"github.com/aler9/rtsp-simple-server/internal/logger"
"github.com/aler9/rtsp-simple-server/internal/rtcpsenderset"
@@ -55,8 +56,8 @@ type rtmpConnParent interface {
type rtmpConn struct {
id string
rtspAddress string
readTimeout time.Duration
writeTimeout time.Duration
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
runOnConnect string
runOnConnectRestart bool
@@ -77,8 +78,8 @@ func newRTMPConn(
parentCtx context.Context,
id string,
rtspAddress string,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
runOnConnect string,
runOnConnectRestart bool,
@@ -184,8 +185,8 @@ func (c *rtmpConn) runInner(ctx context.Context) error {
c.conn.NetConn().Close()
}()
c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout))
c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout))
c.conn.NetConn().SetReadDeadline(time.Now().Add(time.Duration(c.readTimeout)))
c.conn.NetConn().SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
err := c.conn.ServerHandshake()
if err != nil {
return err
@@ -262,7 +263,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
return fmt.Errorf("the stream doesn't contain an H264 track or an AAC track")
}
c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout))
c.conn.NetConn().SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
c.conn.WriteMetadata(videoTrack, audioTrack)
c.ringBuffer = ringbuffer.New(uint64(c.readBufferCount))
@@ -351,7 +352,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
pts -= videoStartPTS
dts := videoDTSEst.Feed(pts)
c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout))
c.conn.NetConn().SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
err = c.conn.WritePacket(av.Packet{
Type: av.H264,
Data: data,
@@ -391,7 +392,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
}
for _, au := range aus {
c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout))
c.conn.NetConn().SetWriteDeadline(time.Now().Add(time.Duration(c.writeTimeout)))
err := c.conn.WritePacket(av.Packet{
Type: av.AAC,
Data: au,
@@ -408,7 +409,7 @@ func (c *rtmpConn) runRead(ctx context.Context) error {
}
func (c *rtmpConn) runPublish(ctx context.Context) error {
c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout))
c.conn.NetConn().SetReadDeadline(time.Now().Add(time.Duration(c.readTimeout)))
videoTrack, audioTrack, err := c.conn.ReadMetadata()
if err != nil {
return err
@@ -483,7 +484,7 @@ func (c *rtmpConn) runPublish(ctx context.Context) error {
}
for {
c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout))
c.conn.NetConn().SetReadDeadline(time.Now().Add(time.Duration(c.readTimeout)))
pkt, err := c.conn.ReadPacket()
if err != nil {
return err

10
internal/core/rtmp_server.go
View File

@@ -8,10 +8,10 @@ import (
"net"
"strconv"
"sync"
"time"
"github.com/aler9/gortsplib"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/logger"
)
@@ -20,8 +20,8 @@ type rtmpServerParent interface {
}
type rtmpServer struct {
readTimeout time.Duration
writeTimeout time.Duration
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
rtspAddress string
runOnConnect string
@@ -45,8 +45,8 @@ type rtmpServer struct {
func newRTMPServer(
parentCtx context.Context,
address string,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
rtspAddress string,
runOnConnect string,

19
internal/core/rtmp_source.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/aler9/gortsplib/pkg/rtph264"
"github.com/notedit/rtmp/av"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/logger"
"github.com/aler9/rtsp-simple-server/internal/rtcpsenderset"
"github.com/aler9/rtsp-simple-server/internal/rtmp"
@@ -29,8 +30,8 @@ type rtmpSourceParent interface {
type rtmpSource struct {
ur string
readTimeout time.Duration
writeTimeout time.Duration
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
wg *sync.WaitGroup
parent rtmpSourceParent
@@ -41,8 +42,8 @@ type rtmpSource struct {
func newRTMPSource(
parentCtx context.Context,
ur string,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
wg *sync.WaitGroup,
parent rtmpSourceParent) *rtmpSource {
ctx, ctxCancel := context.WithCancel(parentCtx)
@@ -103,7 +104,7 @@ func (s *rtmpSource) runInner() bool {
runErr <- func() error {
s.log(logger.Debug, "connecting")
ctx2, cancel2 := context.WithTimeout(innerCtx, s.readTimeout)
ctx2, cancel2 := context.WithTimeout(innerCtx, time.Duration(s.readTimeout))
defer cancel2()
conn, err := rtmp.DialContext(ctx2, s.ur)
@@ -114,8 +115,8 @@ func (s *rtmpSource) runInner() bool {
readDone := make(chan error)
go func() {
readDone <- func() error {
conn.NetConn().SetReadDeadline(time.Now().Add(s.readTimeout))
conn.NetConn().SetWriteDeadline(time.Now().Add(s.writeTimeout))
conn.NetConn().SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
conn.NetConn().SetWriteDeadline(time.Now().Add(time.Duration(s.writeTimeout)))
err = conn.ClientHandshake()
if err != nil {
return err
@@ -123,7 +124,7 @@ func (s *rtmpSource) runInner() bool {
conn.NetConn().SetWriteDeadline(time.Time{})
conn.NetConn().SetReadDeadline(time.Now().Add(s.readTimeout))
conn.NetConn().SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
videoTrack, audioTrack, err := conn.ReadMetadata()
if err != nil {
return err
@@ -171,7 +172,7 @@ func (s *rtmpSource) runInner() bool {
}
for {
conn.NetConn().SetReadDeadline(time.Now().Add(s.readTimeout))
conn.NetConn().SetReadDeadline(time.Now().Add(time.Duration(s.readTimeout)))
pkt, err := conn.ReadPacket()
if err != nil {
return err

5
internal/core/rtsp_conn.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/aler9/gortsplib/pkg/headers"
"github.com/aler9/gortsplib/pkg/liberrors"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/externalcmd"
"github.com/aler9/rtsp-simple-server/internal/logger"
)
@@ -37,7 +38,7 @@ type rtspConnParent interface {
type rtspConn struct {
rtspAddress string
authMethods []headers.AuthMethod
readTimeout time.Duration
readTimeout conf.StringDuration
runOnConnect string
runOnConnectRestart bool
pathManager *pathManager
@@ -54,7 +55,7 @@ type rtspConn struct {
func newRTSPConn(
rtspAddress string,
authMethods []headers.AuthMethod,
readTimeout time.Duration,
readTimeout conf.StringDuration,
runOnConnect string,
runOnConnectRestart bool,
pathManager *pathManager,

10
internal/core/rtsp_server.go
View File

@@ -24,7 +24,7 @@ type rtspServerParent interface {
type rtspServer struct {
authMethods []headers.AuthMethod
readTimeout time.Duration
readTimeout conf.StringDuration
isTLS bool
rtspAddress string
protocols map[conf.Protocol]struct{}
@@ -47,8 +47,8 @@ func newRTSPServer(
parentCtx context.Context,
address string,
authMethods []headers.AuthMethod,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
readBufferSize int,
useUDP bool,
@@ -87,8 +87,8 @@ func newRTSPServer(
s.srv = &gortsplib.Server{
Handler: s,
ReadTimeout: readTimeout,
WriteTimeout: writeTimeout,
ReadTimeout: time.Duration(readTimeout),
WriteTimeout: time.Duration(writeTimeout),
ReadBufferCount: readBufferCount,
ReadBufferSize: readBufferSize,
}

13
internal/core/rtsp_source.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/aler9/gortsplib"
"github.com/aler9/gortsplib/pkg/base"
"github.com/aler9/rtsp-simple-server/internal/conf"
"github.com/aler9/rtsp-simple-server/internal/logger"
)
@@ -31,8 +32,8 @@ type rtspSource struct {
proto *gortsplib.ClientProtocol
anyPortEnable bool
fingerprint string
readTimeout time.Duration
writeTimeout time.Duration
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
readBufferSize int
wg *sync.WaitGroup
@@ -48,8 +49,8 @@ func newRTSPSource(
proto *gortsplib.ClientProtocol,
anyPortEnable bool,
fingerprint string,
readTimeout time.Duration,
writeTimeout time.Duration,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
readBufferSize int,
wg *sync.WaitGroup,
@@ -134,8 +135,8 @@ func (s *rtspSource) runInner() bool {
return nil
},
},
ReadTimeout: s.readTimeout,
WriteTimeout: s.writeTimeout,
ReadTimeout: time.Duration(s.readTimeout),
WriteTimeout: time.Duration(s.writeTimeout),
ReadBufferCount: s.readBufferCount,
ReadBufferSize: s.readBufferSize,
AnyPortEnable: s.anyPortEnable,