peermessagemanager_test.go 3.79 KB
Newer Older
1 2 3 4 5 6 7 8
package peermanager

import (
	"context"
	"math/rand"
	"testing"
	"time"

Hannah Howard's avatar
Hannah Howard committed
9
	basicnode "github.com/ipld/go-ipld-prime/node/basic"
10
	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
11
	"github.com/libp2p/go-libp2p-core/peer"
Hannah Howard's avatar
Hannah Howard committed
12
	"github.com/stretchr/testify/require"
Hannah Howard's avatar
Hannah Howard committed
13 14 15

	"github.com/ipfs/go-graphsync"
	gsmsg "github.com/ipfs/go-graphsync/message"
16
	"github.com/ipfs/go-graphsync/notifications"
Hannah Howard's avatar
Hannah Howard committed
17
	"github.com/ipfs/go-graphsync/testutil"
18 19 20 21 22 23 24
)

type messageSent struct {
	p       peer.ID
	message gsmsg.GraphSyncMessage
}

25 26
var _ PeerQueue = (*fakePeer)(nil)

27 28 29 30 31
type fakePeer struct {
	p            peer.ID
	messagesSent chan messageSent
}

32
func (fp *fakePeer) AllocateAndBuildMessage(blkSize uint64, buildMessage func(b *gsmsg.Builder), notifees []notifications.Notifee) {
33 34 35 36 37 38
	builder := gsmsg.NewBuilder(gsmsg.Topic(0))
	buildMessage(builder)
	message, err := builder.Build()
	if err != nil {
		panic(err)
	}
39 40 41 42

	fp.messagesSent <- messageSent{fp.p, message}
}

43 44 45 46 47 48 49 50
func (fp *fakePeer) Startup()  {}
func (fp *fakePeer) Shutdown() {}

//func (fp *fakePeer) AddRequest(graphSyncRequest gsmsg.GraphSyncRequest, notifees ...notifications.Notifee) {
//	message := gsmsg.New()
//	message.AddRequest(graphSyncRequest)
//	fp.messagesSent <- messageSent{fp.p, message}
//}
51

52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69
func makePeerQueueFactory(messagesSent chan messageSent) PeerQueueFactory {
	return func(ctx context.Context, p peer.ID) PeerQueue {
		return &fakePeer{
			p:            p,
			messagesSent: messagesSent,
		}
	}
}

func TestSendingMessagesToPeers(t *testing.T) {
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
	messagesSent := make(chan messageSent, 5)
	peerQueueFactory := makePeerQueueFactory(messagesSent)

	tp := testutil.GeneratePeers(5)

70 71
	id := graphsync.RequestID(rand.Int31())
	priority := graphsync.Priority(rand.Int31())
72
	root := testutil.GenerateCids(1)[0]
Eric Myhre's avatar
Eric Myhre committed
73
	ssb := builder.NewSelectorSpecBuilder(basicnode.Prototype.Any)
74
	selector := ssb.Matcher().Node()
75 76 77

	peerManager := NewMessageManager(ctx, peerQueueFactory)

78
	request := gsmsg.NewRequest(id, root, selector, priority)
79
	peerManager.AllocateAndBuildMessage(tp[0], 0, func(b *gsmsg.Builder) {
80 81
		b.AddRequest(request)
	}, []notifications.Notifee{})
82
	peerManager.AllocateAndBuildMessage(tp[1], 0, func(b *gsmsg.Builder) {
83 84
		b.AddRequest(request)
	}, []notifications.Notifee{})
85
	cancelRequest := gsmsg.CancelRequest(id)
86
	peerManager.AllocateAndBuildMessage(tp[0], 0, func(b *gsmsg.Builder) {
87 88
		b.AddRequest(cancelRequest)
	}, []notifications.Notifee{})
89

Hannah Howard's avatar
Hannah Howard committed
90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115
	var firstMessage messageSent
	testutil.AssertReceive(ctx, t, messagesSent, &firstMessage, "first message did not send")
	require.Equal(t, tp[0], firstMessage.p, "first message sent to incorrect peer")
	request = firstMessage.message.Requests()[0]
	require.Equal(t, id, request.ID())
	require.False(t, request.IsCancel())
	require.Equal(t, priority, request.Priority())
	require.Equal(t, selector, request.Selector())

	var secondMessage messageSent
	testutil.AssertReceive(ctx, t, messagesSent, &secondMessage, "second message did not send")
	require.Equal(t, tp[1], secondMessage.p, "second message sent to incorrect peer")
	request = secondMessage.message.Requests()[0]
	require.Equal(t, id, request.ID())
	require.False(t, request.IsCancel())
	require.Equal(t, priority, request.Priority())
	require.Equal(t, selector, request.Selector())

	var thirdMessage messageSent
	testutil.AssertReceive(ctx, t, messagesSent, &thirdMessage, "third message did not send")

	require.Equal(t, tp[0], thirdMessage.p, "third message sent to incorrect peer")
	request = thirdMessage.message.Requests()[0]
	require.Equal(t, id, request.ID())
	require.True(t, request.IsCancel())

116
	connectedPeers := peerManager.ConnectedPeers()
Hannah Howard's avatar
Hannah Howard committed
117 118 119 120
	require.Len(t, connectedPeers, 2)

	testutil.AssertContainsPeer(t, connectedPeers, tp[0])
	testutil.AssertContainsPeer(t, connectedPeers, tp[1])
121
}