Define and subscribe to solidity events

ethers.nimble

@@ -4,7 +4,7 @@ description = "library for interacting with Ethereum"
 license = "MIT"
 
 requires "chronos >= 3.0.0 & < 4.0.0"
-requires "contractabi >= 0.4.1 & < 0.5.0"
+requires "contractabi >= 0.4.2 & < 0.5.0"
 requires "questionable >= 0.10.2 & < 0.11.0"
 requires "upraises >= 0.1.0 & < 0.2.0"
 requires "json_rpc"

ethers/contract.nim

@@ -5,6 +5,7 @@ import ./basics
 import ./provider
 import ./signer
 import ./events
+import ./fields
 
 export basics
 export provider
@@ -16,6 +17,7 @@ type
     signer: ?Signer
     address: Address
   ContractError* = object of EthersError
+  EventHandler*[E: Event] = proc(event: E) {.gcsafe, upraises:[].}
 
 func new*(ContractType: type Contract,
           address: Address,
@@ -144,3 +146,17 @@ macro contract*(procedure: untyped{nkProcDef|nkMethodDef}): untyped =
 
 template view* {.pragma.}
 template pure* {.pragma.}
+
+proc subscribe*[E: Event](contract: Contract,
+                          _: type E,
+                          handler: EventHandler[E]):
+                         Future[Subscription] =
+
+  let topic = topic($E, E.fieldTypes).toArray
+  let filter = Filter(address: contract.address, topics: @[topic])
+
+  proc logHandler(log: Log) {.upraises: [].} =
+    if event =? E.decode(log.data, log.topics):
+      handler(event)
+
+  contract.provider.subscribe(filter, logHandler)

ethers/events.nim

@@ -1,10 +1,10 @@
 import std/macros
 import pkg/contractabi
 import ./basics
+import ./provider
 
 type
   Event* = object of RootObj
-  Topic* = array[32, byte]
   ValueType = uint8 | uint16 | uint32 | uint64 | UInt256 | UInt128 |
               int8 | int16 | int32 | int64 | Int256 | Int128 |
               bool | Address

ethers/fields.nim

@@ -0,0 +1,12 @@
+import std/macros
+
+macro fieldValues*(obj: object): tuple =
+  result = newNimNode(nnkTupleConstr)
+  let typ = getTypeImpl(obj)
+  let fields = typ[2]
+  for field in fields:
+    let name = field[0]
+    result.add newDotExpr(obj, name)
+
+template fieldTypes*(T: type): type tuple =
+  typeof fieldValues(T.default)

ethers/provider.nim

@@ -10,6 +10,15 @@ push: {.upraises: [].}
 
 type
   Provider* = ref object of RootObj
+  Subscription* = ref object of RootObj
+  Filter* = object
+    address*: Address
+    topics*: seq[Topic]
+  Log* = object
+    data*: seq[byte]
+    topics*: seq[Topic]
+  LogHandler* = proc(log: Log) {.gcsafe, upraises:[].}
+  Topic* = array[32, byte]
 
 method getBlockNumber*(provider: Provider): Future[UInt256] {.base.} =
   doAssert false, "not implemented"
@@ -32,3 +41,12 @@ method estimateGas*(provider: Provider,
 
 method getChainId*(provider: Provider): Future[UInt256] {.base.} =
   doAssert false, "not implemented"
+
+method subscribe*(provider: Provider,
+                  filter: Filter,
+                  callback: LogHandler):
+                 Future[Subscription] {.base.} =
+  doAssert false, "not implemented"
+
+method unsubscribe*(subscription: Subscription) {.base, async.} =
+  doAssert false, "not implemented"

ethers/providers/conversions.nim

@@ -1,8 +1,15 @@
 import std/json
+import pkg/json_rpc/jsonmarshal
 import pkg/stew/byteutils
-import ../../basics
+import ../basics
-import ../../transaction
+import ../transaction
-import ../../blocktag
+import ../blocktag
+import ../provider
+
+export jsonmarshal
+
+func fromJson*(T: type, json: JsonNode, name = ""): T =
+  fromJson(json, name, result)
 
 # byte sequence
 
@@ -58,3 +65,12 @@ func `%`*(transaction: Transaction): JsonNode =
 
 func `%`*(blockTag: BlockTag): JsonNode =
   %($blockTag)
+
+# Log
+
+func fromJson*(json: JsonNode, name: string, result: var Log) =
+  var data: seq[byte]
+  var topics: seq[Topic]
+  fromJson(json["data"], "data", data)
+  fromJson(json["topics"], "topics", topics)
+  result = Log(data: data, topics: topics)

ethers/providers/jsonrpc.nim

@@ -1,9 +1,12 @@
+import std/json
+import std/tables
 import std/uri
 import pkg/json_rpc/rpcclient
 import ../basics
 import ../provider
 import ../signer
 import ./rpccalls
+import ./conversions
 
 export basics
 export provider
@@ -13,6 +16,10 @@ push: {.upraises: [].}
 type
   JsonRpcProvider* = ref object of Provider
     client: Future[RpcClient]
+    subscriptions: Table[JsonNode, LogHandler]
+  JsonRpcSubscription = ref object of Subscription
+    provider: JsonRpcProvider
+    id: JsonNode
   JsonRpcSigner* = ref object of Signer
     provider: JsonRpcProvider
     address: ?Address
@@ -36,8 +43,12 @@ proc connect(_: type RpcClient, url: string): Future[RpcClient] {.async.} =
     await client.connect(url)
     return client
 
+proc handleSubscriptions(provider: JsonRpcProvider) {.async.}
+
 proc new*(_: type JsonRpcProvider, url=defaultUrl): JsonRpcProvider =
-  JsonRpcProvider(client: RpcClient.connect(url))
+  let provider = JsonRpcProvider(client: RpcClient.connect(url))
+  asyncSpawn provider.handleSubscriptions()
+  provider
 
 proc send*(provider: JsonRpcProvider,
            call: string,
@@ -87,6 +98,42 @@ method getChainId*(provider: JsonRpcProvider): Future[UInt256] {.async.} =
   except CatchableError:
     return parse(await client.net_version(), UInt256)
 
+proc handleSubscriptions(provider: JsonRpcProvider) {.async.} =
+
+  proc getLogHandler(id: JsonNode): ?LogHandler =
+    try:
+      if provider.subscriptions.hasKey(id):
+        provider.subscriptions[id].some
+      else:
+        LogHandler.none
+    except Exception:
+      LogHandler.none
+
+  proc handleSubscription(arguments: JsonNode) {.upraises: [].} =
+    if id =? arguments["subscription"].catch and
+       handler =? getLogHandler(id) and
+       log =? Log.fromJson(arguments["result"]).catch:
+      handler(log)
+
+  let client = await provider.client
+  client.setMethodHandler("eth_subscription", handleSubscription)
+
+method subscribe*(provider: JsonRpcProvider,
+                  filter: Filter,
+                  callback: LogHandler):
+                 Future[Subscription] {.async.} =
+  let client = await provider.client
+  doAssert client of RpcWebSocketClient, "subscriptions require websockets"
+  let id = await client.eth_subscribe("logs", some filter)
+  provider.subscriptions[id] = callback
+  return JsonRpcSubscription(id: id, provider: provider)
+
+method unsubscribe*(subscription: JsonRpcSubscription) {.async.} =
+  let provider = subscription.provider
+  let client = await provider.client
+  discard await client.eth_unsubscribe(subscription.id)
+  provider.subscriptions.del(subscription.id)
+
 # Signer
 
 method provider*(signer: JsonRpcSigner): Provider =

ethers/providers/rpccalls.nim

@@ -3,7 +3,8 @@ import pkg/json_rpc/rpcclient
 import ../basics
 import ../transaction
 import ../blocktag
-import ./rpccalls/conversions
+import ../provider
+import ./conversions
 
 const file = currentSourcePath.parentDir / "rpccalls" / "signatures.nim"
 

ethers/providers/rpccalls/signatures.nim

@@ -8,3 +8,5 @@ proc eth_estimateGas(transaction: Transaction): UInt256
 proc eth_chainId(): UInt256
 proc eth_sendTransaction(transaction: Transaction): array[32, byte]
 proc eth_sign(account: Address, message: seq[byte]): seq[byte]
+proc eth_subscribe(name: string, filter: ?Filter): JsonNode
+proc eth_unsubscribe(id: JsonNode): bool

testmodule/testContracts.nim

@@ -5,9 +5,15 @@ import pkg/ethers
 import ./hardhat
 
 type
+
   Erc20* = ref object of Contract
   TestToken = ref object of Erc20
 
+  Transfer = object of Event
+    sender {.indexed.}: Address
+    receiver {.indexed.}: Address
+    value: UInt256
+
 method totalSupply*(erc20: Erc20): UInt256 {.base, contract, view.}
 method balanceOf*(erc20: Erc20, account: Address): UInt256 {.base, contract, view.}
 method allowance*(erc20: Erc20, owner, spender: Address): UInt256 {.base, contract, view.}
@@ -23,7 +29,7 @@ suite "Contracts":
   var accounts: seq[Address]
 
   setup:
-    provider = JsonRpcProvider.new()
+    provider = JsonRpcProvider.new("ws://localhost:8545")
     snapshot = await provider.send("evm_snapshot")
     accounts = await provider.listAccounts()
     let deployment = readDeployment()
@@ -77,3 +83,40 @@ suite "Contracts":
     check (await token.connect(provider).balanceOf(accounts[0])) == 50.u256
     check (await token.connect(provider).balanceOf(accounts[1])) == 25.u256
     check (await token.connect(provider).balanceOf(accounts[2])) == 25.u256
+
+  test "receives events when subscribed":
+    var transfers: seq[Transfer]
+
+    proc handleTransfer(transfer: Transfer) =
+      transfers.add(transfer)
+
+    let signer0 = provider.getSigner(accounts[0])
+    let signer1 = provider.getSigner(accounts[1])
+
+    let subscription = await token.subscribe(Transfer, handleTransfer)
+    await token.connect(signer0).mint(accounts[0], 100.u256)
+    await token.connect(signer0).transfer(accounts[1], 50.u256)
+    await token.connect(signer1).transfer(accounts[2], 25.u256)
+    await subscription.unsubscribe()
+
+    check transfers == @[
+      Transfer(receiver: accounts[0], value: 100.u256),
+      Transfer(sender: accounts[0], receiver: accounts[1], value: 50.u256),
+      Transfer(sender: accounts[1], receiver: accounts[2], value: 25.u256)
+    ]
+
+  test "stops receiving events when unsubscribed":
+    var transfers: seq[Transfer]
+
+    proc handleTransfer(transfer: Transfer) =
+      transfers.add(transfer)
+
+    let signer0 = provider.getSigner(accounts[0])
+
+    let subscription = await token.subscribe(Transfer, handleTransfer)
+    await token.connect(signer0).mint(accounts[0], 100.u256)
+    await subscription.unsubscribe()
+
+    await token.connect(signer0).transfer(accounts[1], 50.u256)
+
+    check transfers == @[Transfer(receiver: accounts[0], value: 100.u256)]