mirror
/
av
mirror of https://bitbucket.org/ausocean/av.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

revid: rename KeyRB* and RB* to refer to pool 

This also changes lexically significant strings in configuration text.
This commit is contained in:
Dan Kortschak 2021-05-17 11:18:44 +09:30
parent 7541015730
commit 05a6b4435a
No known key found for this signature in database
GPG Key ID: 45817BAD958E3F0C
11 changed files with 172 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
exp/rvcl/config.json
View File

@ -7,5 +7,5 @@
"TimelapseInterval":"10", "TimelapseInterval":"10",
"logging":"Debug", "logging":"Debug",
"Suppress":"false", "Suppress":"false",
"RBStartElementSize":"1000000" "PoolStartElementSize":"1000000"
} }

22
revid/config/config.go
View File

@ -234,17 +234,17 @@ type Config struct {
// localhost:6970. MPEGT-TS packetization is used. // localhost:6970. MPEGT-TS packetization is used.
Outputs []uint8 Outputs []uint8
PSITime uint // Sets the time between a packet being sent. PSITime uint // Sets the time between a packet being sent.
Quantization uint // Quantization defines the quantization level, which will determine variable bitrate quality in the case of input from the Pi Camera. Quantization uint // Quantization defines the quantization level, which will determine variable bitrate quality in the case of input from the Pi Camera.
RBCapacity uint // The number of bytes the pool buffer will occupy. PoolCapacity uint // The number of bytes the pool buffer will occupy.
RBStartElementSize uint // The starting element size of the pool buffer from which element size will increase to accomodate frames. PoolStartElementSize uint // The starting element size of the pool buffer from which element size will increase to accomodate frames.
RBWriteTimeout uint // The pool buffer write timeout in seconds. PoolWriteTimeout uint // The pool buffer write timeout in seconds.
RecPeriod float64 // How many seconds to record at a time. RecPeriod float64 // How many seconds to record at a time.
Rotation uint // Rotation defines the video rotation angle in degrees Raspivid input. Rotation uint // Rotation defines the video rotation angle in degrees Raspivid input.
RTMPURL string // RTMPURL specifies the Rtmp output destination URL. This must be defined if RTMP is to be used as an output. RTMPURL string // RTMPURL specifies the Rtmp output destination URL. This must be defined if RTMP is to be used as an output.
RTPAddress string // RTPAddress defines the RTP output destination. RTPAddress string // RTPAddress defines the RTP output destination.
SampleRate uint // Samples a second (Hz). SampleRate uint // Samples a second (Hz).
Saturation int Saturation int
// Sharpness is the sharpness of capture image/video from a capture device. // Sharpness is the sharpness of capture image/video from a capture device.
Sharpness int Sharpness int

40
revid/config/config_test.go
View File

@ -47,22 +47,22 @@ func TestValidate(t *testing.T) {
dl := &dumbLogger{} dl := &dumbLogger{}
want := Config{ want := Config{
Logger: dl, Logger: dl,
Input: defaultInput, Input: defaultInput,
Outputs: []uint8{defaultOutput}, Outputs: []uint8{defaultOutput},
InputCodec: defaultInputCodec, InputCodec: defaultInputCodec,
RTPAddress: defaultRTPAddr, RTPAddress: defaultRTPAddr,
CameraIP: defaultCameraIP, CameraIP: defaultCameraIP,
BurstPeriod: defaultBurstPeriod, BurstPeriod: defaultBurstPeriod,
MinFrames: defaultMinFrames, MinFrames: defaultMinFrames,
FrameRate: defaultFrameRate, FrameRate: defaultFrameRate,
ClipDuration: defaultClipDuration, ClipDuration: defaultClipDuration,
PSITime: defaultPSITime, PSITime: defaultPSITime,
FileFPS: defaultFileFPS, FileFPS: defaultFileFPS,
RBCapacity: defaultRBCapacity, PoolCapacity: defaultPoolCapacity,
RBStartElementSize: defaultRBStartElementSize, PoolStartElementSize: defaultPoolStartElementSize,
RBWriteTimeout: defaultRBWriteTimeout, PoolWriteTimeout: defaultPoolWriteTimeout,
MinFPS: defaultMinFPS, MinFPS: defaultMinFPS,
} }
got := Config{Logger: dl} got := Config{Logger: dl}
@ -112,8 +112,8 @@ func TestUpdate(t *testing.T) {
"OutputPath": "/outputpath", "OutputPath": "/outputpath",
"Outputs": "Rtmp,Rtp", "Outputs": "Rtmp,Rtp",
"Quantization": "30", "Quantization": "30",
"RBCapacity": "100000", "PoolCapacity": "100000",
"RBWriteTimeout": "50", "PoolWriteTimeout": "50",
"Rotation": "180", "Rotation": "180",
"RTMPURL": "rtmp://url", "RTMPURL": "rtmp://url",
"RTPAddress": "ip:port", "RTPAddress": "ip:port",
@ -162,8 +162,8 @@ func TestUpdate(t *testing.T) {
OutputPath: "/outputpath", OutputPath: "/outputpath",
Outputs: []uint8{OutputRTMP, OutputRTP}, Outputs: []uint8{OutputRTMP, OutputRTP},
Quantization: 30, Quantization: 30,
RBCapacity: 100000, PoolCapacity: 100000,
RBWriteTimeout: 50, PoolWriteTimeout: 50,
Rotation: 180, Rotation: 180,
RTMPURL: "rtmp://url", RTMPURL: "rtmp://url",
RTPAddress: "ip:port", RTPAddress: "ip:port",

6
revid/config/parameter/generate_parameters.go
View File

@ -116,9 +116,9 @@ var params = []Param{
{N: "Outputs", BT: "[]Output", M: "Output", E: []string{"HTTP", "RTMP", "RTP", "File"}}, {N: "Outputs", BT: "[]Output", M: "Output", E: []string{"HTTP", "RTMP", "RTP", "File"}},
{N: "PSITime", BT: "time.Duration"}, {N: "PSITime", BT: "time.Duration"},
{N: "Quantization", BT: "uint"}, {N: "Quantization", BT: "uint"},
{N: "RBCapacity", BT: "uint", Min: 1000000, Max: 100000000}, {N: "PoolCapacity", BT: "uint", Min: 1000000, Max: 100000000},
{N: "RBMaxElements", BT: "uint", Min: 0, Max: math.MaxUint32}, {N: "PoolMaxElements", BT: "uint", Min: 0, Max: math.MaxUint32},
{N: "RBWriteTimeout", BT: "time.Duration"}, {N: "PoolWriteTimeout", BT: "time.Duration"},
{N: "RTMPURL", BT: "string"}, {N: "RTMPURL", BT: "string"},
{N: "RTPAddress", BT: "string"}, {N: "RTPAddress", BT: "string"},
{N: "RecPeriod", BT: "float64"}, // TODO(Trek): bounds. {N: "RecPeriod", BT: "float64"}, // TODO(Trek): bounds.

24
revid/config/parameter/parameters.go
View File

@ -701,10 +701,10 @@ func (q *Quantization) Set(val string) error {
return nil return nil
} }
type RBCapacity uint type PoolCapacity uint
func (r *RBCapacity) Type() string { return "uint" } func (r *PoolCapacity) Type() string { return "uint" }
func (r *RBCapacity) Set(val string) error { func (r *PoolCapacity) Set(val string) error {
_v, err := strconv.Atoi(val) _v, err := strconv.Atoi(val)
if err != nil { if err != nil {
return fmt.Errorf("could not convert set string to int: %w", err) return fmt.Errorf("could not convert set string to int: %w", err)
@ -714,14 +714,14 @@ func (r *RBCapacity) Set(val string) error {
return fmt.Errorf("invalid value %v", _v) return fmt.Errorf("invalid value %v", _v)
} }
*r = RBCapacity(_v) *r = PoolCapacity(_v)
return nil return nil
} }
type RBMaxElements uint type PoolMaxElements uint
func (r *RBMaxElements) Type() string { return "uint" } func (r *PoolMaxElements) Type() string { return "uint" }
func (r *RBMaxElements) Set(val string) error { func (r *PoolMaxElements) Set(val string) error {
_v, err := strconv.Atoi(val) _v, err := strconv.Atoi(val)
if err != nil { if err != nil {
return fmt.Errorf("could not convert set string to int: %w", err) return fmt.Errorf("could not convert set string to int: %w", err)
@ -731,19 +731,19 @@ func (r *RBMaxElements) Set(val string) error {
return fmt.Errorf("invalid value %v", _v) return fmt.Errorf("invalid value %v", _v)
} }
*r = RBMaxElements(_v) *r = PoolMaxElements(_v)
return nil return nil
} }
type RBWriteTimeout time.Duration type PoolWriteTimeout time.Duration
func (r *RBWriteTimeout) Type() string { return "time.Duration" } func (r *PoolWriteTimeout) Type() string { return "time.Duration" }
func (r *RBWriteTimeout) Set(val string) error { func (r *PoolWriteTimeout) Set(val string) error {
_v, err := strconv.Atoi(val) _v, err := strconv.Atoi(val)
if err != nil { if err != nil {
return fmt.Errorf("could not convert set string to int: %w", err) return fmt.Errorf("could not convert set string to int: %w", err)
} }
*r = RBWriteTimeout(time.Duration(_v) * time.Second) *r = PoolWriteTimeout(time.Duration(_v) * time.Second)
return nil return nil
} }

152
revid/config/variables.go
View File

@ -39,66 +39,66 @@ import (
// Config map Keys. // Config map Keys.
const ( const (
KeyAutoWhiteBalance = "AutoWhiteBalance" KeyAutoWhiteBalance = "AutoWhiteBalance"
KeyAWBGains = "AWBGains" KeyAWBGains = "AWBGains"
KeyBitDepth = "BitDepth" KeyBitDepth = "BitDepth"
KeyBitrate = "Bitrate" KeyBitrate = "Bitrate"
KeyBrightness = "Brightness" KeyBrightness = "Brightness"
KeyBurstPeriod = "BurstPeriod" KeyBurstPeriod = "BurstPeriod"
KeyCameraChan = "CameraChan" KeyCameraChan = "CameraChan"
KeyCameraIP = "CameraIP" KeyCameraIP = "CameraIP"
KeyCBR = "CBR" KeyCBR = "CBR"
KeyClipDuration = "ClipDuration" KeyClipDuration = "ClipDuration"
KeyChannels = "Channels" KeyChannels = "Channels"
KeyContrast = "Contrast" KeyContrast = "Contrast"
KeyExposure = "Exposure" KeyExposure = "Exposure"
KeyEV = "EV" KeyEV = "EV"
KeyFileFPS = "FileFPS" KeyFileFPS = "FileFPS"
KeyFilters = "Filters" KeyFilters = "Filters"
KeyFrameRate = "FrameRate" KeyFrameRate = "FrameRate"
KeyHeight = "Height" KeyHeight = "Height"
KeyHorizontalFlip = "HorizontalFlip" KeyHorizontalFlip = "HorizontalFlip"
KeyHTTPAddress = "HTTPAddress" KeyHTTPAddress = "HTTPAddress"
KeyInput = "Input" KeyInput = "Input"
KeyInputCodec = "InputCodec" KeyInputCodec = "InputCodec"
KeyInputPath = "InputPath" KeyInputPath = "InputPath"
KeyISO = "ISO" KeyISO = "ISO"
KeyLogging = "logging" KeyLogging = "logging"
KeyLoop = "Loop" KeyLoop = "Loop"
KeyMinFPS = "MinFPS" KeyMinFPS = "MinFPS"
KeyMinFrames = "MinFrames" KeyMinFrames = "MinFrames"
KeyMode = "mode" KeyMode = "mode"
KeyMotionDownscaling = "MotionDownscaling" KeyMotionDownscaling = "MotionDownscaling"
KeyMotionHistory = "MotionHistory" KeyMotionHistory = "MotionHistory"
KeyMotionInterval = "MotionInterval" KeyMotionInterval = "MotionInterval"
KeyMotionKernel = "MotionKernel" KeyMotionKernel = "MotionKernel"
KeyMotionMinArea = "MotionMinArea" KeyMotionMinArea = "MotionMinArea"
KeyMotionPadding = "MotionPadding" KeyMotionPadding = "MotionPadding"
KeyMotionPixels = "MotionPixels" KeyMotionPixels = "MotionPixels"
KeyMotionThreshold = "MotionThreshold" KeyMotionThreshold = "MotionThreshold"
KeyOutput = "Output" KeyOutput = "Output"
KeyOutputPath = "OutputPath" KeyOutputPath = "OutputPath"
KeyOutputs = "Outputs" KeyOutputs = "Outputs"
KeyPSITime = "PSITime" KeyPSITime = "PSITime"
KeyQuantization = "Quantization" KeyQuantization = "Quantization"
KeyRBCapacity = "RBCapacity" KeyPoolCapacity = "PoolCapacity"
KeyRBStartElementSize = "RBStartElementSize" KeyPoolStartElementSize = "PoolStartElementSize"
KeyRBWriteTimeout = "RBWriteTimeout" KeyPoolWriteTimeout = "PoolWriteTimeout"
KeyRecPeriod = "RecPeriod" KeyRecPeriod = "RecPeriod"
KeyRotation = "Rotation" KeyRotation = "Rotation"
KeyRTMPURL = "RTMPURL" KeyRTMPURL = "RTMPURL"
KeyRTPAddress = "RTPAddress" KeyRTPAddress = "RTPAddress"
KeySampleRate = "SampleRate" KeySampleRate = "SampleRate"
KeySaturation = "Saturation" KeySaturation = "Saturation"
KeySharpness = "Sharpness" KeySharpness = "Sharpness"
KeyJPEGQuality = "JPEGQuality" KeyJPEGQuality = "JPEGQuality"
KeySuppress = "Suppress" KeySuppress = "Suppress"
KeyTimelapseDuration = "TimelapseDuration" KeyTimelapseDuration = "TimelapseDuration"
KeyTimelapseInterval = "TimelapseInterval" KeyTimelapseInterval = "TimelapseInterval"
KeyVBRBitrate = "VBRBitrate" KeyVBRBitrate = "VBRBitrate"
KeyVBRQuality = "VBRQuality" KeyVBRQuality = "VBRQuality"
KeyVerticalFlip = "VerticalFlip" KeyVerticalFlip = "VerticalFlip"
KeyWidth = "Width" KeyWidth = "Width"
) )
// Config map parameter types. // Config map parameter types.
@ -127,9 +127,9 @@ const (
defaultFileFPS = 0 defaultFileFPS = 0
// Ring buffer defaults. // Ring buffer defaults.
defaultRBCapacity = 50000000 // => 50MB defaultPoolCapacity = 50000000 // => 50MB
defaultRBStartElementSize = 1000 // bytes defaultPoolStartElementSize = 1000 // bytes
defaultRBWriteTimeout = 5 // Seconds. defaultPoolWriteTimeout = 5 // Seconds.
// Motion filter parameter defaults. // Motion filter parameter defaults.
defaultMinFPS = 1.0 defaultMinFPS = 1.0
@ -519,25 +519,27 @@ var Variables = []struct {
Update: func(c *Config, v string) { c.Quantization = parseUint(KeyQuantization, v, c) }, Update: func(c *Config, v string) { c.Quantization = parseUint(KeyQuantization, v, c) },
}, },
{ {
Name: KeyRBCapacity, Name: KeyPoolCapacity,
Type_: typeUint,
Update: func(c *Config, v string) { c.RBCapacity = parseUint(KeyRBCapacity, v, c) },
Validate: func(c *Config) { c.RBCapacity = lessThanOrEqual(KeyRBCapacity, c.RBCapacity, 0, c, defaultRBCapacity) },
},
{
Name: KeyRBStartElementSize,
Type_: typeUint, Type_: typeUint,
Update: func(c *Config, v string) { c.RBStartElementSize = parseUint("RBStartElementSize", v, c) }, Update: func(c *Config, v string) { c.PoolCapacity = parseUint(KeyPoolCapacity, v, c) },
Validate: func(c *Config) { Validate: func(c *Config) {
c.RBStartElementSize = lessThanOrEqual("RBStartElementSize", c.RBStartElementSize, 0, c, defaultRBStartElementSize) c.PoolCapacity = lessThanOrEqual(KeyPoolCapacity, c.PoolCapacity, 0, c, defaultPoolCapacity)
}, },
}, },
{ {
Name: KeyRBWriteTimeout, Name: KeyPoolStartElementSize,
Type_: typeUint, Type_: typeUint,
Update: func(c *Config, v string) { c.RBWriteTimeout = parseUint(KeyRBWriteTimeout, v, c) }, Update: func(c *Config, v string) { c.PoolStartElementSize = parseUint("PoolStartElementSize", v, c) },
Validate: func(c *Config) { Validate: func(c *Config) {
c.RBWriteTimeout = lessThanOrEqual(KeyRBWriteTimeout, c.RBWriteTimeout, 0, c, defaultRBWriteTimeout) c.PoolStartElementSize = lessThanOrEqual("PoolStartElementSize", c.PoolStartElementSize, 0, c, defaultPoolStartElementSize)
},
},
{
Name: KeyPoolWriteTimeout,
Type_: typeUint,
Update: func(c *Config, v string) { c.PoolWriteTimeout = parseUint(KeyPoolWriteTimeout, v, c) },
Validate: func(c *Config) {
c.PoolWriteTimeout = lessThanOrEqual(KeyPoolWriteTimeout, c.PoolWriteTimeout, 0, c, defaultPoolWriteTimeout)
}, },
}, },
{ {

10
revid/pipeline.go
View File

@ -172,8 +172,8 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
// Calculate no. of pool buffer elements based on starting element size // Calculate no. of pool buffer elements based on starting element size
// const and config directed max pool buffer size, then create buffer. // const and config directed max pool buffer size, then create buffer.
// This is only used if the selected output uses a pool buffer. // This is only used if the selected output uses a pool buffer.
nElements := r.cfg.RBCapacity / r.cfg.RBStartElementSize nElements := r.cfg.PoolCapacity / r.cfg.PoolStartElementSize
writeTimeout := time.Duration(r.cfg.RBWriteTimeout) * time.Second writeTimeout := time.Duration(r.cfg.PoolWriteTimeout) * time.Second
// We will go through our outputs and create the corresponding senders to add // We will go through our outputs and create the corresponding senders to add
// to mtsSenders if the output requires MPEGTS encoding, or flvSenders if the // to mtsSenders if the output requires MPEGTS encoding, or flvSenders if the
@ -183,7 +183,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
switch out { switch out {
case config.OutputHTTP: case config.OutputHTTP:
r.cfg.Logger.Log(logger.Debug, "using HTTP output") r.cfg.Logger.Log(logger.Debug, "using HTTP output")
pb := pool.NewBuffer(int(r.cfg.RBStartElementSize), int(nElements), writeTimeout) pb := pool.NewBuffer(int(r.cfg.PoolStartElementSize), int(nElements), writeTimeout)
hs := newHTTPSender(r.ns, r.cfg.Logger.Log, r.bitrate.Report) hs := newHTTPSender(r.ns, r.cfg.Logger.Log, r.bitrate.Report)
w = newMTSSender(hs, r.cfg.Logger.Log, pb, r.cfg.ClipDuration) w = newMTSSender(hs, r.cfg.Logger.Log, pb, r.cfg.ClipDuration)
mtsSenders = append(mtsSenders, w) mtsSenders = append(mtsSenders, w)
@ -204,7 +204,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
mtsSenders = append(mtsSenders, w) mtsSenders = append(mtsSenders, w)
case config.OutputFiles: case config.OutputFiles:
r.cfg.Logger.Log(logger.Debug, "using Files output") r.cfg.Logger.Log(logger.Debug, "using Files output")
pb := pool.NewBuffer(int(r.cfg.RBStartElementSize), int(nElements), writeTimeout) pb := pool.NewBuffer(int(r.cfg.PoolStartElementSize), int(nElements), writeTimeout)
fs, err := newFileSender(r.cfg.Logger, r.cfg.OutputPath, true) fs, err := newFileSender(r.cfg.Logger, r.cfg.OutputPath, true)
if err != nil { if err != nil {
return err return err
@ -213,7 +213,7 @@ func (r *Revid) setupPipeline(mtsEnc func(dst io.WriteCloser, rate float64) (io.
mtsSenders = append(mtsSenders, w) mtsSenders = append(mtsSenders, w)
case config.OutputRTMP: case config.OutputRTMP:
r.cfg.Logger.Log(logger.Debug, "using RTMP output") r.cfg.Logger.Log(logger.Debug, "using RTMP output")
pb := pool.NewBuffer(int(r.cfg.RBStartElementSize), int(nElements), writeTimeout) pb := pool.NewBuffer(int(r.cfg.PoolStartElementSize), int(nElements), writeTimeout)
w, err := newRtmpSender(r.cfg.RTMPURL, rtmpConnectionMaxTries, pb, r.cfg.Logger.Log, r.bitrate.Report) w, err := newRtmpSender(r.cfg.RTMPURL, rtmpConnectionMaxTries, pb, r.cfg.Logger.Log, r.bitrate.Report)
if err != nil { if err != nil {
r.cfg.Logger.Log(logger.Warning, "rtmp connect error", "error", err.Error()) r.cfg.Logger.Log(logger.Warning, "rtmp connect error", "error", err.Error())

4
revid/revid.go
View File

@ -45,8 +45,8 @@ import (
// Misc consts. // Misc consts.
const ( const (
rbStartingElementSize = 10000 // Bytes. poolStartingElementSize = 10000 // Bytes.
rtmpConnectionMaxTries = 5 rtmpConnectionMaxTries = 5
) )
type Logger interface { type Logger interface {

36
revid/revid_test.go
View File

@ -49,16 +49,16 @@ func TestRaspistill(t *testing.T) {
// Configuration parameters. // Configuration parameters.
const ( const (
timelapseInterval = "4" timelapseInterval = "4"
timelapseDuration = "25" timelapseDuration = "25"
rbStartElementSize = "1000000" poolStartElementSize = "1000000"
input = "Raspistill" input = "Raspistill"
codec = "JPEG" codec = "JPEG"
output = "Files" output = "Files"
outDir = "out" outDir = "out"
outputPath = outDir + "/" outputPath = outDir + "/"
logging = "Debug" logging = "Debug"
testImgDir = "../../../test/test-data/av/input/jpeg/" testImgDir = "../../../test/test-data/av/input/jpeg/"
) )
const runTime = 40 * time.Second const runTime = 40 * time.Second
@ -84,14 +84,14 @@ func TestRaspistill(t *testing.T) {
err = rv.Update( err = rv.Update(
map[string]string{ map[string]string{
config.KeyInput: input, config.KeyInput: input,
config.KeyInputCodec: codec, config.KeyInputCodec: codec,
config.KeyOutput: output, config.KeyOutput: output,
config.KeyOutputPath: outputPath, config.KeyOutputPath: outputPath,
config.KeyTimelapseInterval: timelapseInterval, config.KeyTimelapseInterval: timelapseInterval,
config.KeyTimelapseDuration: timelapseDuration, config.KeyTimelapseDuration: timelapseDuration,
config.KeyLogging: logging, config.KeyLogging: logging,
config.KeyRBStartElementSize: rbStartElementSize, config.KeyPoolStartElementSize: poolStartElementSize,
}, },
) )
if err != nil { if err != nil {

32
revid/senders.go
View File

@ -53,14 +53,14 @@ type Log func(level int8, message string, params ...interface{})
// Sender pool buffer read timeouts. // Sender pool buffer read timeouts.
const ( const (
rtmpRBReadTimeout = 1 * time.Second rtmpPoolReadTimeout = 1 * time.Second
mtsRBReadTimeout = 1 * time.Second mtsPoolReadTimeout = 1 * time.Second
maxBuffLen = 50000000 maxBuffLen = 50000000
) )
var ( var (
adjustedRTMPRBElementSize int adjustedRTMPPoolElementSize int
adjustedMTSRBElementSize int adjustedMTSPoolElementSize int
) )
// httpSender provides an implemntation of io.Writer to perform sends to a http // httpSender provides an implemntation of io.Writer to perform sends to a http
@ -246,7 +246,7 @@ func (s *mtsSender) output() {
// If chunk is nil then we're ready to get another from the ringBuffer. // If chunk is nil then we're ready to get another from the ringBuffer.
if chunk == nil { if chunk == nil {
var err error var err error
chunk, err = s.pool.Next(mtsRBReadTimeout) chunk, err = s.pool.Next(mtsPoolReadTimeout)
switch err { switch err {
case nil, io.EOF: case nil, io.EOF:
continue continue
@ -304,12 +304,12 @@ func (s *mtsSender) Write(d []byte) (int, error) {
s.pool.Flush() s.pool.Flush()
} }
if err != nil { if err != nil {
s.log(logger.Warning, "ringBuffer write error", "error", err.Error(), "n", n, "size", len(s.buf), "rb element size", adjustedMTSRBElementSize) s.log(logger.Warning, "ringBuffer write error", "error", err.Error(), "n", n, "size", len(s.buf), "rb element size", adjustedMTSPoolElementSize)
if err == pool.ErrTooLong { if err == pool.ErrTooLong {
adjustedMTSRBElementSize = len(s.buf) * 2 adjustedMTSPoolElementSize = len(s.buf) * 2
numElements := maxBuffLen / adjustedMTSRBElementSize numElements := maxBuffLen / adjustedMTSPoolElementSize
s.pool = pool.NewBuffer(maxBuffLen/adjustedMTSRBElementSize, adjustedMTSRBElementSize, 5*time.Second) s.pool = pool.NewBuffer(maxBuffLen/adjustedMTSPoolElementSize, adjustedMTSPoolElementSize, 5*time.Second)
s.log(logger.Info, "adjusted MTS pool buffer element size", "new size", adjustedMTSRBElementSize, "num elements", numElements, "size(MB)", numElements*adjustedMTSRBElementSize) s.log(logger.Info, "adjusted MTS pool buffer element size", "new size", adjustedMTSPoolElementSize, "num elements", numElements, "size(MB)", numElements*adjustedMTSPoolElementSize)
} }
} }
s.buf = s.buf[:0] s.buf = s.buf[:0]
@ -378,7 +378,7 @@ func (s *rtmpSender) output() {
// If chunk is nil then we're ready to get another from the pool buffer. // If chunk is nil then we're ready to get another from the pool buffer.
if chunk == nil { if chunk == nil {
var err error var err error
chunk, err = s.pool.Next(rtmpRBReadTimeout) chunk, err = s.pool.Next(rtmpPoolReadTimeout)
switch err { switch err {
case nil, io.EOF: case nil, io.EOF:
continue continue
@ -426,10 +426,10 @@ func (s *rtmpSender) Write(d []byte) (int, error) {
} else { } else {
s.log(logger.Warning, "pool buffer write error", "error", err.Error()) s.log(logger.Warning, "pool buffer write error", "error", err.Error())
if err == pool.ErrTooLong { if err == pool.ErrTooLong {
adjustedRTMPRBElementSize = len(d) * 2 adjustedRTMPPoolElementSize = len(d) * 2
numElements := maxBuffLen / adjustedRTMPRBElementSize numElements := maxBuffLen / adjustedRTMPPoolElementSize
s.pool = pool.NewBuffer(numElements, adjustedRTMPRBElementSize, 5*time.Second) s.pool = pool.NewBuffer(numElements, adjustedRTMPPoolElementSize, 5*time.Second)
s.log(logger.Info, "adjusted RTMP pool buffer element size", "new size", adjustedRTMPRBElementSize, "num elements", numElements, "size(MB)", numElements*adjustedRTMPRBElementSize) s.log(logger.Info, "adjusted RTMP pool buffer element size", "new size", adjustedRTMPPoolElementSize, "num elements", numElements, "size(MB)", numElements*adjustedRTMPPoolElementSize)
} }
} }
s.report(len(d)) s.report(len(d))

14
revid/senders_test.go
View File

@ -105,9 +105,9 @@ func TestMTSSenderSegment(t *testing.T) {
// Create ringBuffer, sender, sender and the MPEGTS encoder. // Create ringBuffer, sender, sender and the MPEGTS encoder.
const numberOfClips = 11 const numberOfClips = 11
dst := &destination{t: t, done: make(chan struct{}), doneAt: numberOfClips} dst := &destination{t: t, done: make(chan struct{}), doneAt: numberOfClips}
const testRBCapacity = 50000000 const testPoolCapacity = 50000000
nElements := testRBCapacity / rbStartingElementSize nElements := testPoolCapacity / poolStartingElementSize
sender := newMTSSender(dst, (*testLogger)(t).Log, pool.NewBuffer(nElements, rbStartingElementSize, 0), 0) sender := newMTSSender(dst, (*testLogger)(t).Log, pool.NewBuffer(nElements, poolStartingElementSize, 0), 0)
const psiSendCount = 10 const psiSendCount = 10
encoder, err := mts.NewEncoder(sender, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount), mts.Rate(25), mts.MediaType(mts.EncodeH264)) encoder, err := mts.NewEncoder(sender, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount), mts.Rate(25), mts.MediaType(mts.EncodeH264))
@ -186,9 +186,9 @@ func TestMtsSenderFailedSend(t *testing.T) {
// Create destination, the mtsSender and the mtsEncoder // Create destination, the mtsSender and the mtsEncoder
const clipToFailAt = 3 const clipToFailAt = 3
dst := &destination{t: t, testFails: true, failAt: clipToFailAt, done: make(chan struct{})} dst := &destination{t: t, testFails: true, failAt: clipToFailAt, done: make(chan struct{})}
const testRBCapacity = 50000000 // 50MB const testPoolCapacity = 50000000 // 50MB
nElements := testRBCapacity / rbStartingElementSize nElements := testPoolCapacity / poolStartingElementSize
sender := newMTSSender(dst, (*testLogger)(t).Log, pool.NewBuffer(nElements, rbStartingElementSize, 0), 0) sender := newMTSSender(dst, (*testLogger)(t).Log, pool.NewBuffer(nElements, poolStartingElementSize, 0), 0)
const psiSendCount = 10 const psiSendCount = 10
encoder, err := mts.NewEncoder(sender, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount), mts.Rate(25), mts.MediaType(mts.EncodeH264)) encoder, err := mts.NewEncoder(sender, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount), mts.Rate(25), mts.MediaType(mts.EncodeH264))
@ -269,7 +269,7 @@ func TestMtsSenderDiscontinuity(t *testing.T) {
// Create destination, the mtsSender and the mtsEncoder. // Create destination, the mtsSender and the mtsEncoder.
const clipToDelay = 3 const clipToDelay = 3
dst := &destination{t: t, sendDelay: 10 * time.Millisecond, delayAt: clipToDelay, done: make(chan struct{})} dst := &destination{t: t, sendDelay: 10 * time.Millisecond, delayAt: clipToDelay, done: make(chan struct{})}
sender := newMTSSender(dst, (*testLogger)(t).Log, pool.NewBuffer(1, rbStartingElementSize, 0), 0) sender := newMTSSender(dst, (*testLogger)(t).Log, pool.NewBuffer(1, poolStartingElementSize, 0), 0)
const psiSendCount = 10 const psiSendCount = 10
encoder, err := mts.NewEncoder(sender, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount), mts.Rate(25), mts.MediaType(mts.EncodeH264)) encoder, err := mts.NewEncoder(sender, (*testLogger)(t), mts.PacketBasedPSI(psiSendCount), mts.Rate(25), mts.MediaType(mts.EncodeH264))