Engine sends payments for received blocks

dagger/bitswap/engine.nim

@@ -23,6 +23,7 @@ import ../utils/asyncheapqueue
 import ./network
 import ./pendingblocks
 import ./peercontext
+import ./engine/payments
 
 export peercontext
 
@@ -187,6 +188,20 @@ proc resolveBlocks*(b: BitswapEngine, blocks: seq[bt.Block]) =
   b.pendingBlocks.resolve(blocks)
   b.scheduleTasks(blocks)
 
+proc payForBlocks(engine: BitswapEngine,
+                  peer: BitswapPeerCtx,
+                  blocks: seq[bt.Block]) =
+  let sendPayment = engine.request.sendPayment
+  if sendPayment.isNil:
+    return
+
+  var amountOfBytes = 0
+  for blck in blocks:
+    amountOfBytes += blck.data.len
+
+  if payment =? engine.wallet.pay(peer, amountOfBytes):
+    sendPayment(peer.id, payment)
+
 proc blocksHandler*(
   b: BitswapEngine,
   peer: PeerID,
@@ -198,6 +213,10 @@ proc blocksHandler*(
   b.localStore.putBlocks(blocks)
   b.resolveBlocks(blocks)
 
+  let peerCtx = b.getPeerCtx(peer)
+  if peerCtx != nil:
+    b.payForBlocks(peerCtx, blocks)
+
 proc wantListHandler*(
   b: BitswapEngine,
   peer: PeerID,

dagger/bitswap/network.nim

@@ -53,12 +53,14 @@ type
   BlocksBroadcaster* = proc(peer: PeerID, presence: seq[bt.Block]) {.gcsafe.}
   PresenceBroadcaster* = proc(peer: PeerID, presence: seq[BlockPresence]) {.gcsafe.}
   PricingBroadcaster* = proc(peer: PeerID, pricing: Pricing) {.gcsafe.}
+  PaymentBroadcaster* = proc(peer: PeerID, payment: SignedState) {.gcsafe.}
 
   BitswapRequest* = object
     sendWantList*: WantListBroadcaster
     sendBlocks*: BlocksBroadcaster
     sendPresence*: PresenceBroadcaster
     sendPricing*: PricingBroadcaster
+    sendPayment*: PaymentBroadcaster
 
   BitswapNetwork* = ref object of LPProtocol
     peers*: Table[PeerID, NetworkPeer]

tests/dagger/bitswap/testengine.nim

@@ -138,7 +138,7 @@ suite "Bitswap engine handlers":
 
     await done
 
-  test "should handle blocks":
+  test "stores blocks in local store":
     let pending = blocks.mapIt(
       engine.pendingBlocks.addOrAwait( it.cid )
     )
@@ -149,6 +149,21 @@ suite "Bitswap engine handlers":
     for b in blocks:
       check engine.localStore.hasBlock(b.cid)
 
+  test "sends payments for received blocks":
+    let pricing = Pricing.example
+    engine.getPeerCtx(peerId).pricing = pricing.some
+
+    engine.request.sendPayment = proc(receiver: PeerID, payment: SignedState) =
+      let amount = blocks.mapIt(it.data.len).foldl(a+b).u256 * pricing.price
+      let balances = payment.state.outcome.balances(pricing.asset).get
+      check receiver == peerId
+      check balances[pricing.address.toDestination] == amount
+      done.complete()
+
+    engine.blocksHandler(peerId, blocks)
+
+    await done.wait(100.millis)
+
   test "should handle block presence":
     engine.blockPresenceHandler(
       peerId,