Rewrite ivideon source
This commit is contained in:
Alex X
@@ -1,314 +0,0 @@
|
||||
package ivideon
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/AlexxIT/go2rtc/pkg/core"
|
||||
"github.com/AlexxIT/go2rtc/pkg/h264"
|
||||
"github.com/AlexxIT/go2rtc/pkg/iso"
|
||||
"github.com/gorilla/websocket"
|
||||
"github.com/pion/rtp"
|
||||
)
|
||||
|
||||
type State byte
|
||||
|
||||
const (
|
||||
StateNone State = iota
|
||||
StateConn
|
||||
StateHandle
|
||||
)
|
||||
|
||||
// Deprecated: should be rewritten to core.Connection
|
||||
type Client struct {
|
||||
core.Listener
|
||||
|
||||
ID string
|
||||
|
||||
conn *websocket.Conn
|
||||
|
||||
medias []*core.Media
|
||||
receiver *core.Receiver
|
||||
|
||||
msg *message
|
||||
t0 time.Time
|
||||
|
||||
buffer chan []byte
|
||||
state State
|
||||
mu sync.Mutex
|
||||
|
||||
recv int
|
||||
}
|
||||
|
||||
func Dial(source string) (*Client, error) {
|
||||
id := strings.Replace(source[8:], "/", ":", 1)
|
||||
client := &Client{ID: id}
|
||||
if err := client.Dial(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return client, nil
|
||||
}
|
||||
|
||||
func (c *Client) Dial() (err error) {
|
||||
resp, err := http.Get(
|
||||
"https://openapi-alpha.ivideon.com/cameras/" + c.ID +
|
||||
"/live_stream?op=GET&access_token=public&q=2&" +
|
||||
"video_codecs=h264&format=ws-fmp4",
|
||||
)
|
||||
|
||||
data, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var v liveResponse
|
||||
if err = json.Unmarshal(data, &v); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if !v.Success {
|
||||
return fmt.Errorf("wrong response: %s", data)
|
||||
}
|
||||
|
||||
c.conn, _, err = websocket.DefaultDialer.Dial(v.Result.URL, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = c.getTracks(); err != nil {
|
||||
_ = c.conn.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
c.state = StateConn
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) Handle() error {
|
||||
// add delay to the stream for smooth playing (not a best solution)
|
||||
c.t0 = time.Now().Add(time.Second)
|
||||
|
||||
c.mu.Lock()
|
||||
|
||||
if c.state == StateConn {
|
||||
c.buffer = make(chan []byte, 5)
|
||||
c.state = StateHandle
|
||||
|
||||
// processing stream in separate thread for lower delay between packets
|
||||
go c.worker(c.buffer)
|
||||
}
|
||||
|
||||
c.mu.Unlock()
|
||||
|
||||
_, data, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if c.receiver != nil && c.receiver.ID == c.msg.Track {
|
||||
c.mu.Lock()
|
||||
if c.state == StateHandle {
|
||||
c.buffer <- data
|
||||
c.recv += len(data)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// we have one unprocessed msg after getTracks
|
||||
for {
|
||||
_, data, err = c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var msg message
|
||||
if err = json.Unmarshal(data, &msg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case "stream-init":
|
||||
continue
|
||||
|
||||
case "metadata":
|
||||
continue
|
||||
|
||||
case "fragment":
|
||||
_, data, err = c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if c.receiver != nil && c.receiver.ID == msg.Track {
|
||||
c.mu.Lock()
|
||||
if c.state == StateHandle {
|
||||
c.buffer <- data
|
||||
c.recv += len(data)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
default:
|
||||
return fmt.Errorf("wrong message type: %s", data)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) Close() error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
switch c.state {
|
||||
case StateNone:
|
||||
return nil
|
||||
case StateConn:
|
||||
case StateHandle:
|
||||
close(c.buffer)
|
||||
}
|
||||
|
||||
c.state = StateNone
|
||||
|
||||
return c.conn.Close()
|
||||
}
|
||||
|
||||
func (c *Client) getTracks() error {
|
||||
for {
|
||||
_, data, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var msg message
|
||||
if err = json.Unmarshal(data, &msg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case "metadata":
|
||||
continue
|
||||
|
||||
case "stream-init":
|
||||
s := msg.CodecString
|
||||
i := strings.IndexByte(s, '.')
|
||||
if i > 0 {
|
||||
s = s[:i]
|
||||
}
|
||||
|
||||
switch s {
|
||||
case "avc1": // avc1.4d0029
|
||||
// skip multiple identical init
|
||||
if c.receiver != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
i = bytes.Index(msg.Data, []byte("avcC")) - 4
|
||||
if i < 0 {
|
||||
return fmt.Errorf("ivideon: wrong AVC: %s", msg.Data)
|
||||
}
|
||||
|
||||
avccLen := binary.BigEndian.Uint32(msg.Data[i:])
|
||||
data = msg.Data[i+8 : i+int(avccLen)]
|
||||
|
||||
codec := h264.ConfigToCodec(data)
|
||||
|
||||
media := &core.Media{
|
||||
Kind: core.KindVideo,
|
||||
Direction: core.DirectionRecvonly,
|
||||
Codecs: []*core.Codec{codec},
|
||||
}
|
||||
c.medias = append(c.medias, media)
|
||||
|
||||
c.receiver = core.NewReceiver(media, codec)
|
||||
c.receiver.ID = msg.TrackID
|
||||
|
||||
case "mp4a": // mp4a.40.2
|
||||
}
|
||||
|
||||
case "fragment":
|
||||
c.msg = &msg
|
||||
return nil
|
||||
|
||||
default:
|
||||
return fmt.Errorf("wrong message type: %s", data)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) worker(buffer chan []byte) {
|
||||
for data := range buffer {
|
||||
atoms, err := iso.DecodeAtoms(data)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
var trun *iso.Atom
|
||||
var ts uint32
|
||||
|
||||
for _, atom := range atoms {
|
||||
switch atom.Name {
|
||||
case iso.MoofTrafTrun:
|
||||
trun = atom
|
||||
case iso.MoofTrafTfdt:
|
||||
ts = uint32(atom.DecodeTime)
|
||||
case iso.Mdat:
|
||||
data = atom.Data
|
||||
}
|
||||
}
|
||||
|
||||
if trun == nil || trun.SamplesDuration == nil || trun.SamplesSize == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
for i := 0; i < len(trun.SamplesDuration); i++ {
|
||||
duration := trun.SamplesDuration[i]
|
||||
size := trun.SamplesSize[i]
|
||||
|
||||
// synchronize framerate for WebRTC and MSE
|
||||
d := time.Duration(ts)*time.Millisecond - time.Since(c.t0)
|
||||
if d < 0 {
|
||||
d = time.Duration(duration) * time.Millisecond / 2
|
||||
}
|
||||
time.Sleep(d)
|
||||
|
||||
// can be SPS, PPS and IFrame in one packet
|
||||
packet := &rtp.Packet{
|
||||
// ivideon clockrate=1000, RTP clockrate=90000
|
||||
Header: rtp.Header{Timestamp: ts * 90},
|
||||
Payload: data[:size],
|
||||
}
|
||||
c.receiver.WriteRTP(packet)
|
||||
|
||||
data = data[size:]
|
||||
ts += duration
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type liveResponse struct {
|
||||
Result struct {
|
||||
URL string `json:"url"`
|
||||
} `json:"result"`
|
||||
Success bool `json:"success"`
|
||||
}
|
||||
|
||||
type message struct {
|
||||
Type string `json:"type"`
|
||||
|
||||
CodecString string `json:"codec_string"`
|
||||
Data []byte `json:"data"`
|
||||
TrackID byte `json:"track_id"`
|
||||
|
||||
Track byte `json:"track"`
|
||||
StartTime float32 `json:"start_time"`
|
||||
Duration float32 `json:"duration"`
|
||||
IsKey bool `json:"is_key"`
|
||||
DataOffset uint32 `json:"data_offset"`
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
package ivideon
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/AlexxIT/go2rtc/pkg/core"
|
||||
"github.com/AlexxIT/go2rtc/pkg/mp4"
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
type Producer struct {
|
||||
core.Connection
|
||||
conn *websocket.Conn
|
||||
|
||||
buf []byte
|
||||
|
||||
dem *mp4.Demuxer
|
||||
}
|
||||
|
||||
func Dial(source string) (core.Producer, error) {
|
||||
id := strings.Replace(source[8:], "/", ":", 1)
|
||||
|
||||
url, err := GetLiveStream(id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
conn, _, err := websocket.DefaultDialer.Dial(url, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
prod := &Producer{
|
||||
Connection: core.Connection{
|
||||
ID: core.NewID(),
|
||||
FormatName: "ivideon",
|
||||
Protocol: core.Before(url, ":"), // wss
|
||||
RemoteAddr: conn.RemoteAddr().String(),
|
||||
Source: source,
|
||||
URL: url,
|
||||
Transport: conn,
|
||||
},
|
||||
conn: conn,
|
||||
}
|
||||
|
||||
if err = prod.probe(); err != nil {
|
||||
_ = conn.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return prod, nil
|
||||
}
|
||||
|
||||
func GetLiveStream(id string) (string, error) {
|
||||
// &video_codecs=h264,h265&audio_codecs=aac,mp3,pcma,pcmu,none
|
||||
resp, err := http.Get(
|
||||
"https://openapi-alpha.ivideon.com/cameras/" + id +
|
||||
"/live_stream?op=GET&access_token=public&q=2&video_codecs=h264&format=ws-fmp4",
|
||||
)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
var v struct {
|
||||
Message string `json:"message"`
|
||||
Result struct {
|
||||
URL string `json:"url"`
|
||||
} `json:"result"`
|
||||
Success bool `json:"success"`
|
||||
}
|
||||
if err = json.NewDecoder(resp.Body).Decode(&v); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if !v.Success {
|
||||
return "", fmt.Errorf("ivideon: can't get live_stream: " + v.Message)
|
||||
}
|
||||
|
||||
return v.Result.URL, nil
|
||||
}
|
||||
|
||||
func (p *Producer) Start() error {
|
||||
receivers := make(map[uint32]*core.Receiver)
|
||||
for _, receiver := range p.Receivers {
|
||||
trackID := p.dem.GetTrackID(receiver.Codec)
|
||||
receivers[trackID] = receiver
|
||||
}
|
||||
|
||||
ch := make(chan []byte, 10)
|
||||
defer close(ch)
|
||||
|
||||
ch <- p.buf
|
||||
|
||||
go func() {
|
||||
// add delay to the stream for smooth playing (not a best solution)
|
||||
t0 := time.Now()
|
||||
|
||||
for data := range ch {
|
||||
trackID, packets := p.dem.Demux(data)
|
||||
if receiver := receivers[trackID]; receiver != nil {
|
||||
clockRate := time.Duration(receiver.Codec.ClockRate)
|
||||
for _, packet := range packets {
|
||||
// synchronize framerate for WebRTC and MSE
|
||||
ts := time.Second * time.Duration(packet.Timestamp) / clockRate
|
||||
d := ts - time.Since(t0)
|
||||
if d < 0 {
|
||||
d = 10 * time.Millisecond
|
||||
}
|
||||
time.Sleep(d)
|
||||
|
||||
receiver.WriteRTP(packet)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
var msg message
|
||||
if err := p.conn.ReadJSON(&msg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case "stream-init", "metadata":
|
||||
continue
|
||||
|
||||
case "fragment":
|
||||
_, b, err := p.conn.ReadMessage()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
p.Recv += len(b)
|
||||
ch <- b
|
||||
|
||||
default:
|
||||
return errors.New("ivideon: wrong message type: " + msg.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Producer) probe() (err error) {
|
||||
p.dem = &mp4.Demuxer{}
|
||||
|
||||
for {
|
||||
var msg message
|
||||
if err = p.conn.ReadJSON(&msg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case "metadata":
|
||||
continue
|
||||
|
||||
case "stream-init":
|
||||
// it's difficult to maintain audio
|
||||
if strings.HasPrefix(msg.CodecString, "avc1") {
|
||||
medias := p.dem.Probe(msg.Data)
|
||||
p.Medias = append(p.Medias, medias...)
|
||||
}
|
||||
|
||||
case "fragment":
|
||||
_, p.buf, err = p.conn.ReadMessage()
|
||||
return
|
||||
|
||||
default:
|
||||
return errors.New("ivideon: wrong message type: " + msg.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type message struct {
|
||||
Type string `json:"type"`
|
||||
CodecString string `json:"codec_string"`
|
||||
Data []byte `json:"data"`
|
||||
//TrackID byte `json:"track_id"`
|
||||
//Track byte `json:"track"`
|
||||
//StartTime float32 `json:"start_time"`
|
||||
//Duration float32 `json:"duration"`
|
||||
//IsKey bool `json:"is_key"`
|
||||
//DataOffset uint32 `json:"data_offset"`
|
||||
}
|
||||
@@ -1,51 +0,0 @@
|
||||
package ivideon
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"github.com/AlexxIT/go2rtc/pkg/core"
|
||||
)
|
||||
|
||||
func (c *Client) GetMedias() []*core.Media {
|
||||
return c.medias
|
||||
}
|
||||
|
||||
func (c *Client) GetTrack(media *core.Media, codec *core.Codec) (*core.Receiver, error) {
|
||||
if c.receiver != nil {
|
||||
return c.receiver, nil
|
||||
}
|
||||
return nil, core.ErrCantGetTrack
|
||||
}
|
||||
|
||||
func (c *Client) Start() error {
|
||||
err := c.Handle()
|
||||
if c.buffer == nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *Client) Stop() error {
|
||||
if c.receiver != nil {
|
||||
c.receiver.Close()
|
||||
}
|
||||
return c.Close()
|
||||
}
|
||||
|
||||
func (c *Client) MarshalJSON() ([]byte, error) {
|
||||
info := &core.Connection{
|
||||
ID: core.ID(c),
|
||||
FormatName: "ivideon",
|
||||
Protocol: "ws",
|
||||
URL: c.ID,
|
||||
Medias: c.medias,
|
||||
Recv: c.recv,
|
||||
}
|
||||
if c.conn != nil {
|
||||
info.RemoteAddr = c.conn.RemoteAddr().String()
|
||||
}
|
||||
if c.receiver != nil {
|
||||
info.Receivers = []*core.Receiver{c.receiver}
|
||||
}
|
||||
return json.Marshal(info)
|
||||
}
|
||||
Reference in New Issue
Block a user