hub: parse version and require version 1

4 files changed, 10 insertions, 4 deletions

diff --git a/src/hub/src/spreadspace.org/sfive/s5cvt.go b/src/hub/src/spreadspace.org/sfive/s5cvt.go
index d828af7..24b458a 100644
--- a/src/hub/src/spreadspace.org/sfive/s5cvt.go
+++ b/src/hub/src/spreadspace.org/sfive/s5cvt.go
@@ -2,6 +2,7 @@ package sfive
 
 import (
 	"encoding/json"
+	"fmt"
 )
 
 type StatsDecoder interface {
@@ -26,6 +27,9 @@ func NewStatefulDecoder(jsonString []byte) (decoder StatsDecoder, err error) {
 	if err != nil {
 		return
 	}
+	if res.sourceId.Version != 1 {
+		err = fmt.Errorf("unsupported version, expected 1, actual %v", res.sourceId.Version)
+	}
 	decoder = res
 	return
 }
@@ -35,8 +39,8 @@ func NewPlainDecoder() (decoder StatsDecoder) {
 }
 
 func (self *StatefulDecoder) Decode(jsonString []byte) (dat StatisticsData, err error) {
-	err = json.Unmarshal(jsonString, &dat)
 	dat.CopyFrom(&self.sourceId)
+	err = json.Unmarshal(jsonString, &dat)
 	// like in PlainDecoder, let the client decide how to use partial results
 	// (Unmarshal returns partial results in case of errors)
 	return
diff --git a/src/hub/src/spreadspace.org/sfive/s5cvt_test.go b/src/hub/src/spreadspace.org/sfive/s5cvt_test.go
index 418d154..2326911 100644
--- a/src/hub/src/spreadspace.org/sfive/s5cvt_test.go
+++ b/src/hub/src/spreadspace.org/sfive/s5cvt_test.go
@@ -7,9 +7,9 @@ import (
 )
 
 var (
-	sourceIdFields     = `"hostname": "localhost", "streamer-id": {"quality": "low", "content-id": "av", "format": "webm"}, "tags": ["elevate", "2014"]`
+	sourceIdFields     = `"hostname": "localhost", "streamer-id": {"quality": "low", "content-id": "av", "format": "webm"}, "tags": ["elevate", "2014"], "version": 1`
 	sourceIdData       = `{` + sourceIdFields + `}`
-	sourceIdDataStruct = SourceId{Hostname: "localhost", StreamId: StreamId{Quality: "low", ContentId: "av", Format: "webm"}, Tags: []string{"elevate", "2014"}}
+	sourceIdDataStruct = SourceId{Hostname: "localhost", StreamId: StreamId{Quality: "low", ContentId: "av", Format: "webm"}, Tags: []string{"elevate", "2014"}, Version: 1}
 	updateFields       = `"data": {"bytes-sent": 1, "client-count": 3, "bytes-received": 1}, "start-time": "2014-08-24T14:35:33.847282Z", "duration-ms": 5000`
 	updateData         = "{" + updateFields + "}"
 	updateDataStruct   = DataUpdate{Data: SourceData{BytesSent: 1, ClientCount: 3, BytesReceived: 1}, StartTime: time.Date(2014, time.August, 24, 14, 35, 33, 847282000, time.UTC), Duration: 5000}
diff --git a/src/hub/src/spreadspace.org/sfive/s5store_test.go b/src/hub/src/spreadspace.org/sfive/s5store_test.go
index 3ff437e..356b42f 100644
--- a/src/hub/src/spreadspace.org/sfive/s5store_test.go
+++ b/src/hub/src/spreadspace.org/sfive/s5store_test.go
@@ -15,7 +15,7 @@ func TestAppend(t *testing.T) {
 	startTime := time.Date(2014, time.August, 24, 14, 35, 33, 847282000, time.UTC)
 	update := DataUpdate{Data: SourceData{BytesSent: 1, ClientCount: 3, BytesReceived: 1}, StartTime: startTime, Duration: 5 * time.Millisecond}
 	streamId := StreamId{ContentId: "content", Format: "7bitascii", Quality: QualityHigh}
-	source := SourceId{Hostname: "localhost", Tags: []string{"tag1", "master"}, StreamId: streamId}
+	source := SourceId{Hostname: "localhost", Tags: []string{"tag1", "master"}, StreamId: streamId, Version: 1}
 	dat := StatisticsData{source, update}
 
 	err = store.Append(dat)
diff --git a/src/hub/src/spreadspace.org/sfive/s5types.go b/src/hub/src/spreadspace.org/sfive/s5types.go
index 64b3f48..74c7f1c 100644
--- a/src/hub/src/spreadspace.org/sfive/s5types.go
+++ b/src/hub/src/spreadspace.org/sfive/s5types.go
@@ -15,6 +15,7 @@ type StreamId struct {
 }
 
 type SourceId struct {
+	Version  uint     `json:"version" db:"-"`
 	Hostname string   `json:"hostname"`
 	StreamId StreamId `json:"streamer-id" db:"-"`
 	Tags     []string `json:"tags" db:"-"`
@@ -48,6 +49,7 @@ func (self *StatisticsData) CopyFrom(id *SourceId) {
 	self.Hostname = id.Hostname
 	self.StreamId = id.StreamId
 	self.Tags = id.Tags
+	self.Version = id.Version
 }
 
 func (self *StatisticsData) CopyFromUpdate(id *DataUpdate) {