package whisper
import (
"testing"
"time"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
)
func startTestCluster(n int) []*Whisper {
// Create the batch of simulated peers
nodes := make([]*p2p.Peer, n)
for i := 0; i < n; i++ {
nodes[i] = p2p.NewPeer(discover.NodeID{}, "", nil)
}
whispers := make([]*Whisper, n)
for i := 0; i < n; i++ {
whispers[i] = New()
whispers[i].Start()
}
// Wire all the peers to the root one
for i := 1; i < n; i++ {
src, dst := p2p.MsgPipe()
go whispers[0].handlePeer(nodes[i], src)
go whispers[i].handlePeer(nodes[0], dst)
}
return whispers
}
func TestSelfMessage(t *testing.T) {
// Start the single node cluster
client := startTestCluster(1)[0]
// Start watching for self messages, signal any arrivals
self := client.NewIdentity()
done := make(chan struct{})
client.Watch(Filter{
To: &self.PublicKey,
Fn: func(msg *Message) {
close(done)
},
})
// Send a dummy message to oneself
msg := NewMessage([]byte("self whisper"))
envelope, err := msg.Wrap(DefaultPoW, Options{
From: self,
To: &self.PublicKey,
TTL: DefaultTTL,
})
if err != nil {
t.Fatalf("failed to wrap message: %v", err)
}
// Dump the message into the system and wait for it to pop back out
if err := client.Send(envelope); err != nil {
t.Fatalf("failed to send self-message: %v", err)
}
select {
case <-done:
case <-time.After(time.Second):
t.Fatalf("self-message receive timeout")
}
}
func TestDirectMessage(t *testing.T) {
// Start the sender-recipient cluster
cluster := startTestCluster(2)
sender := cluster[0]
senderId := sender.NewIdentity()
recipient := cluster[1]
recipientId := recipient.NewIdentity()
// Watch for arriving messages on the recipient
done := make(chan struct{})
recipient.Watch(Filter{
To: &recipientId.PublicKey,
Fn: func(msg *Message) {
close(done)
},
})
// Send a dummy message from the sender
msg := NewMessage([]byte("direct whisper"))
envelope, err := msg.Wrap(DefaultPoW, Options{
From: senderId,
To: &recipientId.PublicKey,
TTL: DefaultTTL,
})
if err != nil {
t.Fatalf("failed to wrap message: %v", err)
}
if err := sender.Send(envelope); err != nil {
t.Fatalf("failed to send direct message: %v", err)
}
// Wait for an arrival or a timeout
select {
case <-done:
case <-time.After(time.Second):
t.Fatalf("direct message receive timeout")
}
}
func TestAnonymousBroadcast(t *testing.T) {
testBroadcast(true, t)
}
func TestIdentifiedBroadcast(t *testing.T) {
testBroadcast(false, t)
}
func testBroadcast(anonymous bool, t *testing.T) {
// Start the single sender multi recipient cluster
cluster := startTestCluster(3)
sender := cluster[1]
targets := cluster[1:]
for _, target := range targets {
if !anonymous {
target.NewIdentity()
}
}
// Watch for arriving messages on the recipients
dones := make([]chan struct{}, len(targets))
for i := 0; i < len(targets); i++ {
done := make(chan struct{}) // need for the closure
dones[i] = done
targets[i].Watch(Filter{
Topics: NewTopicsFromStrings("broadcast topic"),
Fn: func(msg *Message) {
close(done)
},
})
}
// Send a dummy message from the sender
msg := NewMessage([]byte("broadcast whisper"))
envelope, err := msg.Wrap(DefaultPoW, Options{
Topics: NewTopicsFromStrings("broadcast topic"),
TTL: DefaultTTL,
})
if err != nil {
t.Fatalf("failed to wrap message: %v", err)
}
if err := sender.Send(envelope); err != nil {
t.Fatalf("failed to send broadcast message: %v", err)
}
// Wait for an arrival on each recipient, or timeouts
timeout := time.After(time.Second)
for _, done := range dones {
select {
case <-done:
case <-timeout:
t.Fatalf("broadcast message receive timeout")
}
}
}
func TestMessageExpiration(t *testing.T) {
// Start the single node cluster and inject a dummy message
node := startTestCluster(1)[0]
message := NewMessage([]byte("expiring message"))
envelope, err := message.Wrap(DefaultPoW, Options{
TTL: time.Second,
})
if err != nil {
t.Fatalf("failed to wrap message: %v", err)
}
if err := node.Send(envelope); err != nil {
t.Fatalf("failed to inject message: %v", err)
}
// Check that the message is inside the cache
if _, ok := node.messages[envelope.Hash()]; !ok {
t.Fatalf("message not found in cache")
}
// Wait for expiration and check cache again
time.Sleep(time.Second) // wait for expiration
time.Sleep(expirationCycle) // wait for cleanup cycle
if _, ok := node.messages[envelope.Hash()]; ok {
t.Fatalf("message not expired from cache")
}
}
