whisper: process incoming packets

whisper/whisperv6/libp2p_glue.go

@@ -30,6 +30,7 @@ import (
 	inet "github.com/libp2p/go-libp2p-net"
 	peer "github.com/libp2p/go-libp2p-peer"
 	ma "github.com/multiformats/go-multiaddr"
+	set "gopkg.in/fatih/set.v0"
 )
 
 // LibP2PStream is a wrapper used to implement the MsgReadWriter
@@ -116,25 +117,73 @@ func (stream *LibP2PStream) WriteMsg(msg p2p.Msg) error {
 
 // LibP2PPeer implements Peer for libp2p
 type LibP2PPeer struct {
-	PeerBase
+	*PeerBase
 
 	id peer.ID
 }
 
+func newLibP2PPeer(w *Whisper, pid peer.ID, rw p2p.MsgReadWriter) Peer {
+	return &LibP2PPeer{
+		&PeerBase{
+			host:           w,
+			ws:             rw,
+			trusted:        false,
+			powRequirement: 0.0,
+			known:          set.New(),
+			quit:           make(chan struct{}),
+			bloomFilter:    makeFullNodeBloom(),
+			fullNode:       true,
+		},
+		pid,
+	}
+}
+
 // ID returns the id of the peer
 func (p *LibP2PPeer) ID() string {
 	return p.id.String()
 }
 
+func (p *LibP2PPeer) handshake() error {
+	err := p.handshakeBase()
+	if err != nil {
+		return fmt.Errorf("peer [%x] %s", p.ID(), err.Error())
+	}
+	return nil
+}
+
 // LibP2PWhisperServer implements WhisperServer for libp2p.
 type LibP2PWhisperServer struct {
 	Host host.Host
 
-	Peers []LibP2PPeer
+	Peers []*LibP2PPeer
 }
 
 // Start starts the server
 func (server *LibP2PWhisperServer) Start() error {
+	server.Host.SetStreamHandler(WhisperProtocolString, func (stream inet.Stream) {
+		defer stream.Close()
+	
+		pid := stream.Conn().RemotePeer()
+		var peer Peer
+		for _, p := range server.Peers {
+			if p.id == pid {
+				peer = p
+				break
+			}
+		}
+
+		whisper := server.Peers[0].host
+		lps := &LibP2PStream{stream}
+
+		// Unknown peer
+		if peer == nil {
+			peer = newLibP2PPeer(whisper, pid, lps)
+			// TODO check critical section
+			server.Peers = append(server.Peers, peer.(*LibP2PPeer))
+		}
+
+		whisper.runMessageLoop(peer, lps)
+	})
 	return nil
 }
 
@@ -179,5 +228,7 @@ func NewLibP2PWhisperServer() (WhisperServer, error) {
 	if err != nil {
 		return nil, fmt.Errorf("Error setting up the libp2p network: %s", err)
 	}
-	return &LibP2PWhisperServer{h, []LibP2PPeer{}}, nil
+
+	server := &LibP2PWhisperServer{h, []*LibP2PPeer{}}
+	return server, nil
 }

whisper/whisperv6/peer.go

@@ -95,20 +95,28 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer {
 
 // start initiates the peer updater, periodically broadcasting the whisper packets
 // into the network.
-func (peer *DevP2PPeer) start() {
+func (peer *PeerBase) start() {
 	go peer.update()
-	log.Trace("start", "peer", peer.ID())
+	// log.Trace("start", "peer", peer.ID())
 }
 
 // stop terminates the peer updater, stopping message forwarding to it.
-func (peer *DevP2PPeer) stop() {
+func (peer *PeerBase) stop() {
 	close(peer.quit)
-	log.Trace("stop", "peer", peer.ID())
+	// 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 {
 	// Send the handshake status message asynchronously
 	errc := make(chan error, 1)
 	go func() {
@@ -124,19 +132,19 @@ func (peer *DevP2PPeer) handshake() error {
 		return err
 	}
 	if packet.Code != statusCode {
-		return fmt.Errorf("peer [%x] sent packet %x before status packet", peer.ID(), packet.Code)
+		return fmt.Errorf("peer [%s] sent packet %x before status packet", peer.ID(), packet.Code)
 	}
 	s := rlp.NewStream(packet.Payload, uint64(packet.Size))
 	_, err = s.List()
 	if err != nil {
-		return fmt.Errorf("peer [%x] sent bad status message: %v", peer.ID(), err)
+		return fmt.Errorf("peer [%s] sent bad status message: %v", peer.ID(), err)
 	}
 	peerVersion, err := s.Uint()
 	if err != nil {
-		return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", peer.ID(), err)
+		return fmt.Errorf("peer [%s] sent bad status message (unable to decode version): %v", peer.ID(), err)
 	}
 	if peerVersion != ProtocolVersion {
-		return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", peer.ID(), peerVersion, ProtocolVersion)
+		return fmt.Errorf("peer [%s]: protocol version mismatch %d != %d", peer.ID(), peerVersion, ProtocolVersion)
 	}
 
 	// only version is mandatory, subsequent parameters are optional
@@ -144,7 +152,7 @@ func (peer *DevP2PPeer) handshake() error {
 	if err == nil {
 		pow := math.Float64frombits(powRaw)
 		if math.IsInf(pow, 0) || math.IsNaN(pow) || pow < 0.0 {
-			return fmt.Errorf("peer [%x] sent bad status message: invalid pow", peer.ID())
+			return fmt.Errorf("peer [%s] sent bad status message: invalid pow", peer.ID())
 		}
 		peer.powRequirement = pow
 
@@ -153,14 +161,14 @@ func (peer *DevP2PPeer) handshake() error {
 		if err == nil {
 			sz := len(bloom)
 			if sz != BloomFilterSize && sz != 0 {
-				return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", peer.ID(), sz)
+				return fmt.Errorf("peer [%s] sent bad status message: wrong bloom filter size %d", peer.ID(), sz)
 			}
 			peer.setBloomFilter(bloom)
 		}
 	}
 
 	if err := <-errc; err != nil {
-		return fmt.Errorf("peer [%x] failed to send status packet: %v", peer.ID(), err)
+		return fmt.Errorf("peer [%s] failed to send status packet: %v", peer.ID(), err)
 	}
 	return nil
 }
@@ -191,18 +199,18 @@ func (peer *DevP2PPeer) update() {
 }
 
 // mark marks an envelope known to the peer so that it won't be sent back.
-func (peer *DevP2PPeer) mark(envelope *Envelope) {
+func (peer *PeerBase) mark(envelope *Envelope) {
 	peer.known.Add(envelope.Hash())
 }
 
 // marked checks if an envelope is already known to the remote peer.
-func (peer *DevP2PPeer) marked(envelope *Envelope) bool {
+func (peer *PeerBase) marked(envelope *Envelope) bool {
 	return peer.known.Has(envelope.Hash())
 }
 
 // expire iterates over all the known envelopes in the host and removes all
 // expired (unknown) ones from the known list.
-func (peer *DevP2PPeer) expire() {
+func (peer *PeerBase) expire() {
 	unmark := make(map[common.Hash]struct{})
 	peer.known.Each(func(v interface{}) bool {
 		if !peer.host.isEnvelopeCached(v.(common.Hash)) {
@@ -218,7 +226,7 @@ func (peer *DevP2PPeer) expire() {
 
 // broadcast iterates over the collection of envelopes and transmits yet unknown
 // ones over the network.
-func (peer *DevP2PPeer) broadcast() error {
+func (peer *PeerBase) broadcast() error {
 	envelopes := peer.host.Envelopes()
 	bundle := make([]*Envelope, 0, len(envelopes))
 	for _, envelope := range envelopes {
@@ -248,22 +256,22 @@ func (peer *DevP2PPeer) ID() string {
 	return peer.peer.ID().String()
 }
 
-func (peer *DevP2PPeer) notifyAboutPowRequirementChange(pow float64) error {
+func (peer *PeerBase) notifyAboutPowRequirementChange(pow float64) error {
 	i := math.Float64bits(pow)
 	return p2p.Send(peer.ws, powRequirementCode, i)
 }
 
-func (peer *DevP2PPeer) notifyAboutBloomFilterChange(bloom []byte) error {
+func (peer *PeerBase) notifyAboutBloomFilterChange(bloom []byte) error {
 	return p2p.Send(peer.ws, bloomFilterExCode, bloom)
 }
 
-func (peer *DevP2PPeer) bloomMatch(env *Envelope) bool {
+func (peer *PeerBase) bloomMatch(env *Envelope) bool {
 	peer.bloomMu.Lock()
 	defer peer.bloomMu.Unlock()
 	return peer.fullNode || BloomFilterMatch(peer.bloomFilter, env.Bloom())
 }
 
-func (peer *DevP2PPeer) setBloomFilter(bloom []byte) {
+func (peer *PeerBase) setBloomFilter(bloom []byte) {
 	peer.bloomMu.Lock()
 	defer peer.bloomMu.Unlock()
 	peer.bloomFilter = bloom
@@ -273,19 +281,19 @@ func (peer *DevP2PPeer) setBloomFilter(bloom []byte) {
 	}
 }
 
-func (peer *DevP2PPeer) isTrusted() bool {
+func (peer *PeerBase) isTrusted() bool {
 	return peer.trusted
 }
 
-func (peer *DevP2PPeer) setTrusted(t bool) {
+func (peer *PeerBase) setTrusted(t bool) {
 	peer.trusted = t
 }
 
-func (peer *DevP2PPeer) setPoWRequirement(r float64) {
+func (peer *PeerBase) setPoWRequirement(r float64) {
 	peer.powRequirement = r
 }
 
-func (peer *DevP2PPeer) stream() p2p.MsgReadWriter {
+func (peer *PeerBase) stream() p2p.MsgReadWriter {
 	return peer.ws
 }