mirror of https://github.com/waku-org/nwaku.git
refactor(store): HistoryQuery.direction (#2263)
* Fix issue with default history query ascending value in serde operations: Should use the same value. * Update direction types to PagingDirection.
This commit is contained in:
parent
8b37919ee0
commit
fae20bff20
|
@ -9,6 +9,7 @@ import
|
||||||
|
|
||||||
import
|
import
|
||||||
../../../waku/common/databases/db_sqlite,
|
../../../waku/common/databases/db_sqlite,
|
||||||
|
../../../waku/common/paging,
|
||||||
../../../waku/waku_core,
|
../../../waku/waku_core,
|
||||||
../../../waku/waku_core/message/digest,
|
../../../waku/waku_core/message/digest,
|
||||||
../../../waku/waku_archive/driver/sqlite_driver,
|
../../../waku/waku_archive/driver/sqlite_driver,
|
||||||
|
@ -341,7 +342,7 @@ procSuite "Waku Archive - find messages":
|
||||||
## Given
|
## Given
|
||||||
let req = ArchiveQuery(
|
let req = ArchiveQuery(
|
||||||
pageSize: 4,
|
pageSize: 4,
|
||||||
ascending: true
|
direction: PagingDirection.FORWARD
|
||||||
)
|
)
|
||||||
|
|
||||||
## When
|
## When
|
||||||
|
@ -377,7 +378,7 @@ procSuite "Waku Archive - find messages":
|
||||||
## Given
|
## Given
|
||||||
let req = ArchiveQuery(
|
let req = ArchiveQuery(
|
||||||
pageSize: 4,
|
pageSize: 4,
|
||||||
ascending: false # backward
|
direction: PagingDirection.BACKWARD
|
||||||
)
|
)
|
||||||
|
|
||||||
## When
|
## When
|
||||||
|
@ -454,7 +455,7 @@ procSuite "Waku Archive - find messages":
|
||||||
contentTopics: @[ContentTopic("1")],
|
contentTopics: @[ContentTopic("1")],
|
||||||
startTime: some(ts(15, timeOrigin)),
|
startTime: some(ts(15, timeOrigin)),
|
||||||
endTime: some(ts(55, timeOrigin)),
|
endTime: some(ts(55, timeOrigin)),
|
||||||
ascending: true
|
direction: PagingDirection.FORWARD
|
||||||
)
|
)
|
||||||
|
|
||||||
## When
|
## When
|
||||||
|
|
|
@ -6,6 +6,7 @@ import
|
||||||
chronos
|
chronos
|
||||||
import
|
import
|
||||||
../../../waku/common/protobuf,
|
../../../waku/common/protobuf,
|
||||||
|
../../../waku/common/paging,
|
||||||
../../../waku/waku_core,
|
../../../waku/waku_core,
|
||||||
../../../waku/waku_store/rpc,
|
../../../waku/waku_store/rpc,
|
||||||
../../../waku/waku_store/rpc_codec,
|
../../../waku/waku_store/rpc_codec,
|
||||||
|
@ -53,7 +54,7 @@ procSuite "Waku Store - RPC codec":
|
||||||
## Given
|
## Given
|
||||||
let
|
let
|
||||||
index = PagingIndexRPC.compute(fakeWakuMessage(), receivedTime=ts(), pubsubTopic=DefaultPubsubTopic)
|
index = PagingIndexRPC.compute(fakeWakuMessage(), receivedTime=ts(), pubsubTopic=DefaultPubsubTopic)
|
||||||
pagingInfo = PagingInfoRPC(pageSize: some(1'u64), cursor: some(index), direction: some(PagingDirectionRPC.FORWARD))
|
pagingInfo = PagingInfoRPC(pageSize: some(1'u64), cursor: some(index), direction: some(PagingDirection.FORWARD))
|
||||||
|
|
||||||
## When
|
## When
|
||||||
let pb = pagingInfo.encode()
|
let pb = pagingInfo.encode()
|
||||||
|
@ -88,7 +89,7 @@ procSuite "Waku Store - RPC codec":
|
||||||
## Given
|
## Given
|
||||||
let
|
let
|
||||||
index = PagingIndexRPC.compute(fakeWakuMessage(), receivedTime=ts(), pubsubTopic=DefaultPubsubTopic)
|
index = PagingIndexRPC.compute(fakeWakuMessage(), receivedTime=ts(), pubsubTopic=DefaultPubsubTopic)
|
||||||
pagingInfo = PagingInfoRPC(pageSize: some(1'u64), cursor: some(index), direction: some(PagingDirectionRPC.BACKWARD))
|
pagingInfo = PagingInfoRPC(pageSize: some(1'u64), cursor: some(index), direction: some(PagingDirection.BACKWARD))
|
||||||
query = HistoryQueryRPC(
|
query = HistoryQueryRPC(
|
||||||
contentFilters: @[HistoryContentFilterRPC(contentTopic: DefaultContentTopic), HistoryContentFilterRPC(contentTopic: DefaultContentTopic)],
|
contentFilters: @[HistoryContentFilterRPC(contentTopic: DefaultContentTopic), HistoryContentFilterRPC(contentTopic: DefaultContentTopic)],
|
||||||
pagingInfo: some(pagingInfo),
|
pagingInfo: some(pagingInfo),
|
||||||
|
@ -129,7 +130,7 @@ procSuite "Waku Store - RPC codec":
|
||||||
let
|
let
|
||||||
message = fakeWakuMessage()
|
message = fakeWakuMessage()
|
||||||
index = PagingIndexRPC.compute(message, receivedTime=ts(), pubsubTopic=DefaultPubsubTopic)
|
index = PagingIndexRPC.compute(message, receivedTime=ts(), pubsubTopic=DefaultPubsubTopic)
|
||||||
pagingInfo = PagingInfoRPC(pageSize: some(1'u64), cursor: some(index), direction: some(PagingDirectionRPC.BACKWARD))
|
pagingInfo = PagingInfoRPC(pageSize: some(1'u64), cursor: some(index), direction: some(PagingDirection.BACKWARD))
|
||||||
res = HistoryResponseRPC(messages: @[message], pagingInfo: some(pagingInfo), error: HistoryResponseErrorRPC.INVALID_CURSOR)
|
res = HistoryResponseRPC(messages: @[message], pagingInfo: some(pagingInfo), error: HistoryResponseErrorRPC.INVALID_CURSOR)
|
||||||
|
|
||||||
## When
|
## When
|
||||||
|
|
|
@ -9,6 +9,7 @@ import
|
||||||
|
|
||||||
import
|
import
|
||||||
../../../waku/[
|
../../../waku/[
|
||||||
|
common/paging,
|
||||||
node/peer_manager,
|
node/peer_manager,
|
||||||
waku_core,
|
waku_core,
|
||||||
waku_store,
|
waku_store,
|
||||||
|
@ -46,7 +47,7 @@ suite "Waku Store - query handler":
|
||||||
server = await newTestWakuStore(serverSwitch, handler=queryhandler)
|
server = await newTestWakuStore(serverSwitch, handler=queryhandler)
|
||||||
client = newTestWakuStoreClient(clientSwitch)
|
client = newTestWakuStoreClient(clientSwitch)
|
||||||
|
|
||||||
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], ascending: true)
|
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], direction: PagingDirection.FORWARD)
|
||||||
|
|
||||||
## When
|
## When
|
||||||
let queryRes = await client.query(req, peer=serverPeerInfo)
|
let queryRes = await client.query(req, peer=serverPeerInfo)
|
||||||
|
@ -88,7 +89,7 @@ suite "Waku Store - query handler":
|
||||||
server = await newTestWakuStore(serverSwitch, handler=queryhandler)
|
server = await newTestWakuStore(serverSwitch, handler=queryhandler)
|
||||||
client = newTestWakuStoreClient(clientSwitch)
|
client = newTestWakuStoreClient(clientSwitch)
|
||||||
|
|
||||||
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], ascending: true)
|
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], direction: PagingDirection.FORWARD)
|
||||||
|
|
||||||
## When
|
## When
|
||||||
let queryRes = await client.query(req, peer=serverPeerInfo)
|
let queryRes = await client.query(req, peer=serverPeerInfo)
|
||||||
|
|
|
@ -14,6 +14,7 @@ import
|
||||||
libp2p/protocols/pubsub/gossipsub
|
libp2p/protocols/pubsub/gossipsub
|
||||||
import
|
import
|
||||||
../../../waku/common/databases/db_sqlite,
|
../../../waku/common/databases/db_sqlite,
|
||||||
|
../../../waku/common/paging,
|
||||||
../../../waku/waku_core,
|
../../../waku/waku_core,
|
||||||
../../../waku/waku_core/message/digest,
|
../../../waku/waku_core/message/digest,
|
||||||
../../../waku/node/peer_manager,
|
../../../waku/node/peer_manager,
|
||||||
|
@ -107,7 +108,7 @@ procSuite "WakuNode - Store":
|
||||||
client.mountStoreClient()
|
client.mountStoreClient()
|
||||||
|
|
||||||
## Given
|
## Given
|
||||||
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], pageSize: 7, ascending: true)
|
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], pageSize: 7, direction: PagingDirection.FORWARD)
|
||||||
let serverPeer = server.peerInfo.toRemotePeerInfo()
|
let serverPeer = server.peerInfo.toRemotePeerInfo()
|
||||||
|
|
||||||
## When
|
## When
|
||||||
|
@ -158,7 +159,7 @@ procSuite "WakuNode - Store":
|
||||||
client.mountStoreClient()
|
client.mountStoreClient()
|
||||||
|
|
||||||
## Given
|
## Given
|
||||||
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], pageSize: 7, ascending: false)
|
let req = HistoryQuery(contentTopics: @[DefaultContentTopic], pageSize: 7, direction: PagingDirection.BACKWARD)
|
||||||
let serverPeer = server.peerInfo.toRemotePeerInfo()
|
let serverPeer = server.peerInfo.toRemotePeerInfo()
|
||||||
|
|
||||||
## When
|
## When
|
||||||
|
|
|
@ -0,0 +1,35 @@
|
||||||
|
import std/options
|
||||||
|
|
||||||
|
type
|
||||||
|
PagingDirection* {.pure.} = enum
|
||||||
|
## PagingDirection determines the direction of pagination
|
||||||
|
BACKWARD = uint32(0)
|
||||||
|
FORWARD = uint32(1)
|
||||||
|
|
||||||
|
|
||||||
|
proc default*(): PagingDirection {.inline.} =
|
||||||
|
PagingDirection.FORWARD
|
||||||
|
|
||||||
|
|
||||||
|
proc into*(b: bool): PagingDirection =
|
||||||
|
PagingDirection(b)
|
||||||
|
|
||||||
|
|
||||||
|
proc into*(b: Option[bool]): PagingDirection =
|
||||||
|
if b.isNone():
|
||||||
|
return default()
|
||||||
|
b.get().into()
|
||||||
|
|
||||||
|
|
||||||
|
proc into*(d: PagingDirection): bool =
|
||||||
|
d == PagingDirection.FORWARD
|
||||||
|
|
||||||
|
|
||||||
|
proc into*(d: Option[PagingDirection]): bool =
|
||||||
|
if d.isNone():
|
||||||
|
return false
|
||||||
|
d.get().into()
|
||||||
|
|
||||||
|
|
||||||
|
proc into*(s: string): PagingDirection =
|
||||||
|
(s == "true").into()
|
|
@ -765,7 +765,7 @@ proc toArchiveQuery(request: HistoryQuery): ArchiveQuery =
|
||||||
startTime: request.startTime,
|
startTime: request.startTime,
|
||||||
endTime: request.endTime,
|
endTime: request.endTime,
|
||||||
pageSize: request.pageSize.uint,
|
pageSize: request.pageSize.uint,
|
||||||
ascending: request.ascending
|
direction: request.direction
|
||||||
)
|
)
|
||||||
|
|
||||||
# TODO: Review this mapping logic. Maybe, move it to the appplication code
|
# TODO: Review this mapping logic. Maybe, move it to the appplication code
|
||||||
|
|
|
@ -8,11 +8,14 @@ import
|
||||||
chronicles,
|
chronicles,
|
||||||
json_rpc/rpcserver
|
json_rpc/rpcserver
|
||||||
import
|
import
|
||||||
../../../waku_core,
|
../../../[
|
||||||
../../../waku_store,
|
waku_core,
|
||||||
|
waku_store,
|
||||||
|
waku_node
|
||||||
|
],
|
||||||
../../../waku_store/rpc,
|
../../../waku_store/rpc,
|
||||||
../../../waku_node,
|
|
||||||
../../../node/peer_manager,
|
../../../node/peer_manager,
|
||||||
|
../../../common/paging,
|
||||||
./types
|
./types
|
||||||
|
|
||||||
|
|
||||||
|
@ -27,16 +30,14 @@ proc toPagingInfo*(pagingOptions: StorePagingOptions): PagingInfoRPC =
|
||||||
PagingInfoRPC(
|
PagingInfoRPC(
|
||||||
pageSize: some(pagingOptions.pageSize),
|
pageSize: some(pagingOptions.pageSize),
|
||||||
cursor: pagingOptions.cursor,
|
cursor: pagingOptions.cursor,
|
||||||
direction: if pagingOptions.forward: some(PagingDirectionRPC.FORWARD)
|
direction: some(pagingOptions.forward.into())
|
||||||
else: some(PagingDirectionRPC.BACKWARD)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
proc toPagingOptions*(pagingInfo: PagingInfoRPC): StorePagingOptions =
|
proc toPagingOptions*(pagingInfo: PagingInfoRPC): StorePagingOptions =
|
||||||
StorePagingOptions(
|
StorePagingOptions(
|
||||||
pageSize: pagingInfo.pageSize.get(0'u64),
|
pageSize: pagingInfo.pageSize.get(0'u64),
|
||||||
cursor: pagingInfo.cursor,
|
cursor: pagingInfo.cursor,
|
||||||
forward: if pagingInfo.direction.isNone(): true
|
forward: pagingInfo.direction.into()
|
||||||
else: pagingInfo.direction.get() == PagingDirectionRPC.FORWARD
|
|
||||||
)
|
)
|
||||||
|
|
||||||
proc toJsonRPCStoreResponse*(response: HistoryResponse): StoreResponse =
|
proc toJsonRPCStoreResponse*(response: HistoryResponse): StoreResponse =
|
||||||
|
@ -65,8 +66,8 @@ proc installStoreApiHandlers*(node: WakuNode, server: RpcServer) =
|
||||||
contentTopics: contentFiltersOption.get(@[]).mapIt(it.contentTopic),
|
contentTopics: contentFiltersOption.get(@[]).mapIt(it.contentTopic),
|
||||||
startTime: startTime,
|
startTime: startTime,
|
||||||
endTime: endTime,
|
endTime: endTime,
|
||||||
ascending: if pagingOptions.isNone(): true
|
direction: if pagingOptions.isNone(): default()
|
||||||
else: pagingOptions.get().forward,
|
else: pagingOptions.get().forward.into(),
|
||||||
pageSize: if pagingOptions.isNone(): DefaultPageSize
|
pageSize: if pagingOptions.isNone(): DefaultPageSize
|
||||||
else: min(pagingOptions.get().pageSize, MaxPageSize),
|
else: min(pagingOptions.get().pageSize, MaxPageSize),
|
||||||
cursor: if pagingOptions.isNone(): none(HistoryCursor)
|
cursor: if pagingOptions.isNone(): none(HistoryCursor)
|
||||||
|
|
|
@ -15,6 +15,7 @@ import
|
||||||
../../../waku_store/common,
|
../../../waku_store/common,
|
||||||
../../../waku_node,
|
../../../waku_node,
|
||||||
../../../node/peer_manager,
|
../../../node/peer_manager,
|
||||||
|
../../../common/paging,
|
||||||
../../handlers,
|
../../handlers,
|
||||||
../responses,
|
../responses,
|
||||||
../serdes,
|
../serdes,
|
||||||
|
@ -123,7 +124,7 @@ proc createHistoryQuery(pubsubTopic: Option[string],
|
||||||
startTime: Option[string],
|
startTime: Option[string],
|
||||||
endTime: Option[string],
|
endTime: Option[string],
|
||||||
pageSize: Option[string],
|
pageSize: Option[string],
|
||||||
ascending: Option[string]):
|
direction: Option[string]):
|
||||||
|
|
||||||
Result[HistoryQuery, string] =
|
Result[HistoryQuery, string] =
|
||||||
|
|
||||||
|
@ -164,16 +165,16 @@ proc createHistoryQuery(pubsubTopic: Option[string],
|
||||||
let parsedEndTime = ? parseTime(endTime)
|
let parsedEndTime = ? parseTime(endTime)
|
||||||
|
|
||||||
# Parse ascending field
|
# Parse ascending field
|
||||||
var parsedAscending = true
|
var parsedDirection = default()
|
||||||
if ascending.isSome() and ascending.get() != "":
|
if direction.isSome() and direction.get() != "":
|
||||||
parsedAscending = ascending.get() == "true"
|
parsedDirection = direction.get().into()
|
||||||
|
|
||||||
return ok(
|
return ok(
|
||||||
HistoryQuery(pubsubTopic: parsedPubsubTopic,
|
HistoryQuery(pubsubTopic: parsedPubsubTopic,
|
||||||
contentTopics: parsedContentTopics,
|
contentTopics: parsedContentTopics,
|
||||||
startTime: parsedStartTime,
|
startTime: parsedStartTime,
|
||||||
endTime: parsedEndTime,
|
endTime: parsedEndTime,
|
||||||
ascending: parsedAscending,
|
direction: parsedDirection,
|
||||||
pageSize: parsedPagedSize,
|
pageSize: parsedPagedSize,
|
||||||
cursor: parsedCursor
|
cursor: parsedCursor
|
||||||
))
|
))
|
||||||
|
|
|
@ -124,7 +124,8 @@ paths:
|
||||||
schema:
|
schema:
|
||||||
type: string
|
type: string
|
||||||
description: >
|
description: >
|
||||||
"true" for paging forward, "false" for paging backward
|
"true" for paging forward, "false" for paging backward.
|
||||||
|
If not specified or if specified with an invalid value, the default is "true".
|
||||||
example: "true"
|
example: "true"
|
||||||
|
|
||||||
responses:
|
responses:
|
||||||
|
|
|
@ -11,8 +11,11 @@ import
|
||||||
regex,
|
regex,
|
||||||
metrics
|
metrics
|
||||||
import
|
import
|
||||||
../common/databases/dburl,
|
../common/[
|
||||||
../common/databases/db_sqlite,
|
databases/dburl,
|
||||||
|
databases/db_sqlite,
|
||||||
|
paging
|
||||||
|
],
|
||||||
./driver,
|
./driver,
|
||||||
./retention_policy,
|
./retention_policy,
|
||||||
./retention_policy/retention_policy_capacity,
|
./retention_policy/retention_policy_capacity,
|
||||||
|
@ -131,7 +134,7 @@ proc findMessages*(w: WakuArchive, query: ArchiveQuery): Future[ArchiveResult] {
|
||||||
qEndTime = query.endTime
|
qEndTime = query.endTime
|
||||||
qMaxPageSize = if query.pageSize <= 0: DefaultPageSize
|
qMaxPageSize = if query.pageSize <= 0: DefaultPageSize
|
||||||
else: min(query.pageSize, MaxPageSize)
|
else: min(query.pageSize, MaxPageSize)
|
||||||
qAscendingOrder = query.ascending
|
isAscendingOrder = query.direction.into()
|
||||||
|
|
||||||
if qContentTopics.len > 10:
|
if qContentTopics.len > 10:
|
||||||
return err(ArchiveError.invalidQuery("too many content topics"))
|
return err(ArchiveError.invalidQuery("too many content topics"))
|
||||||
|
@ -145,7 +148,7 @@ proc findMessages*(w: WakuArchive, query: ArchiveQuery): Future[ArchiveResult] {
|
||||||
startTime = qStartTime,
|
startTime = qStartTime,
|
||||||
endTime = qEndTime,
|
endTime = qEndTime,
|
||||||
maxPageSize = qMaxPageSize + 1,
|
maxPageSize = qMaxPageSize + 1,
|
||||||
ascendingOrder = qAscendingOrder
|
ascendingOrder = isAscendingOrder
|
||||||
)
|
)
|
||||||
|
|
||||||
let queryDuration = getTime().toUnixFloat() - queryStartTime
|
let queryDuration = getTime().toUnixFloat() - queryStartTime
|
||||||
|
@ -188,7 +191,7 @@ proc findMessages*(w: WakuArchive, query: ArchiveQuery): Future[ArchiveResult] {
|
||||||
))
|
))
|
||||||
|
|
||||||
# All messages MUST be returned in chronological order
|
# All messages MUST be returned in chronological order
|
||||||
if not qAscendingOrder:
|
if not isAscendingOrder:
|
||||||
reverse(messages)
|
reverse(messages)
|
||||||
|
|
||||||
return ok(ArchiveResponse(messages: messages, cursor: cursor))
|
return ok(ArchiveResponse(messages: messages, cursor: cursor))
|
||||||
|
|
|
@ -9,7 +9,8 @@ import
|
||||||
stew/byteutils,
|
stew/byteutils,
|
||||||
nimcrypto/sha2
|
nimcrypto/sha2
|
||||||
import
|
import
|
||||||
../waku_core
|
../waku_core,
|
||||||
|
../common/paging
|
||||||
|
|
||||||
|
|
||||||
## Waku message digest
|
## Waku message digest
|
||||||
|
@ -54,7 +55,7 @@ type
|
||||||
startTime*: Option[Timestamp]
|
startTime*: Option[Timestamp]
|
||||||
endTime*: Option[Timestamp]
|
endTime*: Option[Timestamp]
|
||||||
pageSize*: uint
|
pageSize*: uint
|
||||||
ascending*: bool
|
direction*: PagingDirection
|
||||||
|
|
||||||
ArchiveResponse* = object
|
ArchiveResponse* = object
|
||||||
messages*: seq[WakuMessage]
|
messages*: seq[WakuMessage]
|
||||||
|
|
|
@ -199,7 +199,7 @@ when defined(waku_exp_store_resume):
|
||||||
startTime: some(queryStartTime),
|
startTime: some(queryStartTime),
|
||||||
endTime: some(queryEndTime),
|
endTime: some(queryEndTime),
|
||||||
pageSize: uint64(pageSize),
|
pageSize: uint64(pageSize),
|
||||||
ascending: true
|
direction: default()
|
||||||
)
|
)
|
||||||
|
|
||||||
var res: WakuStoreResult[seq[WakuMessage]]
|
var res: WakuStoreResult[seq[WakuMessage]]
|
||||||
|
|
|
@ -9,7 +9,8 @@ import
|
||||||
stew/byteutils,
|
stew/byteutils,
|
||||||
nimcrypto/sha2
|
nimcrypto/sha2
|
||||||
import
|
import
|
||||||
../waku_core
|
../waku_core,
|
||||||
|
../common/paging
|
||||||
|
|
||||||
|
|
||||||
const
|
const
|
||||||
|
@ -55,7 +56,7 @@ type
|
||||||
startTime*: Option[Timestamp]
|
startTime*: Option[Timestamp]
|
||||||
endTime*: Option[Timestamp]
|
endTime*: Option[Timestamp]
|
||||||
pageSize*: uint64
|
pageSize*: uint64
|
||||||
ascending*: bool
|
direction*: PagingDirection
|
||||||
|
|
||||||
HistoryResponse* = object
|
HistoryResponse* = object
|
||||||
messages*: seq[WakuMessage]
|
messages*: seq[WakuMessage]
|
||||||
|
|
|
@ -8,11 +8,14 @@ import
|
||||||
stew/results
|
stew/results
|
||||||
import
|
import
|
||||||
../waku_core,
|
../waku_core,
|
||||||
|
../common/paging,
|
||||||
./common
|
./common
|
||||||
|
|
||||||
|
|
||||||
## Wire protocol
|
## Wire protocol
|
||||||
|
|
||||||
|
const HistoryQueryDirectionDefaultValue = default(type HistoryQuery.direction)
|
||||||
|
|
||||||
type PagingIndexRPC* = object
|
type PagingIndexRPC* = object
|
||||||
## This type contains the description of an Index used in the pagination of WakuMessages
|
## This type contains the description of an Index used in the pagination of WakuMessages
|
||||||
pubsubTopic*: PubsubTopic
|
pubsubTopic*: PubsubTopic
|
||||||
|
@ -41,16 +44,11 @@ proc compute*(T: type PagingIndexRPC, msg: WakuMessage, receivedTime: Timestamp,
|
||||||
|
|
||||||
|
|
||||||
type
|
type
|
||||||
PagingDirectionRPC* {.pure.} = enum
|
|
||||||
## PagingDirection determines the direction of pagination
|
|
||||||
BACKWARD = uint32(0)
|
|
||||||
FORWARD = uint32(1)
|
|
||||||
|
|
||||||
PagingInfoRPC* = object
|
PagingInfoRPC* = object
|
||||||
## This type holds the information needed for the pagination
|
## This type holds the information needed for the pagination
|
||||||
pageSize*: Option[uint64]
|
pageSize*: Option[uint64]
|
||||||
cursor*: Option[PagingIndexRPC]
|
cursor*: Option[PagingIndexRPC]
|
||||||
direction*: Option[PagingDirectionRPC]
|
direction*: Option[PagingDirection]
|
||||||
|
|
||||||
|
|
||||||
type
|
type
|
||||||
|
@ -120,14 +118,14 @@ proc toRPC*(query: HistoryQuery): HistoryQueryRPC =
|
||||||
rpc.pagingInfo = block:
|
rpc.pagingInfo = block:
|
||||||
if query.cursor.isNone() and
|
if query.cursor.isNone() and
|
||||||
query.pageSize == default(type query.pageSize) and
|
query.pageSize == default(type query.pageSize) and
|
||||||
query.ascending == default(type query.ascending):
|
query.direction == HistoryQueryDirectionDefaultValue:
|
||||||
none(PagingInfoRPC)
|
none(PagingInfoRPC)
|
||||||
else:
|
else:
|
||||||
let
|
let
|
||||||
pageSize = some(query.pageSize)
|
pageSize = some(query.pageSize)
|
||||||
cursor = query.cursor.map(toRPC)
|
cursor = query.cursor.map(toRPC)
|
||||||
direction = if query.ascending: some(PagingDirectionRPC.FORWARD)
|
direction = some(query.direction)
|
||||||
else: some(PagingDirectionRPC.BACKWARD)
|
|
||||||
some(PagingInfoRPC(
|
some(PagingInfoRPC(
|
||||||
pageSize: pageSize,
|
pageSize: pageSize,
|
||||||
cursor: cursor,
|
cursor: cursor,
|
||||||
|
@ -156,8 +154,8 @@ proc toAPI*(rpc: HistoryQueryRPC): HistoryQuery =
|
||||||
pageSize = if rpc.pagingInfo.isNone() or rpc.pagingInfo.get().pageSize.isNone(): 0'u64
|
pageSize = if rpc.pagingInfo.isNone() or rpc.pagingInfo.get().pageSize.isNone(): 0'u64
|
||||||
else: rpc.pagingInfo.get().pageSize.get()
|
else: rpc.pagingInfo.get().pageSize.get()
|
||||||
|
|
||||||
ascending = if rpc.pagingInfo.isNone() or rpc.pagingInfo.get().direction.isNone(): true
|
direction = if rpc.pagingInfo.isNone() or rpc.pagingInfo.get().direction.isNone(): HistoryQueryDirectionDefaultValue
|
||||||
else: rpc.pagingInfo.get().direction.get() == PagingDirectionRPC.FORWARD
|
else: rpc.pagingInfo.get().direction.get()
|
||||||
|
|
||||||
HistoryQuery(
|
HistoryQuery(
|
||||||
pubsubTopic: pubsubTopic,
|
pubsubTopic: pubsubTopic,
|
||||||
|
@ -166,7 +164,7 @@ proc toAPI*(rpc: HistoryQueryRPC): HistoryQuery =
|
||||||
startTime: startTime,
|
startTime: startTime,
|
||||||
endTime: endTime,
|
endTime: endTime,
|
||||||
pageSize: pageSize,
|
pageSize: pageSize,
|
||||||
ascending: ascending
|
direction: direction
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -7,7 +7,7 @@ import
|
||||||
std/options,
|
std/options,
|
||||||
nimcrypto/hash
|
nimcrypto/hash
|
||||||
import
|
import
|
||||||
../common/protobuf,
|
../common/[protobuf, paging],
|
||||||
../waku_core,
|
../waku_core,
|
||||||
./common,
|
./common,
|
||||||
./rpc
|
./rpc
|
||||||
|
@ -74,7 +74,7 @@ proc encode*(rpc: PagingInfoRPC): ProtoBuffer =
|
||||||
|
|
||||||
pb.write3(1, rpc.pageSize)
|
pb.write3(1, rpc.pageSize)
|
||||||
pb.write3(2, rpc.cursor.map(encode))
|
pb.write3(2, rpc.cursor.map(encode))
|
||||||
pb.write3(3, rpc.direction.map(proc(d: PagingDirectionRPC): uint32 = uint32(ord(d))))
|
pb.write3(3, rpc.direction.map(proc(d: PagingDirection): uint32 = uint32(ord(d))))
|
||||||
pb.finish3()
|
pb.finish3()
|
||||||
|
|
||||||
pb
|
pb
|
||||||
|
@ -99,9 +99,9 @@ proc decode*(T: type PagingInfoRPC, buffer: seq[byte]): ProtobufResult[T] =
|
||||||
|
|
||||||
var direction: uint32
|
var direction: uint32
|
||||||
if not ?pb.getField(3, direction):
|
if not ?pb.getField(3, direction):
|
||||||
rpc.direction = none(PagingDirectionRPC)
|
rpc.direction = none(PagingDirection)
|
||||||
else:
|
else:
|
||||||
rpc.direction = some(PagingDirectionRPC(direction))
|
rpc.direction = some(PagingDirection(direction))
|
||||||
|
|
||||||
ok(rpc)
|
ok(rpc)
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue