logos-storage/nim-ethers
1
0
Fork 0
mirror of https://github.com/logos-storage/nim-ethers.git synced 2026-01-05 23:23:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add resubscribe flag

Browse Source
This commit is contained in:
Arnaud 2025-03-20 15:45:18 +01:00
parent c48ff9b84d
commit 4c19e14853
No known key found for this signature in database
GPG Key ID: 69D6CE281FCAE663
1 changed files with 30 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
ethers/providers/jsonrpc/subscriptions.nim
View File

@ -24,11 +24,33 @@ type
# about them # about them
# This is used of resubscribe all the subscriptions when using websocket with hardhat # This is used of resubscribe all the subscriptions when using websocket with hardhat
logFilters: Table[JsonNode, EventFilter] logFilters: Table[JsonNode, EventFilter]
when defined(resubscribe):
resubscribeFut: Future[void]
MethodHandler* = proc (j: JsonNode) {.gcsafe, raises: [].} MethodHandler* = proc (j: JsonNode) {.gcsafe, raises: [].}
SubscriptionCallback = proc(id: JsonNode, arguments: ?!JsonNode) {.gcsafe, raises:[].} SubscriptionCallback = proc(id: JsonNode, arguments: ?!JsonNode) {.gcsafe, raises:[].}
{.push raises:[].} {.push raises:[].}
when defined(resubscribe):
# This is a workaround to manage the 5 minutes limit due to hardhat.
# See https://github.com/NomicFoundation/hardhat/issues/2053#issuecomment-1061374064
proc resubscribeWebsocketEventsOnTimeout*(subscriptions: JsonRpcSubscriptions) {.async.} =
while true:
await sleepAsync(4.int64.minutes)
for id, callback in subscriptions.callbacks:
var newId: JsonNode
if id in subscriptions.logFilters:
let filter = subscriptions.logFilters[id]
newId = await subscriptions.client.eth_subscribe("logs", filter)
subscriptions.logFilters[newId] = filter
subscriptions.logFilters.del(id)
else:
newId = await subscriptions.client.eth_subscribe("newHeads")
subscriptions.callbacks[newId] = callback
subscriptions.callbacks.del(id)
discard await subscriptions.client.eth_unsubscribe(id)
template convertErrorsToSubscriptionError(body) = template convertErrorsToSubscriptionError(body) =
try: try:
body body
@ -58,7 +80,6 @@ func start*(subscriptions: JsonRpcSubscriptions) =
# true = continue processing message using json_rpc's default message handler # true = continue processing message using json_rpc's default message handler
return ok true return ok true
proc setMethodHandler( proc setMethodHandler(
subscriptions: JsonRpcSubscriptions, subscriptions: JsonRpcSubscriptions,
`method`: string, `method`: string,
@ -89,6 +110,10 @@ method close*(subscriptions: JsonRpcSubscriptions) {.async: (raises: [Subscripti
for id in ids: for id in ids:
await subscriptions.unsubscribe(id) await subscriptions.unsubscribe(id)
when defined(resubscribe):
if not subscriptions.resubscribeFut.isNil:
await subscriptions.resubscribeFut.cancelAndWait()
proc getCallback(subscriptions: JsonRpcSubscriptions, proc getCallback(subscriptions: JsonRpcSubscriptions,
id: JsonNode): ?SubscriptionCallback {. raises:[].} = id: JsonNode): ?SubscriptionCallback {. raises:[].} =
try: try:
@ -110,6 +135,10 @@ proc new*(_: type JsonRpcSubscriptions,
if callback =? subscriptions.getCallback(id): if callback =? subscriptions.getCallback(id):
callback(id, success(arguments)) callback(id, success(arguments))
subscriptions.setMethodHandler("eth_subscription", subscriptionHandler) subscriptions.setMethodHandler("eth_subscription", subscriptionHandler)
when defined(resubscribe):
subscriptions.resubscribeFut = resubscribeWebsocketEventsOnTimeout(subscriptions)
subscriptions subscriptions
method subscribeBlocks(subscriptions: WebSocketSubscriptions, method subscribeBlocks(subscriptions: WebSocketSubscriptions,