/* 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" "sync" "time" "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" "bitbucket.org/ausocean/utils/ring" ) // 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 (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.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) (*fileSender, 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 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 ringBuf *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, ringBuf: 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.ringBuf.Next(readTimeout) switch err { case nil: 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()) fallthrough case io.EOF: 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.ringBuf.Write(s.buf) if err != nil { s.log(logger.Warning, pkg+"mtsSender: ringBuffer write error", "error", err.Error()) } s.ringBuf.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 } // 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 nil } return s.conn.Close() } // 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) { 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 }