fix(bs/notifications) use SubOnceEach to provide uniqueness guarantee

License: MIT
Signed-off-by: Brian Tiger Chow <brian@perfmode.com>

vendor forked pubsub to get SubOnceEach

License: MIT
Signed-off-by: Brian Tiger Chow <brian@perfmode.com>

notifications/notifications.go

@@ -2,7 +2,7 @@ package notifications
 
 import (
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
-	pubsub "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/tuxychandru/pubsub"
+	pubsub "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/maybebtc/pubsub"
 
 	blocks "github.com/jbenet/go-ipfs/blocks"
 	u "github.com/jbenet/go-ipfs/util"
@@ -39,20 +39,16 @@ func (ps *impl) Shutdown() {
 func (ps *impl) Subscribe(ctx context.Context, keys ...u.Key) <-chan *blocks.Block {
 
 	blocksCh := make(chan *blocks.Block, len(keys))
-	valuesCh := make(chan interface{}, len(keys))
-	ps.wrapped.AddSub(valuesCh, toStrings(keys)...)
-
+	if len(keys) == 0 {
+		close(blocksCh)
+		return blocksCh
+	}
+	valuesCh := ps.wrapped.SubOnceEach(toStrings(keys)...)
 	go func() {
 		defer func() {
-			ps.wrapped.Unsub(valuesCh, toStrings(keys)...)
 			close(blocksCh)
 		}()
-		seen := make(map[u.Key]struct{})
-		i := 0 // req'd because it only counts unique block sends
 		for {
-			if i >= len(keys) {
-				return
-			}
 			select {
 			case <-ctx.Done():
 				return
@@ -64,22 +60,10 @@ func (ps *impl) Subscribe(ctx context.Context, keys ...u.Key) <-chan *blocks.Blo
 				if !ok {
 					return
 				}
-				if _, ok := seen[block.Key()]; ok {
-					continue
-				}
 				select {
 				case <-ctx.Done():
 					return
 				case blocksCh <- block: // continue
-					// Unsub alone is insufficient for keeping out duplicates.
-					// It's a race to unsubscribe before pubsub handles the
-					// next Publish call. Therefore, must also check for
-					// duplicates manually. Unsub is a performance
-					// consideration to avoid lots of unnecessary channel
-					// chatter.
-					ps.wrapped.Unsub(valuesCh, string(block.Key()))
-					i++
-					seen[block.Key()] = struct{}{}
 				}
 			}
 		}