mirror of
https://github.com/logos-messaging/logos-messaging-go-bindings.git
synced 2026-07-07 03:09:31 +00:00
c653002984
TestRelaySendReceive connects two relay nodes, publishes a random payload from one and requires the other to receive it — end-to-end over the single library. Compiled by the PR gate; run under `make test` / locally against a built liblogosdelivery. (Replaces the former examples/kernel, removed alongside the ffi flatten.) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
91 lines
2.8 KiB
Go
91 lines
2.8 KiB
Go
package kernel
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/logos-messaging/logos-delivery-go-bindings/pkg/kernel/common"
|
|
"github.com/logos-messaging/logos-delivery-go-bindings/pkg/kernel/pb"
|
|
"github.com/stretchr/testify/require"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
// TestRelaySendReceive is an end-to-end check against the single liblogosdelivery
|
|
// library: two relay nodes are connected, one publishes a random payload and the
|
|
// other must receive it. It exercises the unified node lifecycle and the Kernel
|
|
// relay ops (waku_relay_subscribe/publish) over the one library.
|
|
func TestRelaySendReceive(t *testing.T) {
|
|
const clusterID, shardID = 16, 64
|
|
|
|
newNode := func(name string) *WakuNode {
|
|
node, err := StartWakuNode(name, &common.WakuConfig{
|
|
Relay: true,
|
|
LogLevel: "ERROR",
|
|
Discv5Discovery: false,
|
|
ClusterID: clusterID,
|
|
Shards: []uint16{shardID},
|
|
})
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { _ = node.StopAndDestroy() })
|
|
return node
|
|
}
|
|
|
|
sender := newNode("sender")
|
|
receiver := newNode("receiver")
|
|
|
|
topic := FormatWakuRelayTopic(clusterID, shardID)
|
|
require.NoError(t, sender.RelaySubscribe(topic))
|
|
require.NoError(t, receiver.RelaySubscribe(topic))
|
|
|
|
// Dial the receiver from the sender using the receiver's listen multiaddr
|
|
// (it already embeds the peer id).
|
|
addrs, err := receiver.ListenAddresses()
|
|
require.NoError(t, err)
|
|
require.NotEmpty(t, addrs)
|
|
|
|
connCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
require.NoError(t, sender.Connect(connCtx, addrs[0]))
|
|
require.Eventually(t, func() bool {
|
|
n, _ := sender.GetNumConnectedPeers()
|
|
return n >= 1
|
|
}, 15*time.Second, time.Second, "sender never connected to the receiver")
|
|
|
|
payload := make([]byte, 16)
|
|
_, err = rand.Read(payload)
|
|
require.NoError(t, err)
|
|
|
|
publish := func() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
// May fail with NoPeersToPublish until gossipsub grafts the mesh; retried.
|
|
_, _ = sender.RelayPublish(ctx, &pb.WakuMessage{
|
|
Payload: payload,
|
|
ContentTopic: "/kernel-test/1/relay/proto",
|
|
Version: proto.Uint32(0),
|
|
Timestamp: proto.Int64(time.Now().UnixNano()),
|
|
}, topic)
|
|
}
|
|
|
|
// Publish immediately, then retry each second while waiting for delivery —
|
|
// the relay mesh takes a moment to form after the connection is established.
|
|
publish()
|
|
ticker := time.NewTicker(1 * time.Second)
|
|
defer ticker.Stop()
|
|
deadline := time.After(30 * time.Second)
|
|
for {
|
|
select {
|
|
case env := <-receiver.MsgChan:
|
|
if string(env.Message().GetPayload()) == string(payload) {
|
|
return // received our exact message — success
|
|
}
|
|
case <-ticker.C:
|
|
publish()
|
|
case <-deadline:
|
|
t.Fatal("timed out waiting for the receiver to get the message")
|
|
}
|
|
}
|
|
}
|