av/revid/senders.go

502 lines
13 KiB
Go
Raw Permalink Normal View History

2018-06-09 07:38:48 +03:00
/*
NAME
senders.go
DESCRIPTION
See Readme.md
AUTHORS
Saxon A. Nelson-Milton <saxon@ausocean.org>
Alan Noble <alan@ausocean.org>
LICENSE
revid is Copyright (C) 2017-2018 the Australian Ocean Lab (AusOcean)
It is free software: you can redistribute it and/or modify them
under the terms of the GNU General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
It is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
for more details.
You should have received a copy of the GNU General Public License
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
*/
package revid
import (
"errors"
"fmt"
"io"
"net"
2018-06-09 07:38:48 +03:00
"os"
"sync"
"time"
2018-06-09 07:38:48 +03:00
"github.com/Comcast/gots/packet"
2019-03-25 04:21:03 +03:00
"bitbucket.org/ausocean/av/container/mts"
"bitbucket.org/ausocean/av/protocol/rtmp"
"bitbucket.org/ausocean/av/protocol/rtp"
"bitbucket.org/ausocean/av/revid/config"
"bitbucket.org/ausocean/iot/pi/netsender"
"bitbucket.org/ausocean/utils/logger"
"bitbucket.org/ausocean/utils/ring"
2018-06-09 07:38:48 +03:00
)
2019-03-24 12:34:35 +03:00
// Log is used by the multiSender.
type Log func(level int8, message string, params ...interface{})
// Sender ring buffer read timeouts.
const (
rtmpRBReadTimeout = 1 * time.Second
mtsRBReadTimeout = 1 * time.Second
2020-04-10 10:05:45 +03:00
maxBuffLen = 50000000
)
var (
adjustedRTMPRBElementSize int
adjustedMTSRBElementSize int
)
2019-04-02 06:15:36 +03:00
// httpSender provides an implemntation of io.Writer to perform sends to a http
// destination.
type httpSender struct {
2020-01-22 07:56:14 +03:00
client *netsender.Sender
log func(lvl int8, msg string, args ...interface{})
report func(sent int)
}
// newHttpSender returns a pointer to a new httpSender.
func newHTTPSender(ns *netsender.Sender, log func(lvl int8, msg string, args ...interface{}), report func(sent int)) *httpSender {
return &httpSender{
2020-01-22 07:56:14 +03:00
client: ns,
log: log,
report: report,
}
}
2019-04-02 06:15:36 +03:00
// Write implements io.Writer.
func (s *httpSender) Write(d []byte) (int, error) {
s.log(logger.Debug, "HTTP sending")
err := httpSend(d, s.client, s.log)
if err == nil {
s.log(logger.Debug, "good send", "len", len(d))
2020-01-22 07:56:14 +03:00
s.report(len(d))
} else {
s.log(logger.Debug, "bad send", "error", err)
}
return len(d), err
}
func (s *httpSender) Close() error { return nil }
func httpSend(d []byte, client *netsender.Sender, log func(lvl int8, msg string, args ...interface{})) error {
// Only send if "V0" or "S0" are configured as input.
send := false
ip := client.Param("ip")
log(logger.Debug, "making pins, and sending recv request", "ip", ip)
pins := netsender.MakePins(ip, "V,S")
for i, pin := range pins {
2020-08-15 11:08:45 +03:00
switch pin.Name {
case "V0":
pins[i].MimeType = "video/mp2t"
case "S0":
pins[i].MimeType = "audio/x-wav"
default:
continue
}
2020-08-15 11:08:45 +03:00
pins[i].Value = len(d)
pins[i].Data = d
send = true
break
}
if !send {
return nil
}
var err error
var reply string
reply, _, err = client.Send(netsender.RequestRecv, pins)
if err != nil {
return err
}
log(logger.Debug, "good request", "reply", reply)
return extractMeta(reply, log)
}
// extractMeta looks at a reply at extracts any time or location data - then used
// to update time and location information in the mpegts encoder.
func extractMeta(r string, log func(lvl int8, msg string, args ...interface{})) error {
dec, err := netsender.NewJSONDecoder(r)
if err != nil {
return nil
}
// Extract time from reply
t, err := dec.Int("ts")
if err != nil {
log(logger.Warning, "No timestamp in reply")
} else {
log(logger.Debug, fmt.Sprintf("got timestamp: %v", t))
mts.RealTime.Set(time.Unix(int64(t), 0))
}
// Extract location from reply
g, err := dec.String("ll")
if err != nil {
log(logger.Debug, "No location in reply")
} else {
log(logger.Debug, fmt.Sprintf("got location: %v", g))
mts.Meta.Add("loc", g)
}
return nil
}
2018-06-09 07:38:48 +03:00
// fileSender implements loadSender for a local file destination.
type fileSender struct {
file *os.File
data []byte
multiFile bool
path string
init bool
log config.Logger
2018-06-09 07:38:48 +03:00
}
// newFileSender returns a new fileSender. Setting multi true will write a new
// file for each write to this sender.
func newFileSender(l config.Logger, path string, multiFile bool) (*fileSender, error) {
return &fileSender{
path: path,
log: l,
multiFile: multiFile,
init: true,
}, nil
2018-06-09 07:38:48 +03:00
}
// Write implements io.Writer.
func (s *fileSender) Write(d []byte) (int, error) {
if s.init || s.multiFile {
fileName := s.path + time.Now().String()
s.log.Debug("creating new output file", "init", s.init, "multiFile", s.multiFile, "fileName", fileName)
f, err := os.Create(fileName)
if err != nil {
return 0, fmt.Errorf("could not create file to write media to: %w", err)
}
s.file = f
s.init = false
}
s.log.Debug("writing output file", "len(d)", len(d))
return s.file.Write(d)
2018-06-09 07:38:48 +03:00
}
func (s *fileSender) Close() error { return s.file.Close() }
2018-06-09 07:38:48 +03:00
// mtsSender implements io.WriteCloser and provides sending capability specifically
// for use with MPEGTS packetization. It handles the construction of appropriately
// lengthed clips based on clip duration and PSI. It also accounts for
// discontinuities by setting the discontinuity indicator for the first packet of a clip.
type mtsSender struct {
dst io.WriteCloser
buf []byte
ring *ring.Buffer
next []byte
pkt packet.Packet
repairer *mts.DiscontinuityRepairer
curPid int
clipDur time.Duration
prev time.Time
2019-04-18 10:25:48 +03:00
done chan struct{}
log func(lvl int8, msg string, args ...interface{})
wg sync.WaitGroup
}
// newMtsSender returns a new mtsSender.
func newMTSSender(dst io.WriteCloser, log func(lvl int8, msg string, args ...interface{}), rb *ring.Buffer, clipDur time.Duration) *mtsSender {
s := &mtsSender{
dst: dst,
repairer: mts.NewDiscontinuityRepairer(),
log: log,
ring: rb,
2019-04-18 10:25:48 +03:00
done: make(chan struct{}),
clipDur: clipDur,
}
s.wg.Add(1)
go s.output()
return s
}
// output starts an mtsSender's data handling routine.
func (s *mtsSender) output() {
var chunk *ring.Chunk
for {
select {
2019-04-18 10:25:48 +03:00
case <-s.done:
s.log(logger.Info, "terminating sender output routine")
defer s.wg.Done()
return
default:
// If chunk is nil then we're ready to get another from the ringBuffer.
if chunk == nil {
var err error
chunk, err = s.ring.Next(mtsRBReadTimeout)
switch err {
case nil, io.EOF:
continue
case ring.ErrTimeout:
s.log(logger.Debug, "mtsSender: ring buffer read timeout")
continue
default:
s.log(logger.Error, "unexpected error", "error", err.Error())
continue
}
}
err := s.repairer.Repair(chunk.Bytes())
if err != nil {
chunk.Close()
chunk = nil
continue
}
s.log(logger.Debug, "mtsSender: writing")
_, err = s.dst.Write(chunk.Bytes())
if err != nil {
s.log(logger.Debug, "failed write, repairing MTS", "error", err)
s.repairer.Failed()
continue
} else {
s.log(logger.Debug, "good write")
}
chunk.Close()
chunk = nil
}
}
}
// Write implements io.Writer.
func (s *mtsSender) Write(d []byte) (int, error) {
if len(d) < mts.PacketSize {
return 0, errors.New("do not have full MTS packet")
}
if s.next != nil {
s.buf = append(s.buf, s.next...)
}
bytes := make([]byte, len(d))
copy(bytes, d)
s.next = bytes
p, _ := mts.PID(bytes)
s.curPid = int(p)
if time.Now().Sub(s.prev) >= s.clipDur && s.curPid == mts.PatPid && len(s.buf) > 0 {
s.prev = time.Now()
n, err := s.ring.Write(s.buf)
if err == nil {
s.ring.Flush()
}
if err != nil {
s.log(logger.Warning, "ringBuffer write error", "error", err.Error(), "n", n, "size", len(s.buf), "rb element size", adjustedMTSRBElementSize)
if err == ring.ErrTooLong {
adjustedMTSRBElementSize = len(s.buf) * 2
numElements := maxBuffLen / adjustedMTSRBElementSize
s.ring = ring.NewBuffer(maxBuffLen/adjustedMTSRBElementSize, adjustedMTSRBElementSize, 5*time.Second)
s.log(logger.Info, "adjusted MTS ring buffer element size", "new size", adjustedMTSRBElementSize, "num elements", numElements, "size(MB)", numElements*adjustedMTSRBElementSize)
}
}
s.buf = s.buf[:0]
}
return len(d), nil
}
// Close implements io.Closer.
func (s *mtsSender) Close() error {
s.log(logger.Debug, "closing sender output routine")
2019-04-18 10:25:48 +03:00
close(s.done)
s.wg.Wait()
s.log(logger.Info, "sender output routine closed")
return nil
}
2018-06-09 07:38:48 +03:00
// rtmpSender implements loadSender for a native RTMP destination.
type rtmpSender struct {
2020-01-22 07:56:14 +03:00
conn *rtmp.Conn
url string
retries int
log func(lvl int8, msg string, args ...interface{})
ring *ring.Buffer
2020-01-22 07:56:14 +03:00
done chan struct{}
wg sync.WaitGroup
report func(sent int)
2018-06-09 07:38:48 +03:00
}
func newRtmpSender(url string, retries int, rb *ring.Buffer, log func(lvl int8, msg string, args ...interface{}), report func(sent int)) (*rtmpSender, error) {
var conn *rtmp.Conn
2018-06-09 07:38:48 +03:00
var err error
for n := 0; n < retries; n++ {
conn, err = rtmp.Dial(url, log)
2018-06-09 07:38:48 +03:00
if err == nil {
break
}
log(logger.Error, "dial error", "error", err)
2018-06-09 07:38:48 +03:00
if n < retries-1 {
log(logger.Info, "retrying dial")
2018-06-09 07:38:48 +03:00
}
}
s := &rtmpSender{
2020-01-22 07:56:14 +03:00
conn: conn,
url: url,
retries: retries,
log: log,
ring: rb,
done: make(chan struct{}),
report: report,
2018-06-09 07:38:48 +03:00
}
2019-04-15 04:18:12 +03:00
s.wg.Add(1)
go s.output()
return s, err
2018-06-09 07:38:48 +03:00
}
2019-04-15 04:18:12 +03:00
// output starts an mtsSender's data handling routine.
func (s *rtmpSender) output() {
var chunk *ring.Chunk
2019-04-15 04:18:12 +03:00
for {
select {
2019-04-18 10:25:48 +03:00
case <-s.done:
s.log(logger.Info, "terminating sender output routine")
2019-04-15 04:18:12 +03:00
defer s.wg.Done()
return
default:
// If chunk is nil then we're ready to get another from the ring buffer.
if chunk == nil {
2019-04-15 04:18:12 +03:00
var err error
chunk, err = s.ring.Next(rtmpRBReadTimeout)
2019-04-15 04:18:12 +03:00
switch err {
case nil, io.EOF:
continue
case ring.ErrTimeout:
s.log(logger.Debug, "rtmpSender: ring buffer read timeout")
2019-04-15 04:18:12 +03:00
continue
default:
s.log(logger.Error, "unexpected error", "error", err.Error())
2019-04-15 04:18:12 +03:00
continue
}
}
if s.conn == nil {
s.log(logger.Warning, "no rtmp connection, re-dialing")
2019-04-15 04:18:12 +03:00
err := s.restart()
if err != nil {
s.log(logger.Warning, "could not restart connection", "error", err)
2019-04-15 04:18:12 +03:00
continue
}
}
_, err := s.conn.Write(chunk.Bytes())
switch err {
case nil, rtmp.ErrInvalidFlvTag:
s.log(logger.Debug, "good write to conn")
default:
s.log(logger.Warning, "send error, re-dialing", "error", err)
2019-04-15 04:18:12 +03:00
err = s.restart()
if err != nil {
s.log(logger.Warning, "could not restart connection", "error", err)
2019-04-15 04:18:12 +03:00
}
continue
}
chunk.Close()
chunk = nil
2019-04-15 04:18:12 +03:00
}
}
}
// Write implements io.Writer.
func (s *rtmpSender) Write(d []byte) (int, error) {
s.log(logger.Debug, "writing to ring buffer")
2019-04-18 10:25:48 +03:00
_, err := s.ring.Write(d)
if err == nil {
s.ring.Flush()
s.log(logger.Debug, "good ring buffer write", "len", len(d))
} else {
s.log(logger.Warning, "ring buffer write error", "error", err.Error())
if err == ring.ErrTooLong {
adjustedRTMPRBElementSize = len(d) * 2
numElements := maxBuffLen / adjustedRTMPRBElementSize
s.ring = ring.NewBuffer(numElements, adjustedRTMPRBElementSize, 5*time.Second)
s.log(logger.Info, "adjusted RTMP ring buffer element size", "new size", adjustedRTMPRBElementSize, "num elements", numElements, "size(MB)", numElements*adjustedRTMPRBElementSize)
}
}
2020-01-22 07:56:14 +03:00
s.report(len(d))
2019-04-15 04:18:12 +03:00
return len(d), nil
2018-06-09 07:38:48 +03:00
}
func (s *rtmpSender) restart() error {
s.close()
2019-03-03 10:11:35 +03:00
var err error
2018-06-09 07:38:48 +03:00
for n := 0; n < s.retries; n++ {
s.log(logger.Debug, "dialing", "dials", n)
s.conn, err = rtmp.Dial(s.url, s.log)
2018-06-09 07:38:48 +03:00
if err == nil {
break
}
s.log(logger.Error, "dial error", "error", err)
2018-06-09 07:38:48 +03:00
if n < s.retries-1 {
s.log(logger.Info, "retry rtmp connection")
2018-06-09 07:38:48 +03:00
}
}
return err
}
func (s *rtmpSender) Close() error {
s.log(logger.Debug, "closing output routine")
2019-04-18 10:25:48 +03:00
if s.done != nil {
close(s.done)
}
s.wg.Wait()
s.log(logger.Info, "output routine closed")
return s.close()
}
func (s *rtmpSender) close() error {
s.log(logger.Debug, "closing connection")
if s.conn == nil {
return nil
}
return s.conn.Close()
2018-06-09 07:38:48 +03:00
}
// TODO: Write restart func for rtpSender
// rtpSender implements loadSender for a native udp destination with rtp packetization.
type rtpSender struct {
log func(lvl int8, msg string, args ...interface{})
encoder *rtp.Encoder
2019-03-09 07:58:07 +03:00
data []byte
2020-01-23 02:37:28 +03:00
report func(sent int)
}
2020-01-23 02:37:28 +03:00
func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint, report func(sent int)) (*rtpSender, error) {
conn, err := net.Dial("udp", addr)
if err != nil {
return nil, err
}
s := &rtpSender{
log: log,
encoder: rtp.NewEncoder(conn, int(fps)),
2020-01-23 02:37:28 +03:00
report: report,
}
return s, nil
}
// Write implements io.Writer.
func (s *rtpSender) Write(d []byte) (int, error) {
2019-04-09 09:14:18 +03:00
s.data = make([]byte, len(d))
copy(s.data, d)
_, err := s.encoder.Write(s.data)
if err != nil {
s.log(logger.Warning, "rtpSender: write error", err.Error())
2019-04-09 09:14:18 +03:00
}
2020-01-23 02:37:28 +03:00
s.report(len(d))
2019-04-09 09:14:18 +03:00
return len(d), nil
}
func (s *rtpSender) Close() error { return nil }