av/revid/senders.go

287 lines
6.8 KiB
Go

/*
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"
"os"
"strconv"
"github.com/Comcast/gots/packet"
"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"
)
// Log is used by the multiSender.
type Log func(level int8, message string, params ...interface{})
// 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,
}
}
// Write implements io.Writer.
func (s *httpSender) Write(d []byte) (int, error) {
return len(d), httpSend(d, s.client, s.log)
}
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.Warning, pkg+"No location in reply")
} else {
log(logger.Debug, fmt.Sprintf("%v got location: %v", pkg, g))
mts.Meta.Add("loc", g)
}
return nil
}
// fileSender implements loadSender for a local file destination.
type fileSender struct {
file *os.File
data []byte
}
func newFileSender(path string) (io.Writer, error) {
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)
}
func (s *fileSender) close() error { return s.file.Close() }
// mtsSender implements loadSender and provides sending capability specifically
// for use with MPEGTS packetization. It handles the construction of appropriately
// lengthed clips based on PSI. It also fixes accounts for discontinuities by
// setting the discontinuity indicator for the first packet of a clip.
type mtsSender struct {
dst io.Writer
buf []byte
next []byte
pkt packet.Packet
repairer *mts.DiscontinuityRepairer
curPid int
}
// newMtsSender returns a new mtsSender.
func newMtsSender(dst io.Writer, log func(lvl int8, msg string, args ...interface{})) *mtsSender {
return &mtsSender{
dst: dst,
repairer: mts.NewDiscontinuityRepairer(),
}
}
// 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.repairer.Repair(s.buf)
if err == nil {
_, err = s.dst.Write(s.buf)
if err == nil {
goto done
}
}
s.repairer.Failed()
done:
s.buf = s.buf[:0]
}
return len(d), nil
}
// rtmpSender implements loadSender for a native RTMP destination.
type rtmpSender struct {
conn *rtmp.Conn
url string
timeout uint
retries int
log func(lvl int8, msg string, args ...interface{})
data []byte
}
func newRtmpSender(url string, timeout uint, retries int, log func(lvl int8, msg string, args ...interface{})) (*rtmpSender, error) {
var conn *rtmp.Conn
var err error
for n := 0; n < retries; n++ {
conn, err = rtmp.Dial(url, timeout, log)
if err == nil {
break
}
log(logger.Error, err.Error())
if n < retries-1 {
log(logger.Info, pkg+"retry rtmp connection")
}
}
s := &rtmpSender{
conn: conn,
url: url,
timeout: timeout,
retries: retries,
log: log,
}
return s, err
}
// Write implements io.Writer.
func (s *rtmpSender) Write(d []byte) (int, error) {
if s.conn == nil {
return 0, errors.New("no rtmp connection, cannot write")
}
_, err := s.conn.Write(d)
if err != nil {
err = s.restart()
}
return len(d), err
}
func (s *rtmpSender) restart() error {
s.close()
var err error
for n := 0; n < s.retries; n++ {
s.conn, err = rtmp.Dial(s.url, s.timeout, s.log)
if err == nil {
break
}
s.log(logger.Error, err.Error())
if n < s.retries-1 {
s.log(logger.Info, pkg+"retry rtmp connection")
}
}
return err
}
func (s *rtmpSender) close() error {
if s.conn != nil {
return s.conn.Close()
}
return nil
}
// 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
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) {
return s.encoder.Write(s.data)
}