[#1800] Internalize key prefix handling to ipfs package for PubSub

The consumers of the Publisher/Subscriber should not need to worry about
the key prefix of the topics they are interested in. Made an internal
concern.

Author: placer14

core/net.go

@@ -122,7 +122,7 @@ func (n *OpenBazaarNode) SendOfflineMessage(p peer.ID, k *libp2p.PubKey, m *pb.M
 	go func() {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
-		err := n.Pubsub.Publisher.Publish(ctx, ipfs.MessageTopicPrefix+pointer.Cid.String(), ciphertext)
+		err := n.Pubsub.Publisher.Publish(ctx, pointer.Cid.String(), ciphertext)
 		if err != nil {
 			log.Error(err)
 		}

ipfs/pubsub.go

@@ -17,10 +17,10 @@ import (
 )
 
 const (
-	MessageTopicPrefix = "/offlinemessage/"
-	GlobalIPNSTopic    = "IPNS"
-	GlobalBlockTopic   = "BLOCK"
-	GlobalCIDTopic     = "CID"
+	globalIPNSTopic    = "IPNS"
+	globalBlockTopic   = "BLOCK"
+	globalCIDTopic     = "CID"
+	messageTopicPrefix = "/offlinemessage/"
 )
 
 type Pubsub struct {
@@ -79,40 +79,42 @@ func NewPubsubSubscriber(ctx context.Context, host p2phost.Host, cr routing.Cont
 
 func (p *PubsubPublisher) Publish(ctx context.Context, topic string, data []byte) error {
 	p.mx.Lock()
-	_, ok := p.subs[topic]
+	id := messageTopicPrefix + topic
+	_, ok := p.subs[id]
 
 	if !ok {
-		p.subs[topic] = struct{}{}
+		p.subs[id] = struct{}{}
 		p.mx.Unlock()
 
-		bootstrapPubsub(p.ctx, p.cr, p.host, topic)
+		bootstrapPubsub(p.ctx, p.cr, p.host, id)
 	} else {
 		p.mx.Unlock()
 	}
 
-	log.Debugf("PubsubPublish: publish data for %s", topic)
-	return p.ps.Publish(topic, data)
+	log.Debugf("PubsubPublish: publish data for %s", id)
+	return p.ps.Publish(id, data)
 }
 
 func (r *PubsubSubscriber) Subscribe(ctx context.Context, topic string) (chan []byte, error) {
 	r.mx.Lock()
 	// see if we already have a pubsub subscription; if not, subscribe
-	_, ok := r.subs[topic]
+	id := messageTopicPrefix + topic
+	_, ok := r.subs[id]
 	resp := make(chan []byte)
 	if !ok {
-		sub, err := r.ps.Subscribe(topic)
+		sub, err := r.ps.Subscribe(id)
 		if err != nil {
 			r.mx.Unlock()
 			return nil, err
 		}
 
-		log.Debugf("PubsubSubscribe: subscribed to %s", topic)
+		log.Debugf("PubsubSubscribe: subscribed to %s", id)
 
-		r.subs[topic] = sub
+		r.subs[id] = sub
 
 		ctx, cancel := context.WithCancel(r.ctx)
-		go r.handleSubscription(sub, topic, resp, cancel)
-		go bootstrapPubsub(ctx, r.cr, r.host, topic)
+		go r.handleSubscription(sub, id, resp, cancel)
+		go bootstrapPubsub(ctx, r.cr, r.host, id)
 	}
 	r.mx.Unlock()
 	return resp, nil