mirror of https://bitbucket.org/ausocean/av.git
revid: updated code in senders.go to work with meta changes
This commit is contained in:
parent
10343375a3
commit
1f1546a284
|
@ -35,6 +35,7 @@ import (
|
||||||
"net"
|
"net"
|
||||||
"os"
|
"os"
|
||||||
"os/exec"
|
"os/exec"
|
||||||
|
"strconv"
|
||||||
|
|
||||||
"bitbucket.org/ausocean/av/rtmp"
|
"bitbucket.org/ausocean/av/rtmp"
|
||||||
"bitbucket.org/ausocean/av/stream/mts"
|
"bitbucket.org/ausocean/av/stream/mts"
|
||||||
|
@ -172,7 +173,7 @@ func (s *httpSender) extractMeta(r string) error {
|
||||||
s.log(logger.Warning, pkg+"No timestamp in reply")
|
s.log(logger.Warning, pkg+"No timestamp in reply")
|
||||||
} else {
|
} else {
|
||||||
s.log(logger.Debug, fmt.Sprintf("%v got timestamp: %v", pkg, t))
|
s.log(logger.Debug, fmt.Sprintf("%v got timestamp: %v", pkg, t))
|
||||||
mts.MetaData.SetTimeStamp(uint64(t))
|
mts.Meta.Add("ts", strconv.Itoa(t))
|
||||||
}
|
}
|
||||||
|
|
||||||
// Extract location from reply
|
// Extract location from reply
|
||||||
|
@ -181,7 +182,7 @@ func (s *httpSender) extractMeta(r string) error {
|
||||||
s.log(logger.Warning, pkg+"No location in reply")
|
s.log(logger.Warning, pkg+"No location in reply")
|
||||||
} else {
|
} else {
|
||||||
s.log(logger.Debug, fmt.Sprintf("%v got location: %v", pkg, g))
|
s.log(logger.Debug, fmt.Sprintf("%v got location: %v", pkg, g))
|
||||||
mts.MetaData.SetLocation(g)
|
mts.Meta.Add("loc", g)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|
|
@ -32,6 +32,7 @@ import (
|
||||||
"io"
|
"io"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"bitbucket.org/ausocean/av/stream/mts/meta"
|
||||||
"bitbucket.org/ausocean/av/stream/mts/pes"
|
"bitbucket.org/ausocean/av/stream/mts/pes"
|
||||||
"bitbucket.org/ausocean/av/stream/mts/psi"
|
"bitbucket.org/ausocean/av/stream/mts/psi"
|
||||||
)
|
)
|
||||||
|
@ -88,10 +89,10 @@ const (
|
||||||
)
|
)
|
||||||
|
|
||||||
// global Meta
|
// global Meta
|
||||||
var Meta *Metadata
|
var Meta *meta.Metadata
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
Meta = NewMeta()
|
Meta = meta.New()
|
||||||
}
|
}
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
|
|
@ -1,145 +0,0 @@
|
||||||
/*
|
|
||||||
NAME
|
|
||||||
meta.go
|
|
||||||
|
|
||||||
DESCRIPTION
|
|
||||||
See Readme.md
|
|
||||||
|
|
||||||
AUTHOR
|
|
||||||
Saxon Nelson-Milton <saxon@ausocean.org>
|
|
||||||
|
|
||||||
LICENSE
|
|
||||||
meta.go is Copyright (C) 2017-2019 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 mts
|
|
||||||
|
|
||||||
import (
|
|
||||||
"errors"
|
|
||||||
"strings"
|
|
||||||
"sync"
|
|
||||||
)
|
|
||||||
|
|
||||||
const headSize = 4
|
|
||||||
|
|
||||||
const (
|
|
||||||
majVer = 1
|
|
||||||
minVer = 0
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
|
||||||
dataLenIdx1 = 2
|
|
||||||
dataLenIdx2 = 3
|
|
||||||
)
|
|
||||||
|
|
||||||
var (
|
|
||||||
errKeyAbsent = errors.New("Key does not exist in map")
|
|
||||||
)
|
|
||||||
|
|
||||||
type Metadata struct {
|
|
||||||
mu sync.RWMutex
|
|
||||||
data map[string]string
|
|
||||||
enc []byte
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewMeta() *Metadata {
|
|
||||||
return &Metadata{
|
|
||||||
data: make(map[string]string),
|
|
||||||
enc: []byte{
|
|
||||||
0x00, // Reserved byte
|
|
||||||
(majVer << 4) | minVer, // MS and LS versions
|
|
||||||
0x00, // Data len byte1
|
|
||||||
0x00, // Data len byte2
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Add adds metadata with key and val, if already exists return error
|
|
||||||
func (m *Metadata) Add(key, val string) {
|
|
||||||
m.mu.Lock()
|
|
||||||
m.data[key] = val
|
|
||||||
m.mu.Unlock()
|
|
||||||
}
|
|
||||||
|
|
||||||
// All returns the a copy of the map containing the meta data
|
|
||||||
func (m *Metadata) All() map[string]string {
|
|
||||||
m.mu.Lock()
|
|
||||||
cpy := make(map[string]string)
|
|
||||||
for k, v := range m.data {
|
|
||||||
cpy[k] = v
|
|
||||||
}
|
|
||||||
m.mu.Unlock()
|
|
||||||
return cpy
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get returns the meta data for the passed key
|
|
||||||
func (m *Metadata) Get(key string) (string, error) {
|
|
||||||
m.mu.Lock()
|
|
||||||
val, ok := m.data[key]
|
|
||||||
m.mu.Unlock()
|
|
||||||
if !ok {
|
|
||||||
return "", errKeyAbsent
|
|
||||||
}
|
|
||||||
|
|
||||||
return val, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Remove deletes a meta entry in the map and returns error if it doesn’t exist
|
|
||||||
func (m *Metadata) Delete(key string) error {
|
|
||||||
m.mu.Lock()
|
|
||||||
defer m.mu.Unlock()
|
|
||||||
if _, ok := m.data[key]; ok {
|
|
||||||
delete(m.data, key)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return errKeyAbsent
|
|
||||||
}
|
|
||||||
|
|
||||||
// Encode takes the meta data map and encods into a byte slice with header
|
|
||||||
// describing the version, length of data and data in TSV format.
|
|
||||||
func (m *Metadata) Encode() []byte {
|
|
||||||
m.enc = m.enc[:headSize]
|
|
||||||
|
|
||||||
// Iterate over map and append entries, only adding tab if we're not on the last entry
|
|
||||||
var i int
|
|
||||||
var entry string
|
|
||||||
for k, v := range m.data {
|
|
||||||
i++
|
|
||||||
entry += k + "=" + v
|
|
||||||
if i < len(m.data) {
|
|
||||||
entry += "\t"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
m.enc = append(m.enc, []byte(entry)...)
|
|
||||||
|
|
||||||
// Calculate and set data length in encoded meta header.
|
|
||||||
dataLen := len(m.enc[headSize:])
|
|
||||||
m.enc[dataLenIdx1] = byte(dataLen >> 8)
|
|
||||||
m.enc[dataLenIdx2] = byte(dataLen)
|
|
||||||
|
|
||||||
return m.enc
|
|
||||||
}
|
|
||||||
|
|
||||||
func ReadFrom(d []byte, key string) (string, error) {
|
|
||||||
entries := strings.Split(string(d), "\t")
|
|
||||||
for _, entry := range entries {
|
|
||||||
kv := strings.Split(entry, "=")
|
|
||||||
if kv[0] == key {
|
|
||||||
return kv[1], nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return "", errors.New("could not find key in metadata")
|
|
||||||
}
|
|
|
@ -25,7 +25,7 @@ LICENSE
|
||||||
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
|
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package mts
|
package meta
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
|
@ -55,7 +55,7 @@ type Metadata struct {
|
||||||
enc []byte
|
enc []byte
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewMeta() *Metadata {
|
func New() *Metadata {
|
||||||
return &Metadata{
|
return &Metadata{
|
||||||
data: make(map[string]string),
|
data: make(map[string]string),
|
||||||
enc: []byte{
|
enc: []byte{
|
||||||
|
@ -67,7 +67,7 @@ func NewMeta() *Metadata {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Add adds metadata with key and val, if already exists return error
|
// Add adds metadata with key and val
|
||||||
func (m *Metadata) Add(key, val string) {
|
func (m *Metadata) Add(key, val string) {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
m.data[key] = val
|
m.data[key] = val
|
||||||
|
|
|
@ -25,7 +25,7 @@ LICENSE
|
||||||
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
|
along with revid in gpl.txt. If not, see http://www.gnu.org/licenses.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package mts
|
package meta
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
@ -44,7 +44,7 @@ const (
|
||||||
|
|
||||||
// TestAddAndGet ensures that we can add metadata and then successfully get it.
|
// TestAddAndGet ensures that we can add metadata and then successfully get it.
|
||||||
func TestAddAndGet(t *testing.T) {
|
func TestAddAndGet(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
meta.Add(tstKey1, tstData1)
|
meta.Add(tstKey1, tstData1)
|
||||||
meta.Add(tstKey2, tstData2)
|
meta.Add(tstKey2, tstData2)
|
||||||
errors.New("Trying to delete map entry that doesn't exist")
|
errors.New("Trying to delete map entry that doesn't exist")
|
||||||
|
@ -66,7 +66,7 @@ func TestAddAndGet(t *testing.T) {
|
||||||
// TestUpdate checks that we can use Meta.Add to actually update metadata
|
// TestUpdate checks that we can use Meta.Add to actually update metadata
|
||||||
// if it already exists in the Meta map.
|
// if it already exists in the Meta map.
|
||||||
func TestUpdate(t *testing.T) {
|
func TestUpdate(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
meta.Add(tstKey1, tstData1)
|
meta.Add(tstKey1, tstData1)
|
||||||
meta.Add(tstKey1, tstData3)
|
meta.Add(tstKey1, tstData3)
|
||||||
|
|
||||||
|
@ -80,7 +80,7 @@ func TestUpdate(t *testing.T) {
|
||||||
|
|
||||||
// TestAll ensures we can get a correct map using Meta.All() after adding some data
|
// TestAll ensures we can get a correct map using Meta.All() after adding some data
|
||||||
func TestAll(t *testing.T) {
|
func TestAll(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
tstMap := map[string]string{
|
tstMap := map[string]string{
|
||||||
tstKey1: tstData1,
|
tstKey1: tstData1,
|
||||||
tstKey2: tstData2,
|
tstKey2: tstData2,
|
||||||
|
@ -98,7 +98,7 @@ func TestAll(t *testing.T) {
|
||||||
// TestGetAbsentKey ensures that we get the expected error when we try to get with
|
// TestGetAbsentKey ensures that we get the expected error when we try to get with
|
||||||
// key that does not yet exist in the Meta map.
|
// key that does not yet exist in the Meta map.
|
||||||
func TestGetAbsentKey(t *testing.T) {
|
func TestGetAbsentKey(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
|
|
||||||
if _, err := meta.Get(tstKey1); err != errKeyAbsent {
|
if _, err := meta.Get(tstKey1); err != errKeyAbsent {
|
||||||
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
||||||
|
@ -107,7 +107,7 @@ func TestGetAbsentKey(t *testing.T) {
|
||||||
|
|
||||||
// TestDelete ensures we can remove a data entry in the Meta map.
|
// TestDelete ensures we can remove a data entry in the Meta map.
|
||||||
func TestDelete(t *testing.T) {
|
func TestDelete(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
meta.Add(tstKey1, tstData1)
|
meta.Add(tstKey1, tstData1)
|
||||||
if err := meta.Delete(tstKey1); err != nil {
|
if err := meta.Delete(tstKey1); err != nil {
|
||||||
t.Errorf("Did not expect error: %v", err.Error())
|
t.Errorf("Did not expect error: %v", err.Error())
|
||||||
|
@ -120,7 +120,7 @@ func TestDelete(t *testing.T) {
|
||||||
// TestDeleteAbsentKey checks that we get an expected error when we try to delete
|
// TestDeleteAbsentKey checks that we get an expected error when we try to delete
|
||||||
// an entry in the Meta map that doesn't exist.
|
// an entry in the Meta map that doesn't exist.
|
||||||
func TestDeleteAbsentKey(t *testing.T) {
|
func TestDeleteAbsentKey(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
if err := meta.Delete(tstKey1); err != errKeyAbsent {
|
if err := meta.Delete(tstKey1); err != errKeyAbsent {
|
||||||
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
||||||
}
|
}
|
||||||
|
@ -128,7 +128,7 @@ func TestDeleteAbsentKey(t *testing.T) {
|
||||||
|
|
||||||
// TestEncode checks that we're getting the correct byte slice from Meta.Encode().
|
// TestEncode checks that we're getting the correct byte slice from Meta.Encode().
|
||||||
func TestEncode(t *testing.T) {
|
func TestEncode(t *testing.T) {
|
||||||
meta := NewMeta()
|
meta := New()
|
||||||
meta.Add(tstKey1, tstData1)
|
meta.Add(tstKey1, tstData1)
|
||||||
meta.Add(tstKey2, tstData2)
|
meta.Add(tstKey2, tstData2)
|
||||||
|
|
||||||
|
|
|
@ -1,150 +0,0 @@
|
||||||
/*
|
|
||||||
NAME
|
|
||||||
meta_test.go
|
|
||||||
|
|
||||||
DESCRIPTION
|
|
||||||
See Readme.md
|
|
||||||
|
|
||||||
AUTHOR
|
|
||||||
Saxon Nelson-Milton <saxon@ausocean.org>
|
|
||||||
|
|
||||||
LICENSE
|
|
||||||
meta_test.go is Copyright (C) 2017-2019 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 mts
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"errors"
|
|
||||||
"reflect"
|
|
||||||
"testing"
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
|
||||||
tstKey1 = "loc"
|
|
||||||
tstData1 = "a,b,c"
|
|
||||||
tstKey2 = "ts"
|
|
||||||
tstData2 = "12345678"
|
|
||||||
tstData3 = "d,e,f"
|
|
||||||
)
|
|
||||||
|
|
||||||
// TestAddAndGet ensures that we can add metadata and then successfully get it.
|
|
||||||
func TestAddAndGet(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
meta.Add(tstKey1, tstData1)
|
|
||||||
meta.Add(tstKey2, tstData2)
|
|
||||||
errors.New("Trying to delete map entry that doesn't exist")
|
|
||||||
if data, err := meta.Get(tstKey1); err != nil {
|
|
||||||
t.Errorf("Could not get data for key: loc: %v", err.Error())
|
|
||||||
if data != tstData1 {
|
|
||||||
t.Errorf("Did not get expected data")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if data, err := meta.Get(tstKey2); err != nil {
|
|
||||||
t.Errorf("Could not get data for key: ts: %v", err.Error())
|
|
||||||
if data != tstData2 {
|
|
||||||
t.Errorf("Did not get expected data")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestUpdate checks that we can use Meta.Add to actually update metadata
|
|
||||||
// if it already exists in the Meta map.
|
|
||||||
func TestUpdate(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
meta.Add(tstKey1, tstData1)
|
|
||||||
meta.Add(tstKey1, tstData3)
|
|
||||||
|
|
||||||
if data, err := meta.Get(tstKey1); err != nil {
|
|
||||||
t.Errorf("Did not expect err: %v", err.Error())
|
|
||||||
if data != tstData2 {
|
|
||||||
t.Errorf("Data did not correctly update for key \"loc\"")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestAll ensures we can get a correct map using Meta.All() after adding some data
|
|
||||||
func TestAll(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
tstMap := map[string]string{
|
|
||||||
tstKey1: tstData1,
|
|
||||||
tstKey2: tstData2,
|
|
||||||
}
|
|
||||||
|
|
||||||
meta.Add(tstKey1, tstData1)
|
|
||||||
meta.Add(tstKey2, tstData2)
|
|
||||||
metaMap := meta.All()
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(metaMap, tstMap) {
|
|
||||||
t.Errorf("Map not correct. Got: %v, want: %v", metaMap, tstMap)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestGetAbsentKey ensures that we get the expected error when we try to get with
|
|
||||||
// key that does not yet exist in the Meta map.
|
|
||||||
func TestGetAbsentKey(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
|
|
||||||
if _, err := meta.Get(tstKey1); err != errKeyAbsent {
|
|
||||||
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestDelete ensures we can remove a data entry in the Meta map.
|
|
||||||
func TestDelete(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
meta.Add(tstKey1, tstData1)
|
|
||||||
if err := meta.Delete(tstKey1); err != nil {
|
|
||||||
t.Errorf("Did not expect error: %v", err.Error())
|
|
||||||
}
|
|
||||||
if _, err := meta.Get(tstKey1); err != errKeyAbsent {
|
|
||||||
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestDeleteAbsentKey checks that we get an expected error when we try to delete
|
|
||||||
// an entry in the Meta map that doesn't exist.
|
|
||||||
func TestDeleteAbsentKey(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
if err := meta.Delete(tstKey1); err != errKeyAbsent {
|
|
||||||
t.Errorf("Did not get expected err: %v", errKeyAbsent.Error())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestEncode checks that we're getting the correct byte slice from Meta.Encode().
|
|
||||||
func TestEncode(t *testing.T) {
|
|
||||||
meta := NewMeta()
|
|
||||||
meta.Add(tstKey1, tstData1)
|
|
||||||
meta.Add(tstKey2, tstData2)
|
|
||||||
|
|
||||||
dataLen := len(tstKey1+tstData1+tstKey2+tstData2) + 3
|
|
||||||
expectedOut := []byte{
|
|
||||||
0x00,
|
|
||||||
0x10,
|
|
||||||
byte(dataLen >> 8),
|
|
||||||
byte(dataLen),
|
|
||||||
}
|
|
||||||
expectedOut = append(expectedOut, []byte(
|
|
||||||
tstKey1+"="+tstData1+"\t"+
|
|
||||||
tstKey2+"="+tstData2)...)
|
|
||||||
|
|
||||||
got := meta.Encode()
|
|
||||||
if !bytes.Equal(expectedOut, got) {
|
|
||||||
t.Errorf("Did not get expected out. \nGot : %v \nwant: %v", got, expectedOut)
|
|
||||||
}
|
|
||||||
}
|
|
Loading…
Reference in New Issue