mirror of
https://github.com/logos-messaging/go-libp2p-pubsub.git
synced 2026-07-31 18:23:13 +00:00
basic gossipsub tests
This commit is contained in:
vyzo
Steven Allen
parent
1f5959bf54
commit
0a82522cf6
@ -55,8 +55,16 @@ func connect(t *testing.T, a, b host.Host) {
|
||||
}
|
||||
|
||||
func sparseConnect(t *testing.T, hosts []host.Host) {
|
||||
connectSome(t, hosts, 3)
|
||||
}
|
||||
|
||||
func denseConnect(t *testing.T, hosts []host.Host) {
|
||||
connectSome(t, hosts, 10)
|
||||
}
|
||||
|
||||
func connectSome(t *testing.T, hosts []host.Host, d int) {
|
||||
for i, a := range hosts {
|
||||
for j := 0; j < 3; j++ {
|
||||
for j := 0; j < d; j++ {
|
||||
n := rand.Intn(len(hosts))
|
||||
if n == i {
|
||||
j--
|
||||
|
||||
239
gossipsub_test.go
Normal file
239
gossipsub_test.go
Normal file
@ -0,0 +1,239 @@
|
||||
package floodsub
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
host "github.com/libp2p/go-libp2p-host"
|
||||
)
|
||||
|
||||
func getGossipsubs(ctx context.Context, hs []host.Host, opts ...Option) []*PubSub {
|
||||
var psubs []*PubSub
|
||||
for _, h := range hs {
|
||||
ps, err := NewGossipSub(ctx, h, opts...)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
psubs = append(psubs, ps)
|
||||
}
|
||||
return psubs
|
||||
}
|
||||
|
||||
func TestSparseGossipsub(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
hosts := getNetHosts(t, ctx, 20)
|
||||
|
||||
psubs := getGossipsubs(ctx, hosts)
|
||||
|
||||
var msgs []*Subscription
|
||||
for _, ps := range psubs {
|
||||
subch, err := ps.Subscribe("foobar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
msgs = append(msgs, subch)
|
||||
}
|
||||
|
||||
sparseConnect(t, hosts)
|
||||
|
||||
// wait for heartbeats to build mesh
|
||||
time.Sleep(time.Second * 2)
|
||||
|
||||
for i := 0; i < 100; i++ {
|
||||
msg := []byte(fmt.Sprintf("%d it's not a floooooood %d", i, i))
|
||||
|
||||
owner := rand.Intn(len(psubs))
|
||||
|
||||
psubs[owner].Publish("foobar", msg)
|
||||
|
||||
for _, sub := range msgs {
|
||||
got, err := sub.Next(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(sub.err)
|
||||
}
|
||||
if !bytes.Equal(msg, got.Data) {
|
||||
t.Fatal("got wrong message!")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDenseGossipsub(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
hosts := getNetHosts(t, ctx, 20)
|
||||
|
||||
psubs := getGossipsubs(ctx, hosts)
|
||||
|
||||
var msgs []*Subscription
|
||||
for _, ps := range psubs {
|
||||
subch, err := ps.Subscribe("foobar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
msgs = append(msgs, subch)
|
||||
}
|
||||
|
||||
denseConnect(t, hosts)
|
||||
|
||||
// wait for heartbeats to build mesh
|
||||
time.Sleep(time.Second * 2)
|
||||
|
||||
for i := 0; i < 100; i++ {
|
||||
msg := []byte(fmt.Sprintf("%d it's not a floooooood %d", i, i))
|
||||
|
||||
owner := rand.Intn(len(psubs))
|
||||
|
||||
psubs[owner].Publish("foobar", msg)
|
||||
|
||||
for _, sub := range msgs {
|
||||
got, err := sub.Next(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(sub.err)
|
||||
}
|
||||
if !bytes.Equal(msg, got.Data) {
|
||||
t.Fatal("got wrong message!")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMixedGossipsub(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
hosts := getNetHosts(t, ctx, 30)
|
||||
|
||||
gsubs := getGossipsubs(ctx, hosts[:20])
|
||||
fsubs := getPubsubs(ctx, hosts[20:])
|
||||
psubs := append(gsubs, fsubs...)
|
||||
|
||||
var msgs []*Subscription
|
||||
for _, ps := range psubs {
|
||||
subch, err := ps.Subscribe("foobar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
msgs = append(msgs, subch)
|
||||
}
|
||||
|
||||
sparseConnect(t, hosts)
|
||||
|
||||
// wait for heartbeats to build mesh
|
||||
time.Sleep(time.Second * 2)
|
||||
|
||||
for i := 0; i < 100; i++ {
|
||||
msg := []byte(fmt.Sprintf("%d it's not a floooooood %d", i, i))
|
||||
|
||||
owner := rand.Intn(len(psubs))
|
||||
|
||||
psubs[owner].Publish("foobar", msg)
|
||||
|
||||
for _, sub := range msgs {
|
||||
got, err := sub.Next(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(sub.err)
|
||||
}
|
||||
if !bytes.Equal(msg, got.Data) {
|
||||
t.Fatal("got wrong message!")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGossipsubMultihops(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
hosts := getNetHosts(t, ctx, 6)
|
||||
|
||||
psubs := getGossipsubs(ctx, hosts)
|
||||
|
||||
connect(t, hosts[0], hosts[1])
|
||||
connect(t, hosts[1], hosts[2])
|
||||
connect(t, hosts[2], hosts[3])
|
||||
connect(t, hosts[3], hosts[4])
|
||||
connect(t, hosts[4], hosts[5])
|
||||
|
||||
var subs []*Subscription
|
||||
for i := 1; i < 6; i++ {
|
||||
ch, err := psubs[i].Subscribe("foobar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
subs = append(subs, ch)
|
||||
}
|
||||
|
||||
// wait for heartbeats to build mesh
|
||||
time.Sleep(time.Second * 2)
|
||||
|
||||
msg := []byte("i like cats")
|
||||
err := psubs[0].Publish("foobar", msg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// last node in the chain should get the message
|
||||
select {
|
||||
case out := <-subs[4].ch:
|
||||
if !bytes.Equal(out.GetData(), msg) {
|
||||
t.Fatal("got wrong data")
|
||||
}
|
||||
case <-time.After(time.Second * 5):
|
||||
t.Fatal("timed out waiting for message")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGossipsubTreeTopology(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
hosts := getNetHosts(t, ctx, 10)
|
||||
psubs := getGossipsubs(ctx, hosts)
|
||||
|
||||
connect(t, hosts[0], hosts[1])
|
||||
connect(t, hosts[1], hosts[2])
|
||||
connect(t, hosts[1], hosts[4])
|
||||
connect(t, hosts[2], hosts[3])
|
||||
connect(t, hosts[0], hosts[5])
|
||||
connect(t, hosts[5], hosts[6])
|
||||
connect(t, hosts[5], hosts[8])
|
||||
connect(t, hosts[6], hosts[7])
|
||||
connect(t, hosts[8], hosts[9])
|
||||
|
||||
/*
|
||||
[0] -> [1] -> [2] -> [3]
|
||||
| L->[4]
|
||||
v
|
||||
[5] -> [6] -> [7]
|
||||
|
|
||||
v
|
||||
[8] -> [9]
|
||||
*/
|
||||
|
||||
var chs []*Subscription
|
||||
for _, ps := range psubs {
|
||||
ch, err := ps.Subscribe("fizzbuzz")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
chs = append(chs, ch)
|
||||
}
|
||||
|
||||
// wait for heartbeats to build mesh
|
||||
time.Sleep(time.Second * 2)
|
||||
|
||||
assertPeerLists(t, hosts, psubs[0], 1, 5)
|
||||
assertPeerLists(t, hosts, psubs[1], 0, 2, 4)
|
||||
assertPeerLists(t, hosts, psubs[2], 1, 3)
|
||||
|
||||
checkMessageRouting(t, "fizzbuzz", []*PubSub{psubs[9], psubs[3]}, chs)
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user