pubsub: remove subscription management; naive implementation

Author: keks

pubsub.go

@@ -3,7 +3,7 @@ package shell
 import (
 	"encoding/base64"
 	"encoding/json"
-	"sync"
+	//	"sync"
 )
 
 type b64String string
@@ -42,27 +42,35 @@ type PubSubRecord struct {
 ///
 
 type Subscription struct {
-	topic string
-	ch    chan *PubSubRecord
+	resp *Response
 }
 
-func newSubscription(topic string, ch chan *PubSubRecord) *Subscription {
+func newSubscription(resp *Response) *Subscription {
 	return &Subscription{
-		topic: topic,
-		ch:    ch,
+		resp: resp,
 	}
 }
 
-func (s *Subscription) Next() *PubSubRecord {
-	return <-s.ch
+func (s *Subscription) Next() (*PubSubRecord, error) {
+	if s.resp.Error != nil {
+		return nil, s.resp.Error
+	}
+
+	d := json.NewDecoder(s.resp.Output)
+
+	r := &PubSubRecord{}
+	err := d.Decode(r)
+
+	return r, err
 }
 
-func (s *Subscription) Topic() string {
-	return s.topic
+func (s *Subscription) Cancel() error {
+	return s.resp.Output.Close()
 }
 
 ///
 
+/*
 type subscriptionHandler struct {
 	topic string
 	resp  *Response
@@ -154,7 +162,7 @@ L:
 		close(rdCh)
 	}
 
-	sh.resp.Output.Close()
+	//sh.resp.Output.Close()
 	sh = nil
 }
 
@@ -234,3 +242,4 @@ func (sm *subscriptionManager) dropHandler(sh *subscriptionHandler) {
 
 	delete(sm.subs, sh.topic)
 }
+*/

shell.go

@@ -23,7 +23,7 @@ type Shell struct {
 	url     string
 	httpcli *gohttp.Client
 
-	sm *subscriptionManager
+	//sm *subscriptionManager
 }
 
 func NewShell(url string) *Shell {
@@ -34,7 +34,7 @@ func NewShell(url string) *Shell {
 	}
 
 	s := NewShellWithClient(url, c)
-	s.sm = newSubscriptionManager(s)
+	//s.sm = newSubscriptionManager(s)
 
 	return s
 }
@@ -695,11 +695,15 @@ func (s *Shell) ObjectPut(obj *IpfsObject) (string, error) {
 }
 
 func (s *Shell) PubSubSubscribe(topic string) (*Subscription, error) {
-	return s.sm.Sub(topic)
-}
+	// connect
+	req := s.newRequest("pubsub/sub", topic)
+
+	resp, err := req.Send(s.httpcli)
+	if err != nil {
+		return nil, err
+	}
 
-func (s *Shell) PubSubCancelSubscription(sub *Subscription) {
-	s.sm.Drop(sub)
+	return newSubscription(resp), nil
 }
 
 func (s *Shell) PubSubPublish(topic, data string) error {

shell_test.go

@@ -123,18 +123,21 @@ func TestPubSub(t *testing.T) {
 	s := NewShell(shellUrl)
 
 	var (
+		topic = "test"
+
 		sub   *Subscription
 		err   error
 		wait1 <-chan time.Time
 		wait  = make(chan struct{})
 	)
 
 	go func() {
-		wait1 = time.After(time.Second)
+		wait1 = time.After(20 * time.Millisecond) // workaround for go-ipfs#3304
 		wait <- struct{}{}
 		t.Log("subscribing...")
-		sub, err = s.PubSubSubscribe("test")
+		sub, err = s.PubSubSubscribe(topic)
 		is.Nil(err)
+		is.NotNil(sub)
 		t.Log("sub: done")
 
 		wait <- struct{}{}
@@ -144,15 +147,33 @@ func TestPubSub(t *testing.T) {
 	<-wait1
 
 	t.Log("publishing...")
-	is.Nil(s.PubSubPublish("test", "Hello World!"))
+	is.Nil(s.PubSubPublish(topic, "Hello World!"))
 	t.Log("pub: done")
 
 	<-wait
 
 	t.Log("next()...")
-	r := sub.Next()
+	r, err := sub.Next()
 	t.Log("next: done. ")
 
 	is.NotNil(r)
+	is.Nil(err)
 	is.Equal(r.Data, "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 = sub.Next()
+	is.NotNil(r)
+	is.Nil(err)
+	is.Equal(r.Data, "Hallo Welt!")
+
+	is.Nil(sub.Cancel())
 }