swarm/network/stream: add SubscribeErrorMsg and UnsubscribeMsg

swarm/network/stream/messages.go

@@ -66,7 +66,17 @@ type SubscribeMsg struct {
 	Priority uint8 // delivered on priority channel
 }
 
-func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error {
+func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
+	defer func() {
+		if err != nil {
+			if e := p.Send(SubscribeErrorMsg{
+				Error: err.Error(),
+			}); e != nil {
+				log.Error("send stream subscribe error message", "err", err)
+			}
+		}
+	}()
+
 	f, err := p.streamer.GetServerFunc(req.Stream)
 	if err != nil {
 		return err
@@ -77,7 +87,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error {
 	}
 	os, err := p.setServer(req.Stream, req.Key, s, req.Priority)
 	if err != nil {
-		return nil
+		return err
 	}
 	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
 	go func() {
@@ -88,6 +98,24 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error {
 	return nil
 }
 
+type SubscribeErrorMsg struct {
+	Error string
+}
+
+func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) {
+	return fmt.Errorf("subscribe to peer %s: %v", p.ID(), req.Error)
+}
+
+type UnsubscribeMsg struct {
+	Stream string
+	Key    []byte
+}
+
+func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error {
+	p.removeServer(req.Stream, req.Key)
+	return nil
+}
+
 // OfferedHashesMsg is the protocol msg for offering to hand over a
 // stream section
 type OfferedHashesMsg struct {
@@ -247,5 +275,3 @@ func (p *Peer) handleTakeoverProofMsg(req *TakeoverProofMsg) error {
 	// store the strongest takeoverproof for the stream in streamer
 	return nil
 }
-
-type UnsubscribeMsg struct{}

swarm/network/stream/peer.go

@@ -18,6 +18,7 @@ package stream
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -30,6 +31,8 @@ import (
 
 var sendTimeout = 5 * time.Second
 
+var errServerNotFound = errors.New("server not found")
+
 // Peer is the Peer extention for the streaming protocol
 type Peer struct {
 	*protocols.Peer
@@ -145,6 +148,20 @@ func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*serve
 	return os, nil
 }
 
+func (p *Peer) removeServer(s string, key []byte) error {
+	p.serverMu.Lock()
+	defer p.serverMu.Unlock()
+
+	sk := s + keyToString(key)
+	server, ok := p.servers[sk]
+	if !ok {
+		return errServerNotFound
+	}
+	server.Close()
+	delete(p.servers, sk)
+	return nil
+}
+
 func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()

swarm/network/stream/stream.go

@@ -151,8 +151,7 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, t
 	}
 	log.Debug("Subscribe ", "peer", peerId, "stream", s, "key", t, "from", from, "to", to)
 
-	peer.SendPriority(msg, priority)
-	return nil
+	return peer.SendPriority(msg, priority)
 }
 
 func (r *Registry) Retrieve(chunk *storage.Chunk) error {
@@ -218,6 +217,12 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 	case *SubscribeMsg:
 		return p.handleSubscribeMsg(msg)
 
+	case *SubscribeErrorMsg:
+		return p.handleSubscribeErrorMsg(msg)
+
+	case *UnsubscribeMsg:
+		return p.handleUnsubscribeMsg(msg)
+
 	case *OfferedHashesMsg:
 		return p.handleOfferedHashesMsg(msg)