mirror of https://github.com/status-im/go-waku.git
test: replace peer_events example app by a test
This commit is contained in:
Richard Ramos
parent
555e5dd283
commit
6a3a5b08b3
|
|
@ -1,6 +0,0 @@
|
|||
.PHONY: all build
|
||||
|
||||
build:
|
||||
go build -o build/peer_events .
|
||||
|
||||
all: build
|
||||
|
|
@ -1,23 +0,0 @@
|
|||
# Using the `peer_events` application
|
||||
|
||||
## Background
|
||||
|
||||
The `peer_events` application is a basic example app that demonstrates how peer event handling works in go-waku
|
||||
|
||||
## Preparation
|
||||
```
|
||||
make
|
||||
```
|
||||
|
||||
## Basic application usage
|
||||
|
||||
To start the `peer_events` application run the following from the project directory
|
||||
|
||||
```
|
||||
./build/peer_events
|
||||
```
|
||||
The app will run the following nodes sequentially:
|
||||
- relayNode1 and relayNode2
|
||||
- relayNode2 is stopped, and relayNode3 is started
|
||||
- relayNode3 is stopped, and storeNode is started
|
||||
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
*
|
||||
!.gitignore
|
||||
|
|
@ -1,13 +0,0 @@
|
|||
module peer_events
|
||||
|
||||
go 1.15
|
||||
|
||||
// replace github.com/status-im/go-waku => ../..
|
||||
|
||||
replace github.com/ethereum/go-ethereum v1.9.5 => github.com/status-im/go-ethereum v1.9.5-status.12
|
||||
|
||||
require (
|
||||
github.com/ethereum/go-ethereum v1.10.4
|
||||
github.com/ipfs/go-log v1.0.5
|
||||
github.com/status-im/go-waku v0.0.0-20211012131444-baf57d82a30a // indirect
|
||||
)
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,231 +0,0 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ecdsa"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/crypto"
|
||||
logging "github.com/ipfs/go-log"
|
||||
"github.com/status-im/go-waku/waku/v2/node"
|
||||
"github.com/status-im/go-waku/waku/v2/protocol/pb"
|
||||
"github.com/status-im/go-waku/waku/v2/protocol/relay"
|
||||
"github.com/status-im/go-waku/waku/v2/utils"
|
||||
//"github.com/status-im/go-waku/waku/v2/protocol/store"
|
||||
)
|
||||
|
||||
var log = logging.Logger("peer_events")
|
||||
|
||||
var pubSubTopic = relay.DefaultWakuTopic
|
||||
|
||||
const contentTopic = "test"
|
||||
|
||||
func main() {
|
||||
lvl, err := logging.LevelFromString("info")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
logging.SetAllLoggers(lvl)
|
||||
|
||||
type AddrAndKey struct {
|
||||
addr *net.TCPAddr
|
||||
key *ecdsa.PrivateKey
|
||||
}
|
||||
nodeCount := 4
|
||||
addrsAndKeys := make([]*AddrAndKey, nodeCount)
|
||||
|
||||
for i := 0; i < nodeCount; i++ {
|
||||
addr, _ := net.ResolveTCPAddr("tcp", fmt.Sprintf("0.0.0.0:6000%d", i))
|
||||
key, err := randomHex(32)
|
||||
if err != nil {
|
||||
log.Error("Could not generate random key")
|
||||
return
|
||||
}
|
||||
prvKey, err := crypto.HexToECDSA(key)
|
||||
addrAndKey := &AddrAndKey{
|
||||
addr: addr,
|
||||
key: prvKey,
|
||||
}
|
||||
addrsAndKeys[i] = addrAndKey
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
//connStatusChan := make(chan node.ConnStatus)
|
||||
log.Info("### create relayNode1")
|
||||
relayNode1, err := node.New(ctx,
|
||||
node.WithPrivateKey(addrsAndKeys[0].key),
|
||||
node.WithHostAddress([]*net.TCPAddr{addrsAndKeys[0].addr}),
|
||||
node.WithWakuRelay(),
|
||||
//node.WithConnStatusChan(connStatusChan),
|
||||
node.WithWakuStore(true, false),
|
||||
node.WithKeepAlive(time.Duration(2)*time.Second),
|
||||
)
|
||||
|
||||
if err = relayNode1.Start(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
// relayNode2, err := node.New(ctx,
|
||||
// node.WithPrivateKey(addrsAndKeys[1].key),
|
||||
// node.WithHostAddress([]net.Addr{addrsAndKeys[1].addr}),
|
||||
// node.WithWakuRelay(),
|
||||
// )
|
||||
|
||||
log.Info("### before DialPeer")
|
||||
//staticNode := "/ip4/8.210.222.231/tcp/30303/p2p/16Uiu2HAm4v86W3bmT1BiH6oSPzcsSr24iDQpSN5Qa992BCjjwgrD"
|
||||
staticNode := "/ip4/188.166.135.145/tcp/30303/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e"
|
||||
relayNode1.DialPeer(ctx, staticNode)
|
||||
//relayNode2.DialPeer(relayNode1.ListenAddresses()[0])
|
||||
|
||||
//go writeLoop(ctx, relayNode1)
|
||||
//go readLoop(relayNode1)
|
||||
|
||||
//go readLoop(relayNode2)
|
||||
|
||||
log.Info("### Peer dialled")
|
||||
// printNodeConns := func(node *node.WakuNode) {
|
||||
// log.Info(node.Host().ID(), ": ", "peerCount: ", len(node.Host().Peerstore().Peers()))
|
||||
// log.Info("node peers: ")
|
||||
// for k, v := range node.GetPeerStats() {
|
||||
// log.Info(k, " ", v)
|
||||
// }
|
||||
// log.Info(node.Host().ID(), ": ", "isOnline/hasHistory ", node.IsOnline(), " ", node.HasHistory())
|
||||
// log.Info("end")
|
||||
|
||||
// }
|
||||
|
||||
// go func() {
|
||||
// for connStatus := range connStatusChan {
|
||||
// log.Info("Conn status update: ", connStatus)
|
||||
// }
|
||||
// }()
|
||||
// go func() {
|
||||
// ticker := time.NewTicker(time.Millisecond * 1000)
|
||||
// defer ticker.Stop()
|
||||
// for {
|
||||
// select {
|
||||
// case <-ticker.C:
|
||||
// printNodeConns(relayNode1)
|
||||
// }
|
||||
// }
|
||||
// }()
|
||||
|
||||
// time.Sleep(3 * time.Second)
|
||||
// log.Info("stop relayNode2")
|
||||
// relayNode2.Host().Close()
|
||||
|
||||
// time.Sleep(3 * time.Second)
|
||||
|
||||
// log.Info("start relayNode3")
|
||||
|
||||
// relayNode3, err := node.New(ctx,
|
||||
// node.WithPrivateKey(addrsAndKeys[2].key),
|
||||
// node.WithHostAddress([]net.Addr{addrsAndKeys[2].addr}),
|
||||
// node.WithWakuRelay(),
|
||||
// )
|
||||
|
||||
// relayNode3.DialPeer(relayNode1.ListenAddresses()[0])
|
||||
|
||||
// time.Sleep(3 * time.Second)
|
||||
// log.Info("stop relayNode3")
|
||||
// //relayNode3.Stop()
|
||||
|
||||
// log.Info("start storeNode")
|
||||
// // Start a store node
|
||||
// storeNode, _ := node.New(ctx,
|
||||
// node.WithPrivateKey(addrsAndKeys[3].key),
|
||||
// node.WithHostAddress([]net.Addr{addrsAndKeys[3].addr}),
|
||||
// node.WithWakuRelay(),
|
||||
// node.WithWakuStore(true),
|
||||
// )
|
||||
// tCtx, _ := context.WithTimeout(ctx, 5*time.Second)
|
||||
// log.Info("#before AddStorePeer")
|
||||
// storeNodeId, err := relayNode1.AddStorePeer(storeNode.ListenAddresses()[0])
|
||||
// time.Sleep(3 * time.Second)
|
||||
// log.Info("#before Query")
|
||||
// _, err = relayNode1.Query(tCtx, []string{contentTopic}, 0, 0, store.WithPeer(*storeNodeId))
|
||||
// log.Info("storeNode.ListenAddresses(): ", storeNode.ListenAddresses(), storeNodeId)
|
||||
// if err != nil {
|
||||
// log.Info("### error adding store peer: ", err)
|
||||
// }
|
||||
|
||||
// time.Sleep(3 * time.Second)
|
||||
// log.Info("stop storeNode")
|
||||
// storeNode.Stop()
|
||||
|
||||
// time.Sleep(3 * time.Second)
|
||||
// // // Wait for a SIGINT or SIGTERM signal
|
||||
ch := make(chan os.Signal, 1)
|
||||
signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-ch
|
||||
fmt.Println("\n\n\nReceived signal, shutting down...")
|
||||
|
||||
// // shut the nodes down
|
||||
// relayNode1.Stop()
|
||||
|
||||
}
|
||||
|
||||
func randomHex(n int) (string, error) {
|
||||
bytes := make([]byte, n)
|
||||
if _, err := rand.Read(bytes); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(bytes), nil
|
||||
}
|
||||
|
||||
func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
|
||||
var version uint32 = 0
|
||||
var timestamp float64 = utils.GetUnixEpoch()
|
||||
|
||||
p := new(node.Payload)
|
||||
p.Data = []byte(wakuNode.ID() + ": " + msgContent)
|
||||
p.Key = &node.KeyInfo{Kind: node.None}
|
||||
|
||||
payload, err := p.Encode(version)
|
||||
|
||||
msg := &pb.WakuMessage{
|
||||
Payload: payload,
|
||||
Version: version,
|
||||
ContentTopic: contentTopic,
|
||||
Timestamp: timestamp,
|
||||
}
|
||||
|
||||
_, err = wakuNode.Relay().Publish(ctx, msg, nil)
|
||||
if err != nil {
|
||||
log.Error("Error sending a message: ", err)
|
||||
}
|
||||
}
|
||||
|
||||
func writeLoop(ctx context.Context, wakuNode *node.WakuNode) {
|
||||
for {
|
||||
time.Sleep(2 * time.Second)
|
||||
write(ctx, wakuNode, "Hello world!")
|
||||
//log.Info("peerCount: ", len(wakuNode.Host().Peerstore().Peers()))
|
||||
}
|
||||
}
|
||||
|
||||
func readLoop(ctx context.Context, wakuNode *node.WakuNode) {
|
||||
sub, err := wakuNode.Relay().Subscribe(ctx, &pubSubTopic)
|
||||
if err != nil {
|
||||
log.Error("Could not subscribe: ", err)
|
||||
return
|
||||
}
|
||||
|
||||
for value := range sub.C {
|
||||
payload, err := node.DecodePayload(value.Message(), &node.KeyInfo{Kind: node.None})
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Received msg, ", wakuNode.ID(), ", payload: ", string(payload.Data))
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,122 @@
|
|||
package node
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestConnectionStatusChanges(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
connStatusChan := make(chan ConnStatus)
|
||||
|
||||
// Node1: Only Relay
|
||||
hostAddr1, err := net.ResolveTCPAddr("tcp", "0.0.0.0:0")
|
||||
require.NoError(t, err)
|
||||
node1, err := New(ctx,
|
||||
WithHostAddress([]*net.TCPAddr{hostAddr1}),
|
||||
WithWakuRelay(),
|
||||
WithConnStatusChan(connStatusChan),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
err = node1.Start()
|
||||
require.NoError(t, err)
|
||||
|
||||
// Node2: Relay
|
||||
hostAddr2, err := net.ResolveTCPAddr("tcp", "0.0.0.0:0")
|
||||
require.NoError(t, err)
|
||||
node2, err := New(ctx,
|
||||
WithHostAddress([]*net.TCPAddr{hostAddr2}),
|
||||
WithWakuRelay(),
|
||||
)
|
||||
err = node2.Start()
|
||||
require.NoError(t, err)
|
||||
|
||||
// Node3: Relay + Store
|
||||
hostAddr3, err := net.ResolveTCPAddr("tcp", "0.0.0.0:0")
|
||||
require.NoError(t, err)
|
||||
node3, err := New(ctx,
|
||||
WithHostAddress([]*net.TCPAddr{hostAddr3}),
|
||||
WithWakuRelay(),
|
||||
WithWakuStore(false, false),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
err = node3.Start()
|
||||
require.NoError(t, err)
|
||||
|
||||
node1.DialPeer(ctx, node2.ListenAddresses()[0].String())
|
||||
node1.DialPeer(ctx, node3.ListenAddresses()[0].String())
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
connStatus := <-connStatusChan
|
||||
_, ok := connStatus.Peers[node2.Host().ID()]
|
||||
require.True(t, connStatus.IsOnline)
|
||||
require.True(t, ok)
|
||||
require.False(t, connStatus.HasHistory)
|
||||
|
||||
connStatus = <-connStatusChan
|
||||
_, ok = connStatus.Peers[node3.Host().ID()]
|
||||
require.True(t, connStatus.IsOnline)
|
||||
require.True(t, ok)
|
||||
require.True(t, connStatus.HasHistory)
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
connStatus := <-connStatusChan
|
||||
_, ok := connStatus.Peers[node3.Host().ID()]
|
||||
require.True(t, connStatus.IsOnline)
|
||||
require.False(t, ok) // Peer3 should have been disconnected
|
||||
require.False(t, connStatus.HasHistory) // No history, because there are no peers connected with store protocol
|
||||
require.Len(t, node1.Host().Network().Peers(), 1) // No peers connected
|
||||
}()
|
||||
|
||||
node3.Stop()
|
||||
wg.Wait()
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
connStatus := <-connStatusChan
|
||||
_, ok := connStatus.Peers[node3.Host().ID()]
|
||||
require.False(t, connStatus.IsOnline) // Peers are not connected. Should be offline
|
||||
require.False(t, ok) // Peer2 should have been disconnected
|
||||
require.False(t, connStatus.HasHistory) // No history, because there are no peers connected with store protocol
|
||||
require.Len(t, node1.Host().Network().Peers(), 0) // No peers connected
|
||||
}()
|
||||
|
||||
err = node1.ClosePeerById(node2.Host().ID())
|
||||
require.NoError(t, err)
|
||||
wg.Wait()
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
connStatus := <-connStatusChan
|
||||
_, ok := connStatus.Peers[node2.Host().ID()]
|
||||
require.True(t, connStatus.IsOnline) // Peers2 is connected. Should be online
|
||||
require.True(t, ok) // Peer2 should have been connected
|
||||
require.False(t, connStatus.HasHistory) // No history because peer2 only has relay
|
||||
require.Len(t, node1.Host().Network().Peers(), 1)
|
||||
}()
|
||||
|
||||
err = node1.DialPeerByID(ctx, node2.Host().ID())
|
||||
require.NoError(t, err)
|
||||
wg.Wait()
|
||||
}
|
||||
Loading…
Reference in New Issue