make PubSubRecord an interface

Author: keks

pubsub.go

@@ -1,50 +1,46 @@
 package shell
 
 import (
-	"encoding/base64"
+	"encoding/binary"
 	"encoding/json"
-)
 
-// floodsub uses base64 encoding for just about everything.
-// To Decode the base64 while also decoding the JSON the type b64String was added.
-type b64String string
+	"github.com/libp2p/go-floodsub"
+	"github.com/libp2p/go-libp2p-peer"
+)
 
-// UnmarshalJSON implements the json.Unmarshaler interface.
-func (bs *b64String) UnmarshalJSON(in []byte) error {
-	var b64 string
+// PubSubRecord is a record received via PubSub.
+type PubSubRecord interface {
+	// From returns the peer ID of the node that published this record
+	From() peer.ID
 
-	err := json.Unmarshal(in, &b64)
-	if err != nil {
-		return err
-	}
+	// Data returns the data field
+	Data() []byte
 
-	bsStr, err := base64.StdEncoding.DecodeString(b64)
+	// SeqNo is the sequence number of this record
+	SeqNo() int64
 
-	*bs = b64String(bsStr)
-	return err
+	//TopicIDs is the list of topics this record belongs to
+	TopicIDs() []string
 }
 
-func (bs *b64String) Marshal() (string, error) {
-	jsonBytes, err := json.Marshal(
-		base64.StdEncoding.EncodeToString(
-			[]byte(*bs)))
+type floodsubRecord struct {
+	msg *floodsub.Message
+}
 
-	return string(jsonBytes), err
+func (r floodsubRecord) From() peer.ID {
+	return r.msg.GetFrom()
 }
 
-///
+func (r floodsubRecord) Data() []byte {
+	return r.msg.GetData()
+}
 
-// PubSubRecord is a record received via PubSub.
-type PubSubRecord struct {
-	From     string    `json:"from"`
-	Data     b64String `json:"data"`
-	SeqNo    b64String `json:"seqno"`
-	TopicIDs []string  `json:"topicIDs"`
+func (r floodsubRecord) SeqNo() int64 {
+	return int64(binary.BigEndian.Uint64(r.msg.GetSeqno()))
 }
 
-// DataString returns the string representation of the data field.
-func (r PubSubRecord) DataString() string {
-	return string(r.Data)
+func (r floodsubRecord) TopicIDs() []string {
+	return r.msg.GetTopicIDs()
 }
 
 ///
@@ -64,17 +60,17 @@ func newPubSubSubscription(resp *Response) *PubSubSubscription {
 }
 
 // Next waits for the next record and returns that.
-func (s *PubSubSubscription) Next() (*PubSubRecord, error) {
+func (s *PubSubSubscription) Next() (PubSubRecord, error) {
 	if s.resp.Error != nil {
 		return nil, s.resp.Error
 	}
 
 	d := json.NewDecoder(s.resp.Output)
 
-	r := &PubSubRecord{}
+	r := &floodsub.Message{}
 	err := d.Decode(r)
 
-	return r, err
+	return floodsubRecord{msg: r}, err
 }
 
 // Cancel cancels the given subscription.

shell_test.go

@@ -147,7 +147,7 @@ func TestPubSub(t *testing.T) {
 
 	is.Nil(err)
 	is.NotNil(r)
-	is.Equal(r.DataString(), "Hello World!")
+	is.Equal(r.Data(), "Hello World!")
 
 	sub2, err := s.PubSubSubscribe(topic)
 	is.Nil(err)
@@ -158,12 +158,12 @@ func TestPubSub(t *testing.T) {
 	r, err = sub2.Next()
 	is.Nil(err)
 	is.NotNil(r)
-	is.Equal(r.Data, "Hallo Welt!")
+	is.Equal(r.Data(), "Hallo Welt!")
 
 	r, err = sub.Next()
 	is.NotNil(r)
 	is.Nil(err)
-	is.Equal(r.Data, "Hallo Welt!")
+	is.Equal(r.Data(), "Hallo Welt!")
 
 	is.Nil(sub.Cancel())
 }