swarm/pss: Rebase after deduplication changes

swarm/pss/protocol_test.go

@@ -22,9 +22,9 @@ type protoCtrl struct {
 
 // simple ping pong protocol test for the pss devp2p emulation
 func TestProtocol(t *testing.T) {
-	//t.Run("32", testProtocol)
+	t.Run("32", testProtocol)
 	t.Run("8", testProtocol)
-	//t.Run("0", testProtocol)
+	t.Run("0", testProtocol)
 }
 
 func testProtocol(t *testing.T) {

swarm/pss/pss.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"crypto/ecdsa"
 	"crypto/rand"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -115,7 +116,7 @@ type Pss struct {
 
 	// message handling
 	handlers   map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle()
-	handlersMu sync.Mutex
+	handlersMu sync.RWMutex
 	allowRaw   bool
 	hashPool   sync.Pool
 
@@ -615,7 +616,8 @@ func (self *Pss) SendRaw(msg []byte, address PssAddress) error {
 		},
 	}
 	self.addFwdCache(pssmsg)
-	return self.enqueue(pssmsg)
+	self.outbox <- pssmsg
+	return nil
 }
 
 // Send a message using symmetric encryption
@@ -709,6 +711,7 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [
 		Payload: envelope,
 	}
 	self.outbox <- pssmsg
+	return nil
 }
 
 // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct

swarm/pss/pss_test.go

@@ -792,11 +792,11 @@ func testNetwork(t *testing.T) {
 		}
 		a = adapters.NewExecAdapter(dirname)
 	} else if adapter == "sock" {
-		a = adapters.NewSocketAdapter(newServices())
+		a = adapters.NewSocketAdapter(newServices(false))
 	} else if adapter == "tcp" {
-		a = adapters.NewTCPAdapter(newServices())
+		a = adapters.NewTCPAdapter(newServices(false))
 	} else if adapter == "sim" {
-		a = adapters.NewSimAdapter(newServices())
+		a = adapters.NewSimAdapter(newServices(false))
 	}
 	net := simulations.NewNetwork(a, &simulations.NetworkConfig{
 		ID: "0",