package bitswap import ( "sync" "testing" blocks "github.com/ipfs/go-ipfs/blocks" bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message" bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network" mockrouting "github.com/ipfs/go-ipfs/routing/mock" delay "github.com/ipfs/go-ipfs/thirdparty/delay" testutil "github.com/ipfs/go-ipfs/thirdparty/testutil" peer "gx/ipfs/QmZpD74pUj6vuxTp1o6LhA3JavC2Bvh9fsWPPVvHnD9sE7/go-libp2p-peer" context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context" ) 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.Data()) == 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 }