mirror of
https://github.com/waku-org/nwaku.git
synced 2025-01-15 17:35:45 +00:00
2cc86c51da
* feat: Waku Sync Protocol * feat: state machine (#2656) * feat: pruning storage mehcanism (#2673) * feat: message transfer mechanism & tests (#2688) * update docker files * added ENR filed for sync & misc. fixes * adding new sync range param & fixes --------- Co-authored-by: Ivan FB <128452529+Ivansete-status@users.noreply.github.com> Co-authored-by: Prem Chaitanya Prathi <chaitanyaprem@gmail.com>
62 lines
2.1 KiB
Nim
62 lines
2.1 KiB
Nim
{.push raises: [].}
|
|
|
|
import std/options, results, chronicles, chronos, metrics, bearssl/rand
|
|
import
|
|
../node/peer_manager, ../utils/requests, ./protocol_metrics, ./common, ./rpc_codec
|
|
|
|
logScope:
|
|
topics = "waku store client"
|
|
|
|
const DefaultPageSize*: uint = 20
|
|
# A recommended default number of waku messages per page
|
|
|
|
type WakuStoreClient* = ref object
|
|
peerManager: PeerManager
|
|
rng: ref rand.HmacDrbgContext
|
|
|
|
proc new*(
|
|
T: type WakuStoreClient, peerManager: PeerManager, rng: ref rand.HmacDrbgContext
|
|
): T {.gcsafe.} =
|
|
WakuStoreClient(peerManager: peerManager, rng: rng)
|
|
|
|
proc sendStoreRequest(
|
|
self: WakuStoreClient, request: StoreQueryRequest, connection: Connection
|
|
): Future[StoreQueryResult] {.async, gcsafe.} =
|
|
var req = request
|
|
|
|
req.requestId = generateRequestId(self.rng)
|
|
|
|
let writeRes = catch:
|
|
await connection.writeLP(req.encode().buffer)
|
|
if writeRes.isErr():
|
|
return err(StoreError(kind: ErrorCode.BAD_REQUEST, cause: writeRes.error.msg))
|
|
|
|
let readRes = catch:
|
|
await connection.readLp(DefaultMaxRpcSize.int)
|
|
|
|
let buf = readRes.valueOr:
|
|
return err(StoreError(kind: ErrorCode.BAD_RESPONSE, cause: error.msg))
|
|
|
|
let res = StoreQueryResponse.decode(buf).valueOr:
|
|
waku_store_errors.inc(labelValues = [decodeRpcFailure])
|
|
return err(StoreError(kind: ErrorCode.BAD_RESPONSE, cause: decodeRpcFailure))
|
|
|
|
if res.statusCode != uint32(StatusCode.SUCCESS):
|
|
waku_store_errors.inc(labelValues = [res.statusDesc])
|
|
return err(StoreError.new(res.statusCode, res.statusDesc))
|
|
|
|
return ok(res)
|
|
|
|
proc query*(
|
|
self: WakuStoreClient, request: StoreQueryRequest, peer: RemotePeerInfo | PeerId
|
|
): Future[StoreQueryResult] {.async, gcsafe.} =
|
|
if request.paginationCursor.isSome() and request.paginationCursor.get() == EmptyCursor:
|
|
return err(StoreError(kind: ErrorCode.BAD_REQUEST, cause: "invalid cursor"))
|
|
|
|
let connection = (await self.peerManager.dialPeer(peer, WakuStoreCodec)).valueOr:
|
|
waku_store_errors.inc(labelValues = [dialFailure])
|
|
|
|
return err(StoreError(kind: ErrorCode.PEER_DIAL_FAILURE, address: $peer))
|
|
|
|
return await self.sendStoreRequest(request, connection)
|