whisper: activate write deadline

This imposes to cleanup peers from the server
as the deadline occurs. As a result, I decided
to make the `update` function dependent
on the type of peer; the libp2p version
will try to remove the peers as itreceives
the
shutdown signal from a peer.

Note that the read deadlines triggers no
matter what, and so wnode stops working.

whisper/whisperv6/devp2p_glue.go

@@ -17,9 +17,85 @@
 package whisperv6
 
 import (
+	"fmt"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
+	set "gopkg.in/fatih/set.v0"
+	"time"
 )
 
+// DevP2PPeer is the DevP2P implementation of the Peer interface
+type DevP2PPeer struct {
+	*PeerBase
+
+	peer *p2p.Peer
+}
+
+// newPeer creates a new whisper peer object, but does not run the handshake itself.
+func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer {
+	return &DevP2PPeer{
+		&PeerBase{
+			host:           host,
+			ws:             rw,
+			trusted:        false,
+			powRequirement: 0.0,
+			known:          set.New(),
+			quit:           make(chan struct{}),
+			bloomFilter:    makeFullNodeBloom(),
+			fullNode:       true,
+		},
+		remote,
+	}
+}
+
+// handshake sends the protocol initiation status message to the remote peer and
+// verifies the remote status too.
+func (peer *DevP2PPeer) handshake() error {
+	err := peer.handshakeBase()
+	if err != nil {
+		return fmt.Errorf("peer [%x] %s", peer.ID(), err.Error())
+	}
+	return nil
+}
+
+// start initiates the peer updater, periodically broadcasting the whisper packets
+// into the network.
+func (peer *DevP2PPeer) start() {
+	go peer.update()
+	log.Trace("start", "peer", peer.ID())
+}
+
+// stop terminates the peer updater, stopping message forwarding to it.
+func (peer *DevP2PPeer) stop() {
+	close(peer.quit)
+	log.Trace("stop", "peer", peer.ID())
+}
+
+// update executes periodic operations on the peer, including message transmission
+// and expiration.
+func (peer *DevP2PPeer) update() {
+	// Start the tickers for the updates
+	expire := time.NewTicker(expirationCycle)
+	transmit := time.NewTicker(transmissionCycle)
+
+	// Loop and transmit until termination is requested
+	for {
+		select {
+		case <-expire.C:
+			peer.expire()
+
+		case <-transmit.C:
+			if err := peer.broadcast(); err != nil {
+				log.Trace("broadcast failed", "reason", err)
+				return
+			}
+
+		case <-peer.quit:
+			return
+		}
+	}
+}
+
 // DevP2PWhisperServer implements WhisperServer with a DevP2P backend
 type DevP2PWhisperServer struct {
 	Server *p2p.Server

whisper/whisperv6/libp2p_glue.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"github.com/ethereum/go-ethereum/log"
 	"io"
+	"time"
 
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -77,8 +78,6 @@ func (stream *LibP2PStream) ReadMsg() (p2p.Msg, error) {
 // WriteMsg implements the MsgReadWriter interface to write messages
 // to lilbp2p streams.
 func (stream *LibP2PStream) WriteMsg(msg p2p.Msg) error {
-
-
 	stream.lp2pStream.SetWriteDeadline(time.Now().Add(transmissionCycle))
 
 	if err := rlp.Encode(stream.lp2pStream, msg.Code); err != nil {
@@ -94,10 +93,12 @@ type LibP2PPeer struct {
 
 	id peer.ID
 
+	server *LibP2PWhisperServer
+
 	connectionStream *LibP2PStream
 }
 
-func newLibP2PPeer(w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer {
+func newLibP2PPeer(s *LibP2PWhisperServer, w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer {
 	return &LibP2PPeer{
 		&PeerBase{
 			host:           w,
@@ -110,6 +111,7 @@ func newLibP2PPeer(w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer {
 			fullNode:       true,
 		},
 		pid,
+		s,
 		nil,
 	}
 }
@@ -127,6 +129,52 @@ func (p *LibP2PPeer) handshake() error {
 	return nil
 }
 
+// start initiates the peer updater, periodically broadcasting the whisper packets
+// into the network.
+func (p *LibP2PPeer) start() {
+	go p.update()
+	log.Trace("start", "peer", p.ID())
+}
+
+// stop terminates the peer updater, stopping message forwarding to it.
+func (p *LibP2PPeer) stop() {
+	close(p.quit)
+	fmt.Println("stop", "peer", p.ID())
+}
+
+// update executes periodic operations on the peer, including message transmission
+// and expiration.
+func (p *LibP2PPeer) update() {
+	// Start the tickers for the updates
+	expire := time.NewTicker(expirationCycle)
+	transmit := time.NewTicker(transmissionCycle)
+
+	// Loop and transmit until termination is requested
+updateLoop:
+	for {
+		select {
+		case <-expire.C:
+			p.expire()
+
+		case <-transmit.C:
+			if err := p.broadcast(); err != nil {
+				break updateLoop
+			}
+
+		case <-p.quit:
+			break updateLoop
+		}
+	}
+
+	// Cleanup and remove the peer from the list
+	for i, it := range p.server.Peers {
+		if it.id == p.id {
+			p.server.Peers = append(p.server.Peers[:i], p.server.Peers[i+1:]...)
+			break
+		}
+	}
+}
+
 // LibP2PWhisperServer implements WhisperServer for libp2p.
 type LibP2PWhisperServer struct {
 	Host host.Host
@@ -177,7 +225,7 @@ func (server *LibP2PWhisperServer) Start() error {
 
 		// Unknown peer
 		if peer == nil {
-			peer = newLibP2PPeer(server.whisper, pid, lps)
+			peer = newLibP2PPeer(server, server.whisper, pid, lps)
 			// TODO check critical section
 			server.Peers = append(server.Peers, peer.(*LibP2PPeer))
 		}
@@ -236,7 +284,7 @@ func (server *LibP2PWhisperServer) AddPeer(addr ma.Multiaddr) *LibP2PPeer {
 	ipfsaddrpart, _ := ma.NewMultiaddr(fmt.Sprintf("/ipfs/%s", pid))
 	ipaddr := addr.Decapsulate(ipfsaddrpart)
 	server.Host.Peerstore().AddAddr(peerid, ipaddr, pstore.PermanentAddrTTL)
-	newPeer := newLibP2PPeer(server.whisper, peerid, nil).(*LibP2PPeer)
+	newPeer := newLibP2PPeer(server, server.whisper, peerid, nil).(*LibP2PPeer)
 	server.Peers = append(server.Peers, newPeer)
 
 	return newPeer

whisper/whisperv6/peer.go

@@ -52,7 +52,6 @@ type Peer interface {
 	start()
 	stop()
 	handshake() error
-	update()
 	mark(*Envelope)
 	marked(*Envelope) bool
 	expire()
@@ -65,55 +64,6 @@ type Peer interface {
 	setTrusted(bool)
 	setPoWRequirement(float64)
 	stream() p2p.MsgReadWriter
-
-	// newPeer(*Whisper, p2p.MsgReadWriter) Peer
-}
-
-// DevP2PPeer is the DevP2P implementation of the Peer interface
-type DevP2PPeer struct {
-	*PeerBase
-
-	peer *p2p.Peer
-}
-
-// newPeer creates a new whisper peer object, but does not run the handshake itself.
-func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer {
-	return &DevP2PPeer{
-		&PeerBase{
-			host:           host,
-			ws:             rw,
-			trusted:        false,
-			powRequirement: 0.0,
-			known:          set.New(),
-			quit:           make(chan struct{}),
-			bloomFilter:    MakeFullNodeBloom(),
-			fullNode:       true,
-		},
-		remote,
-	}
-}
-
-// start initiates the peer updater, periodically broadcasting the whisper packets
-// into the network.
-func (peer *PeerBase) start() {
-	go peer.update()
-	// log.Trace("start", "peer", peer.ID())
-}
-
-// stop terminates the peer updater, stopping message forwarding to it.
-func (peer *PeerBase) stop() {
-	close(peer.quit)
-	// log.Trace("stop", "peer", peer.ID())
-}
-
-// handshake sends the protocol initiation status message to the remote peer and
-// verifies the remote status too.
-func (peer *DevP2PPeer) handshake() error {
-	err := peer.handshakeBase()
-	if err != nil {
-		return fmt.Errorf("peer [%x] %s", peer.ID(), err.Error())
-	}
-	return nil
 }
 
 func (peer *PeerBase) handshakeBase() error {
@@ -173,31 +123,6 @@ func (peer *PeerBase) handshakeBase() error {
 	return nil
 }
 
-// update executes periodic operations on the peer, including message transmission
-// and expiration.
-func (peer *DevP2PPeer) update() {
-	// Start the tickers for the updates
-	expire := time.NewTicker(expirationCycle)
-	transmit := time.NewTicker(transmissionCycle)
-
-	// Loop and transmit until termination is requested
-	for {
-		select {
-		case <-expire.C:
-			peer.expire()
-
-		case <-transmit.C:
-			if err := peer.broadcast(); err != nil {
-				log.Trace("broadcast failed", "reason", err, "peer", peer.ID())
-				return
-			}
-
-		case <-peer.quit:
-			return
-		}
-	}
-}
-
 // mark marks an envelope known to the peer so that it won't be sent back.
 func (peer *PeerBase) mark(envelope *Envelope) {
 	peer.known.Add(envelope.Hash())