mirror of
https://github.com/ipfs/kubo.git
synced 2026-02-27 21:37:57 +08:00
3eba14aa24
Updates: * go-kad-dht: Query performance improvements, DHT client fixes, validates records on *local* put. * go-libp2p-swarm/go-libp2p-transport: Timeout improvements. * go-multiaddr-net: Exposes useful Conn methods (CloseWrite, CloseRead, etc.) * go-log: fixes possible panic when enabling/disabling events. * go-multiaddr: fixes possible panic when stringifying malformed multiaddrs, adds support for consuming /p2p/ multiaddrs. fixes #5113 unblocks #4895 License: MIT Signed-off-by: Steven Allen <steven@stebalien.com>
99 lines
2.5 KiB
Go
99 lines
2.5 KiB
Go
package bitswap
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
|
|
bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
|
|
bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
|
|
|
|
delay "gx/ipfs/QmRJVNatYJwTAHgdSM1Xef9QVQ1Ch3XHdmcrykjP5Y4soL/go-ipfs-delay"
|
|
blocks "gx/ipfs/QmTRCUvZLiir12Qr6MV3HKfKMHX8Nf1Vddn6t2g5nsQSb9/go-block-format"
|
|
mockrouting "gx/ipfs/QmWLQyLU7yopJnwMvpHM5VSMG4xmbKgcq6P246mDy9xy5E/go-ipfs-routing/mock"
|
|
testutil "gx/ipfs/QmcW4FGAt24fdK1jBgWQn3yP4R9ZLyWQqjozv9QK7epRhL/go-testutil"
|
|
peer "gx/ipfs/QmdVrMn1LhB4ybb8hMVaMLXnA8XRSewMnK6YqXKXoTcRvN/go-libp2p-peer"
|
|
)
|
|
|
|
func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
|
|
net := VirtualNetwork(mockrouting.NewServer(), delay.Fixed(0))
|
|
responderPeer := testutil.RandIdentityOrFatal(t)
|
|
waiter := net.Adapter(testutil.RandIdentityOrFatal(t))
|
|
responder := net.Adapter(responderPeer)
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
wg.Add(1)
|
|
|
|
expectedStr := "received async"
|
|
|
|
responder.SetDelegate(lambda(func(
|
|
ctx context.Context,
|
|
fromWaiter peer.ID,
|
|
msgFromWaiter bsmsg.BitSwapMessage) {
|
|
|
|
msgToWaiter := bsmsg.New(true)
|
|
msgToWaiter.AddBlock(blocks.NewBlock([]byte(expectedStr)))
|
|
waiter.SendMessage(ctx, fromWaiter, msgToWaiter)
|
|
}))
|
|
|
|
waiter.SetDelegate(lambda(func(
|
|
ctx context.Context,
|
|
fromResponder peer.ID,
|
|
msgFromResponder bsmsg.BitSwapMessage) {
|
|
|
|
// TODO assert that this came from the correct peer and that the message contents are as expected
|
|
ok := false
|
|
for _, b := range msgFromResponder.Blocks() {
|
|
if string(b.RawData()) == expectedStr {
|
|
wg.Done()
|
|
ok = true
|
|
}
|
|
}
|
|
|
|
if !ok {
|
|
t.Fatal("Message not received from the responder")
|
|
}
|
|
}))
|
|
|
|
messageSentAsync := bsmsg.New(true)
|
|
messageSentAsync.AddBlock(blocks.NewBlock([]byte("data")))
|
|
errSending := waiter.SendMessage(
|
|
context.Background(), responderPeer.ID(), messageSentAsync)
|
|
if errSending != nil {
|
|
t.Fatal(errSending)
|
|
}
|
|
|
|
wg.Wait() // until waiter delegate function is executed
|
|
}
|
|
|
|
type receiverFunc func(ctx context.Context, p peer.ID,
|
|
incoming bsmsg.BitSwapMessage)
|
|
|
|
// lambda returns a Receiver instance given a receiver function
|
|
func lambda(f receiverFunc) bsnet.Receiver {
|
|
return &lambdaImpl{
|
|
f: f,
|
|
}
|
|
}
|
|
|
|
type lambdaImpl struct {
|
|
f func(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage)
|
|
}
|
|
|
|
func (lam *lambdaImpl) ReceiveMessage(ctx context.Context,
|
|
p peer.ID, incoming bsmsg.BitSwapMessage) {
|
|
lam.f(ctx, p, incoming)
|
|
}
|
|
|
|
func (lam *lambdaImpl) ReceiveError(err error) {
|
|
// TODO log error
|
|
}
|
|
|
|
func (lam *lambdaImpl) PeerConnected(p peer.ID) {
|
|
// TODO
|
|
}
|
|
func (lam *lambdaImpl) PeerDisconnected(peer.ID) {
|
|
// TODO
|
|
}
|