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" testutil "gx/ipfs/QmUJzxQQ2kzwQubsMqBTr1NGDpLfh7pGA2E1oaJULcKDPq/go-testutil" mockrouting "gx/ipfs/QmcE3B6ittYBmctva8Q155LPa1YPcVqg8N7pPcgt9i7iAQ/go-ipfs-routing/mock" peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer" blocks "gx/ipfs/Qmej7nf81hi2x2tvjRBF3mcp74sQyuDH4VMYDGd1YtXjb2/go-block-format" ) 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 }