av/revid/senders.go

414 lines
10 KiB
Go
Raw 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 (
"fmt"
"io"
"net"
2018-06-09 07:38:48 +03:00
"os"
"strconv"
"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/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{})
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 {
client *netsender.Sender
log func(lvl int8, msg string, args ...interface{})
}
// newMinimalHttpSender returns a pointer to a new minimalHttpSender.
func newHttpSender(ns *netsender.Sender, log func(lvl int8, msg string, args ...interface{})) *httpSender {
return &httpSender{
client: ns,
log: log,
}
}
2019-04-02 06:15:36 +03:00
// Write implements io.Writer.
func (s *httpSender) Write(d []byte) (int, error) {
return len(d), httpSend(d, s.client, s.log)
}
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" is configured as an input.
send := false
ip := client.Param("ip")
pins := netsender.MakePins(ip, "V")
for i, pin := range pins {
if pin.Name == "V0" {
send = true
pins[i].Value = len(d)
pins[i].Data = d
pins[i].MimeType = "video/mp2t"
break
}
}
if !send {
return nil
}
var err error
var reply string
reply, _, err = client.Send(netsender.RequestRecv, pins)
if err != nil {
return err
}
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, pkg+"No timestamp in reply")
} else {
log(logger.Debug, fmt.Sprintf("%v got timestamp: %v", pkg, t))
mts.Meta.Add("ts", strconv.Itoa(t))
}
// Extract location from reply
g, err := dec.String("ll")
if err != nil {
log(logger.Debug, pkg+"No location in reply")
} else {
log(logger.Debug, fmt.Sprintf("%v got location: %v", pkg, 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
2019-03-09 07:58:07 +03:00
data []byte
2018-06-09 07:38:48 +03:00
}
func newFileSender(path string) (*fileSender, error) {
2018-06-09 07:38:48 +03:00
f, err := os.Create(path)
if err != nil {
return nil, err
}
return &fileSender{file: f}, nil
}
// Write implements io.Writer.
func (s *fileSender) Write(d []byte) (int, error) {
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 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
rb *ring.Buffer
next []byte
pkt packet.Packet
repairer *mts.DiscontinuityRepairer
curPid int
quit 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{}), rbSize int, rbElementSize int, wTimeout time.Duration) *mtsSender {
s := &mtsSender{
dst: dst,
repairer: mts.NewDiscontinuityRepairer(),
log: log,
rb: ring.NewBuffer(rbSize, rbElementSize, wTimeout),
quit: make(chan struct{}),
}
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 {
case <-s.quit:
s.log(logger.Info, pkg+"mtsSender: got quit signal, terminating 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.rb.Next(rTimeout)
switch err {
case nil, io.EOF:
continue
case ring.ErrTimeout:
s.log(logger.Debug, pkg+"mtsSender: ring buffer read timeout")
continue
default:
s.log(logger.Error, pkg+"mtsSender: unexpected error", "error", err.Error())
continue
}
}
err := s.repairer.Repair(chunk.Bytes())
if err != nil {
chunk.Close()
chunk = nil
continue
}
_, err = s.dst.Write(chunk.Bytes())
if err != nil {
s.repairer.Failed()
continue
}
chunk.Close()
chunk = nil
}
}
}
// Write implements io.Writer.
func (s *mtsSender) Write(d []byte) (int, error) {
if s.next != nil {
s.buf = append(s.buf, s.next...)
}
bytes := make([]byte, len(d))
copy(bytes, d)
s.next = bytes
copy(s.pkt[:], bytes)
s.curPid = s.pkt.PID()
if s.curPid == mts.PatPid && len(s.buf) > 0 {
_, err := s.rb.Write(s.buf)
if err != nil {
s.log(logger.Warning, pkg+"mtsSender: ringBuffer write error", "error", err.Error())
}
s.rb.Flush()
s.buf = s.buf[:0]
}
return len(d), nil
}
// Close implements io.Closer.
func (s *mtsSender) Close() error {
close(s.quit)
s.wg.Wait()
return nil
}
2018-06-09 07:38:48 +03:00
// rtmpSender implements loadSender for a native RTMP destination.
type rtmpSender struct {
2019-04-15 04:18:12 +03:00
conn *rtmp.Conn
2018-06-09 07:38:48 +03:00
url string
timeout uint
retries int
2018-09-11 10:45:45 +03:00
log func(lvl int8, msg string, args ...interface{})
2019-04-15 04:18:12 +03:00
rb *ring.Buffer
quit chan struct{}
wg sync.WaitGroup
2018-06-09 07:38:48 +03:00
}
2018-09-11 10:45:45 +03:00
func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg string, args ...interface{})) (*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, timeout, log)
2018-06-09 07:38:48 +03:00
if err == nil {
break
}
log(logger.Error, err.Error())
2018-06-09 07:38:48 +03:00
if n < retries-1 {
log(logger.Info, pkg+"retry rtmp connection")
2018-06-09 07:38:48 +03:00
}
}
s := &rtmpSender{
conn: conn,
2018-06-09 07:38:48 +03:00
url: url,
timeout: timeout,
retries: retries,
log: log,
2019-04-15 04:18:12 +03:00
rb: ring.NewBuffer(10, rbElementSize, 0),
quit: make(chan struct{}),
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
for {
select {
case <-s.quit:
s.log(logger.Info, pkg+"rtmpSender: got quit signal, terminating 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.rb.Next(rTimeout)
switch err {
case nil, io.EOF:
continue
case ring.ErrTimeout:
s.log(logger.Debug, pkg+"rtmpSender: ring buffer read timeout")
continue
default:
s.log(logger.Error, pkg+"rtmpSender: unexpected error", "error", err.Error())
continue
}
}
if s.conn == nil {
s.log(logger.Warning, pkg+"rtmpSender: no rtmp connection, going to restart...")
err := s.restart()
if err != nil {
s.log(logger.Warning, pkg+"rtmpSender: could not restart connection", "error", err.Error())
continue
}
}
_, err := s.conn.Write(chunk.Bytes())
switch err {
case nil, rtmp.ErrInvalidFlvTag:
default:
2019-04-15 04:18:12 +03:00
s.log(logger.Warning, pkg+"rtmpSender: send error, restarting...", "error", err.Error())
err = s.restart()
if err != nil {
s.log(logger.Warning, pkg+"rtmpSender: could not restart connection", "error", err.Error())
}
continue
}
chunk.Close()
chunk = nil
}
}
}
// Write implements io.Writer.
func (s *rtmpSender) Write(d []byte) (int, error) {
2019-04-15 04:18:12 +03:00
_, err := s.rb.Write(d)
if err != nil {
2019-04-15 04:18:12 +03:00
s.log(logger.Warning, pkg+"rtmpSender: ringBuffer write error", "error", err.Error())
}
2019-04-15 04:18:12 +03:00
s.rb.Flush()
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.conn, err = rtmp.Dial(s.url, s.timeout, s.log)
2018-06-09 07:38:48 +03:00
if err == nil {
break
}
s.log(logger.Error, err.Error())
2018-06-09 07:38:48 +03:00
if n < s.retries-1 {
s.log(logger.Info, pkg+"retry rtmp connection")
2018-06-09 07:38:48 +03:00
}
}
return err
}
func (s *rtmpSender) Close() error {
if s.quit != nil {
close(s.quit)
}
s.wg.Wait()
return s.close()
}
func (s *rtmpSender) close() error {
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
}
func newRtpSender(addr string, log func(lvl int8, msg string, args ...interface{}), fps uint) (*rtpSender, error) {
conn, err := net.Dial("udp", addr)
if err != nil {
return nil, err
}
s := &rtpSender{
log: log,
encoder: rtp.NewEncoder(conn, int(fps)),
}
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, pkg+"rtpSender: write error", err.Error())
}
return len(d), nil
}
func (s *rtpSender) Close() error { return nil }