reflect recent server-side pubsub changes, e.g. flushing

Author: keks

pubsub.go

@@ -3,11 +3,13 @@ package shell
 import (
 	"encoding/base64"
 	"encoding/json"
-	//	"sync"
 )
 
+// 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
 
+// UnmarshalJSON implements the json.Unmarshaler interface.
 func (bs *b64String) UnmarshalJSON(in []byte) error {
 	var b64 string
 
@@ -32,29 +34,36 @@ func (bs *b64String) Marshal() (string, error) {
 
 ///
 
+// 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"`
 }
 
+// DataString returns the string representation of the data field.
 func (r PubSubRecord) DataString() string {
 	return string(r.Data)
 }
 
 ///
 
+// PubSubSubscription allow you to receive pubsub records that where published on the network.
 type PubSubSubscription struct {
 	resp *Response
 }
 
 func newPubSubSubscription(resp *Response) *PubSubSubscription {
-	return &PubSubSubscription{
+	sub := &PubSubSubscription{
 		resp: resp,
 	}
+
+	sub.Next() // skip empty element used for flushing
+	return sub
 }
 
+// Next waits for the next record and returns that.
 func (s *PubSubSubscription) Next() (*PubSubRecord, error) {
 	if s.resp.Error != nil {
 		return nil, s.resp.Error
@@ -68,186 +77,11 @@ func (s *PubSubSubscription) Next() (*PubSubRecord, error) {
 	return r, err
 }
 
+// Cancel cancels the given subscription.
 func (s *PubSubSubscription) Cancel() error {
 	if s.resp.Output == nil {
 		return nil
 	}
 
 	return s.resp.Output.Close()
 }
-
-///
-
-/*
-type subscriptionHandler struct {
-	topic string
-	resp  *Response
-
-	readers map[chan *PubSubRecord]struct{}
-
-	stop      chan struct{}
-	add, drop chan chan *PubSubRecord
-
-	failReason error
-}
-
-func newPubSubSubscriptionHandler(topic string, resp *Response) *subscriptionHandler {
-	sh := &subscriptionHandler{
-		// the topic that is being handled
-		topic: topic,
-		// stop shuts down the subscription handler.
-		stop: make(chan struct{}),
-		// readers is the set of listeners
-		readers: make(map[chan *PubSubRecord]struct{}),
-		//add is the channel in which you add more listeners
-		add: make(chan chan *PubSubRecord),
-		//drop is the channel to which you send channels
-		drop: make(chan chan *PubSubRecord),
-		resp: resp,
-	}
-
-	go sh.work()
-
-	return sh
-}
-
-func (sh *subscriptionHandler) work() {
-	readOne := func(ch chan *PubSubRecord, errCh chan error) {
-		d := json.NewDecoder(sh.resp.Output)
-		if sh.resp.Error != nil {
-			errCh <- sh.resp.Error
-			return
-		}
-
-		r := PubSubRecord{}
-		err := d.Decode(&r)
-		if err != nil {
-			errCh <- err
-			return
-		}
-
-		ch <- &r
-	}
-
-	ch := make(chan *PubSubRecord)
-	errCh := make(chan error)
-
-	go readOne(ch, errCh)
-
-L:
-	for {
-		select {
-		// remove a rdCh from pool
-		case ch := <-sh.drop:
-			delete(sh.readers, ch)
-
-			if len(sh.readers) == 0 {
-				break L
-			}
-
-		// add a rdCh to pool
-		case ch := <-sh.add:
-			sh.readers[ch] = struct{}{}
-
-		case r := <-ch:
-			for rdCh := range sh.readers {
-				rdCh <- r
-			}
-
-			go readOne(ch, errCh)
-
-		case err := <-errCh:
-			sh.failReason = err
-			break L
-
-		case <-sh.stop:
-			break L
-		}
-	}
-
-	for rdCh := range sh.readers {
-		delete(sh.readers, rdCh)
-		close(rdCh)
-	}
-
-	//sh.resp.Output.Close()
-	sh = nil
-}
-
-func (sh *subscriptionHandler) Stop() {
-	sh.stop <- struct{}{}
-}
-
-func (sh *subscriptionHandler) Sub() *PubSubSubscription {
-	ch := make(chan *PubSubRecord)
-
-	sh.add <- ch
-
-	return newPubSubSubscription(sh.topic, ch)
-}
-
-func (sh *subscriptionHandler) Drop(s *PubSubSubscription) {
-	sh.drop <- s.ch
-}
-
-func (sh *subscriptionHandler) Error() error {
-	return sh.failReason
-}
-
-///
-
-type subscriptionManager struct {
-	sync.Mutex
-
-	s    *Shell
-	subs map[string]*subscriptionHandler
-}
-
-func newPubSubSubscriptionManager(s *Shell) *subscriptionManager {
-	return &subscriptionManager{
-		s:    s,
-		subs: make(map[string]*subscriptionHandler),
-	}
-}
-
-func (sm *subscriptionManager) Sub(topic string) (*PubSubSubscription, error) {
-	// lock
-	sm.Lock()
-	defer sm.Unlock()
-
-	// check if already subscribed
-	sh := sm.subs[topic]
-	if sh == nil { // if not, do so!
-		// connect
-		req := sm.s.newRequest("pubsub/sub", topic)
-		resp, err := req.Send(sm.s.httpcli)
-		if err != nil {
-			return nil, err
-		}
-
-		// pass connection to handler and add handler to manager
-		sh = newPubSubSubscriptionHandler(topic, resp)
-		sm.subs[topic] = sh
-	}
-
-	// success
-	return sh.Sub(), nil
-}
-
-func (sm *subscriptionManager) Drop(s *PubSubSubscription) {
-	sm.Lock()
-	defer sm.Unlock()
-
-	sh := sm.subs[s.topic]
-	if sh != nil {
-		sh.Drop(s)
-	}
-}
-
-func (sm *subscriptionManager) dropHandler(sh *subscriptionHandler) {
-	sm.Lock()
-	defer sm.Unlock()
-
-	delete(sm.subs, sh.topic)
-}
-*/

shell.go

@@ -31,10 +31,7 @@ func NewShell(url string) *Shell {
 		},
 	}
 
-	s := NewShellWithClient(url, c)
-	//s.sm = newPubSubSubscriptionManager(s)
-
-	return s
+	return NewShellWithClient(url, c)
 }
 
 func NewShellWithClient(url string, c *gohttp.Client) *Shell {

shell_test.go

@@ -125,50 +125,40 @@ func TestPubSub(t *testing.T) {
 	var (
 		topic = "test"
 
-		sub   *Subscription
-		err   error
-		wait1 <-chan time.Time
-		wait  = make(chan struct{})
+		sub *PubSubSubscription
+		err error
 	)
 
-	go func() {
-		wait1 = time.After(20 * time.Millisecond) // workaround for go-ipfs#3304
-		wait <- struct{}{}
-		t.Log("subscribing...")
-		sub, err = s.PubSubSubscribe(topic)
-		is.Nil(err)
-		is.NotNil(sub)
-		t.Log("sub: done")
-
-		wait <- struct{}{}
-	}()
+	t.Log("subscribing...")
+	sub, err = s.PubSubSubscribe(topic)
+	is.Nil(err)
+	is.NotNil(sub)
+	t.Log("sub: done")
 
-	<-wait
-	<-wait1
+	time.Sleep(10 * time.Millisecond)
 
 	t.Log("publishing...")
 	is.Nil(s.PubSubPublish(topic, "Hello World!"))
 	t.Log("pub: done")
 
-	<-wait
-
 	t.Log("next()...")
 	r, err := sub.Next()
 	t.Log("next: done. ")
 
 	is.NotNil(r)
 	is.Nil(err)
-	is.Equal(r.Data, "Hello World!")
+	is.Equal(r.DataString(), "Hello World!")
 
 	sub2, err := s.PubSubSubscribe(topic)
 	is.Nil(err)
 	is.NotNil(sub2)
 
 	is.Nil(s.PubSubPublish(topic, "Hallo Welt!"))
 
-	r, err = sub2.Next() // duplicate subscription error
-	is.NotNil(err)
-	is.Nil(r)
+	r, err = sub2.Next()
+	is.Nil(err)
+	is.NotNil(r)
+	is.Equal(r.Data, "Hallo Welt!")
 
 	r, err = sub.Next()
 	is.NotNil(r)