mirror of
https://github.com/waku-org/nwaku.git
synced 2025-01-13 16:25:00 +00:00
Pagination feature/indexing waku messages (#233)
* changes the digest type to MDigest[256] and modifies the computeIndex * fixes formatting issue * adds the pagination with its tests stores and retrieves IndexedWakuMessage adds the paginate proc adds the paginate function fixes some formatting issues minor edits indentation and fixes a bug removes unused imports minor fixes indentations and adds a new testcase adds indexedWakuMessageComparison adds `==` proc for IndexedWakuMessage separates the comparison of index and indexed waku messages adds testcases for the Index comparison and IndexedWakuMessage comparison WIP WIP: adds an decoder for Index removes an unnecessary imports WIP adds findIndex() proc removes the equality check '==' for IndexedWakuMessages edits the code format and adds the pagination test edits paginate() proc to work on a copy of the input list deletes unnecessary echo adds the boundary calculations for forward and backward pagination adds test cases for the page boundaries tests corner cases for the queried cursor and pagesize minor adds some comments adds a proc to extract WakuMessages from a list of IndexedWakuMessages integrates pagination into the findMessages proc adds some comments changes paginate to paginateWithIndex removes some echos modifies paginateWithIndex to handle invalid cursors adds test case for an invalid cursor WIP: adds a `$` proc for IndexedWakuMessages adds some debugging message prints adds an integration test for handling query with pagination * fixes a type mismatch issue in the min proc * replaces boolean direction with their enums and updates contentTopics * adds the unit test for the sorting of the indexed waku messages * fixes a flaky test * fixes a flaky test * removes index equality check proc * handles an initial query with an empty cursor * adds test for the initial query * adds integration test for pagination * adds a test for empty message list * adds comments and fixes an issue * adds comments * code cleanup * removes the content topic validation check * resolves the errors related to the windows CI tests * Update waku/protocol/v2/waku_store.nim Co-authored-by: Oskar Thorén <ot@oskarthoren.com> * Update waku/protocol/v2/waku_store.nim Co-authored-by: Oskar Thorén <ot@oskarthoren.com> * Update tests/v2/test_waku_pagination.nim Co-authored-by: Oskar Thorén <ot@oskarthoren.com> * Update waku/protocol/v2/waku_store.nim Co-authored-by: Oskar Thorén <ot@oskarthoren.com> * Update waku/protocol/v2/waku_store.nim Co-authored-by: Oskar Thorén <ot@oskarthoren.com> * Update waku/protocol/v2/waku_store.nim Co-authored-by: Oskar Thorén <ot@oskarthoren.com> * changes the output type of findIndex to Option * Update tests/v2/test_waku_pagination.nim Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com> * Update tests/v2/test_waku_pagination.nim Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com> * Update tests/v2/test_waku_pagination.nim Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com> * Apply suggestions from code review Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com> * adds some comments * fixes an indentation issue * some code modification for array initialization * Apply suggestions from code review Co-authored-by: Oskar Thorén <ot@oskarthoren.com> Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com> * does some code reorganizations and clean up * CreateSampleList to createSampleList * replaces a byte array literal initialization with a for loop * relocates indexedWakuMessageComparison and indexComparison * minor Co-authored-by: Oskar Thorén <ot@oskarthoren.com> Co-authored-by: Dean Eigenmann <7621705+decanus@users.noreply.github.com>
This commit is contained in:
Sanaz Taheri Boshrooyeh
GitHub
parent
9df31621c4
commit
ea5f9993a7
@ -1,11 +1,21 @@
|
||||
{.used.}
|
||||
import
|
||||
std/unittest,
|
||||
std/[unittest,algorithm,options],
|
||||
nimcrypto/sha2,
|
||||
../../waku/node/v2/waku_types,
|
||||
../../waku/protocol/v2/waku_store,
|
||||
../test_helpers
|
||||
|
||||
|
||||
proc createSampleList(s: int): seq[IndexedWakuMessage] =
|
||||
## takes s as input and outputs a sequence with s amount of IndexedWakuMessage
|
||||
var data {.noinit.}: array[32, byte]
|
||||
for x in data.mitems: x = 1
|
||||
for i in 0..<s:
|
||||
result.add(IndexedWakuMessage(msg: WakuMessage(payload: @[byte i]), index: Index(receivedTime: float64(i), digest: MDigest[256](data: data)) ))
|
||||
|
||||
procSuite "pagination":
|
||||
test "computeIndex: empty contentTopic test":
|
||||
test "Index computation test":
|
||||
let
|
||||
wm = WakuMessage(payload: @[byte 1, 2, 3])
|
||||
index = wm.computeIndex()
|
||||
@ -15,10 +25,9 @@ procSuite "pagination":
|
||||
len(index.digest.data) == 32 # sha2 output length in bytes
|
||||
index.receivedTime != 0 # the timestamp should be a non-zero value
|
||||
|
||||
test "computeIndex: identical WakuMessages test":
|
||||
let
|
||||
wm = WakuMessage(payload: @[byte 1, 2, 3], contentTopic: ContentTopic(1))
|
||||
index1 = wm.computeIndex()
|
||||
wm1 = WakuMessage(payload: @[byte 1, 2, 3], contentTopic: ContentTopic(1))
|
||||
index1 = wm1.computeIndex()
|
||||
wm2 = WakuMessage(payload: @[byte 1, 2, 3], contentTopic: ContentTopic(1))
|
||||
index2 = wm2.computeIndex()
|
||||
|
||||
@ -26,3 +35,183 @@ procSuite "pagination":
|
||||
# the digests of two identical WakuMessages must be the same
|
||||
index1.digest == index2.digest
|
||||
|
||||
test "Index comparison, IndexedWakuMessage comparison, and Sorting tests":
|
||||
var data1 {.noinit.}: array[32, byte]
|
||||
for x in data1.mitems: x = 1
|
||||
var data2 {.noinit.}: array[32, byte]
|
||||
for x in data2.mitems: x = 2
|
||||
var data3 {.noinit.}: array[32, byte]
|
||||
for x in data3.mitems: x = 3
|
||||
|
||||
let
|
||||
index1 = Index(receivedTime: 1, digest: MDigest[256](data: data1))
|
||||
index2 = Index(receivedTime: 1, digest: MDigest[256](data: data2))
|
||||
index3 = Index(receivedTime: 2, digest: MDigest[256](data: data3))
|
||||
iwm1 = IndexedWakuMessage(index: index1)
|
||||
iwm2 = IndexedWakuMessage(index: index2)
|
||||
iwm3 = IndexedWakuMessage(index: index3)
|
||||
|
||||
check:
|
||||
indexComparison(index1, index1) == 0
|
||||
indexComparison(index1, index2) == -1
|
||||
indexComparison(index2, index1) == 1
|
||||
indexComparison(index1, index3) == -1
|
||||
indexComparison(index3, index1) == 1
|
||||
|
||||
check:
|
||||
indexedWakuMessageComparison(iwm1, iwm1) == 0
|
||||
indexedWakuMessageComparison(iwm1, iwm2) == -1
|
||||
indexedWakuMessageComparison(iwm2, iwm1) == 1
|
||||
indexedWakuMessageComparison(iwm1, iwm3) == -1
|
||||
indexedWakuMessageComparison(iwm3, iwm1) == 1
|
||||
|
||||
var sortingList = @[iwm3, iwm1, iwm2]
|
||||
sortingList.sort(indexedWakuMessageComparison)
|
||||
check:
|
||||
sortingList[0] == iwm1
|
||||
sortingList[1] == iwm2
|
||||
sortingList[2] == iwm3
|
||||
|
||||
|
||||
test "Find Index test":
|
||||
let msgList = createSampleList(10)
|
||||
check:
|
||||
msgList.findIndex(msgList[3].index).get() == 3
|
||||
msgList.findIndex(Index()).isNone == true
|
||||
|
||||
test "Forward pagination test":
|
||||
var
|
||||
msgList = createSampleList(10)
|
||||
pagingInfo = PagingInfo(pageSize: 2, cursor: msgList[3].index, direction: PagingDirection.FORWARD)
|
||||
|
||||
# test for a normal pagination
|
||||
var (data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 2
|
||||
data == msgList[4..5]
|
||||
newPagingInfo.cursor == msgList[5].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == pagingInfo.pageSize
|
||||
|
||||
# test for an initial pagination request with an empty cursor
|
||||
pagingInfo = PagingInfo(pageSize: 2, direction: PagingDirection.FORWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 2
|
||||
data == msgList[0..1]
|
||||
newPagingInfo.cursor == msgList[1].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 2
|
||||
|
||||
# test for an empty msgList
|
||||
pagingInfo = PagingInfo(pageSize: 2, direction: PagingDirection.FORWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(@[], pagingInfo)
|
||||
check:
|
||||
data.len == 0
|
||||
newPagingInfo.pageSize == 0
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.cursor == pagingInfo.cursor
|
||||
|
||||
# test for a page size larger than the remaining messages
|
||||
pagingInfo = PagingInfo(pageSize: 10, cursor: msgList[3].index, direction: PagingDirection.FORWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 6
|
||||
data == msgList[4..9]
|
||||
newPagingInfo.cursor == msgList[9].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 6
|
||||
|
||||
# test for a page size larger than the maximum allowed page size
|
||||
pagingInfo = PagingInfo(pageSize: MaxPageSize+1, cursor: msgList[3].index, direction: PagingDirection.FORWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len <= MaxPageSize
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize <= MaxPageSize
|
||||
|
||||
# test for a cursor poiting to the end of the message list
|
||||
pagingInfo = PagingInfo(pageSize: 10, cursor: msgList[9].index, direction: PagingDirection.FORWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 0
|
||||
newPagingInfo.cursor == msgList[9].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 0
|
||||
|
||||
# test for an invalid cursor
|
||||
pagingInfo = PagingInfo(pageSize: 10, cursor: computeIndex(WakuMessage(payload: @[byte 10])), direction: PagingDirection.FORWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 0
|
||||
newPagingInfo.cursor == pagingInfo.cursor
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 0
|
||||
|
||||
test "Backward pagination test":
|
||||
var
|
||||
msgList = createSampleList(10)
|
||||
pagingInfo = PagingInfo(pageSize: 2, cursor: msgList[3].index, direction: PagingDirection.BACKWARD)
|
||||
|
||||
# test for a normal pagination
|
||||
var (data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data == msgList[1..2]
|
||||
newPagingInfo.cursor == msgList[1].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == pagingInfo.pageSize
|
||||
|
||||
# test for an empty msgList
|
||||
pagingInfo = PagingInfo(pageSize: 2, direction: PagingDirection.BACKWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(@[], pagingInfo)
|
||||
check:
|
||||
data.len == 0
|
||||
newPagingInfo.pageSize == 0
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.cursor == pagingInfo.cursor
|
||||
|
||||
# test for an initial pagination request with an empty cursor
|
||||
pagingInfo = PagingInfo(pageSize: 2, direction: PagingDirection.BACKWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 2
|
||||
data == msgList[8..9]
|
||||
newPagingInfo.cursor == msgList[8].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 2
|
||||
|
||||
|
||||
# test for a page size larger than the remaining messages
|
||||
pagingInfo = PagingInfo(pageSize: 5, cursor: msgList[3].index, direction: PagingDirection.BACKWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data == msgList[0..2]
|
||||
newPagingInfo.cursor == msgList[0].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 3
|
||||
|
||||
# test for a page size larger than the Maximum allowed page size
|
||||
pagingInfo = PagingInfo(pageSize: MaxPageSize+1, cursor: msgList[3].index, direction: PagingDirection.BACKWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len <= MaxPageSize
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize <= MaxPageSize
|
||||
|
||||
# test for a cursor pointing to the begining of the message list
|
||||
pagingInfo = PagingInfo(pageSize: 5, cursor: msgList[0].index, direction: PagingDirection.BACKWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 0
|
||||
newPagingInfo.cursor == msgList[0].index
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 0
|
||||
|
||||
# test for an invalid cursor
|
||||
pagingInfo = PagingInfo(pageSize: 5, cursor: computeIndex(WakuMessage(payload: @[byte 10])), direction: PagingDirection.BACKWARD)
|
||||
(data, newPagingInfo) = paginateWithIndex(msgList, pagingInfo)
|
||||
check:
|
||||
data.len == 0
|
||||
newPagingInfo.cursor == pagingInfo.cursor
|
||||
newPagingInfo.direction == pagingInfo.direction
|
||||
newPagingInfo.pageSize == 0
|
||||
|
||||
@ -15,6 +15,7 @@ import
|
||||
../../waku/node/v2/waku_types,
|
||||
../test_helpers, ./utils
|
||||
|
||||
|
||||
procSuite "Waku Store":
|
||||
asyncTest "handle query":
|
||||
let
|
||||
@ -57,6 +58,157 @@ procSuite "Waku Store":
|
||||
|
||||
check:
|
||||
(await completionFut.withTimeout(5.seconds)) == true
|
||||
|
||||
asyncTest "handle query with forward pagination":
|
||||
let
|
||||
key = PrivateKey.random(ECDSA, rng[]).get()
|
||||
peer = PeerInfo.init(key)
|
||||
var
|
||||
msgList = @[WakuMessage(payload: @[byte 0], contentTopic: ContentTopic(2)),
|
||||
WakuMessage(payload: @[byte 1],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 2],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 3],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 4],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 5],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 6],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 7],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 8],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 9],contentTopic: ContentTopic(2))]
|
||||
|
||||
var dialSwitch = newStandardSwitch()
|
||||
discard await dialSwitch.start()
|
||||
|
||||
var listenSwitch = newStandardSwitch(some(key))
|
||||
discard await listenSwitch.start()
|
||||
|
||||
let
|
||||
proto = WakuStore.init(dialSwitch, crypto.newRng())
|
||||
subscription = proto.subscription()
|
||||
rpc = HistoryQuery(topics: @[ContentTopic(1)], pagingInfo: PagingInfo(pageSize: 2, direction: PagingDirection.FORWARD) )
|
||||
|
||||
proto.setPeer(listenSwitch.peerInfo)
|
||||
|
||||
var subscriptions = newTable[string, MessageNotificationSubscription]()
|
||||
subscriptions["test"] = subscription
|
||||
|
||||
listenSwitch.mount(proto)
|
||||
|
||||
for wakuMsg in msgList:
|
||||
await subscriptions.notify("foo", wakuMsg)
|
||||
|
||||
var completionFut = newFuture[bool]()
|
||||
|
||||
proc handler(response: HistoryResponse) {.gcsafe, closure.} =
|
||||
check:
|
||||
response.messages.len() == 2
|
||||
response.pagingInfo.pageSize == 2
|
||||
response.pagingInfo.direction == PagingDirection.FORWARD
|
||||
response.pagingInfo.cursor != Index()
|
||||
completionFut.complete(true)
|
||||
|
||||
await proto.query(rpc, handler)
|
||||
|
||||
check:
|
||||
(await completionFut.withTimeout(5.seconds)) == true
|
||||
|
||||
asyncTest "handle query with backward pagination":
|
||||
let
|
||||
key = PrivateKey.random(ECDSA, rng[]).get()
|
||||
peer = PeerInfo.init(key)
|
||||
var
|
||||
msgList = @[WakuMessage(payload: @[byte 0], contentTopic: ContentTopic(2)),
|
||||
WakuMessage(payload: @[byte 1],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 2],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 3],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 4],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 5],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 6],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 7],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 8],contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 9],contentTopic: ContentTopic(2))]
|
||||
|
||||
var dialSwitch = newStandardSwitch()
|
||||
discard await dialSwitch.start()
|
||||
|
||||
var listenSwitch = newStandardSwitch(some(key))
|
||||
discard await listenSwitch.start()
|
||||
|
||||
let
|
||||
proto = WakuStore.init(dialSwitch, crypto.newRng())
|
||||
subscription = proto.subscription()
|
||||
proto.setPeer(listenSwitch.peerInfo)
|
||||
|
||||
var subscriptions = newTable[string, MessageNotificationSubscription]()
|
||||
subscriptions["test"] = subscription
|
||||
|
||||
listenSwitch.mount(proto)
|
||||
|
||||
for wakuMsg in msgList:
|
||||
await subscriptions.notify("foo", wakuMsg)
|
||||
var completionFut = newFuture[bool]()
|
||||
|
||||
proc handler(response: HistoryResponse) {.gcsafe, closure.} =
|
||||
check:
|
||||
response.messages.len() == 2
|
||||
response.pagingInfo.pageSize == 2
|
||||
response.pagingInfo.direction == PagingDirection.BACKWARD
|
||||
response.pagingInfo.cursor != Index()
|
||||
completionFut.complete(true)
|
||||
|
||||
let rpc = HistoryQuery(topics: @[ContentTopic(1)], pagingInfo: PagingInfo(pageSize: 2, direction: PagingDirection.BACKWARD) )
|
||||
await proto.query(rpc, handler)
|
||||
|
||||
check:
|
||||
(await completionFut.withTimeout(5.seconds)) == true
|
||||
|
||||
asyncTest "handle queries with no pagination":
|
||||
let
|
||||
key = PrivateKey.random(ECDSA, rng[]).get()
|
||||
peer = PeerInfo.init(key)
|
||||
var
|
||||
msgList = @[WakuMessage(payload: @[byte 0], contentTopic: ContentTopic(2)),
|
||||
WakuMessage(payload: @[byte 1], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 2], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 3], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 4], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 5], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 6], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 7], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 8], contentTopic: ContentTopic(1)),
|
||||
WakuMessage(payload: @[byte 9], contentTopic: ContentTopic(2))]
|
||||
|
||||
var dialSwitch = newStandardSwitch()
|
||||
discard await dialSwitch.start()
|
||||
|
||||
var listenSwitch = newStandardSwitch(some(key))
|
||||
discard await listenSwitch.start()
|
||||
|
||||
let
|
||||
proto = WakuStore.init(dialSwitch, crypto.newRng())
|
||||
subscription = proto.subscription()
|
||||
proto.setPeer(listenSwitch.peerInfo)
|
||||
|
||||
var subscriptions = newTable[string, MessageNotificationSubscription]()
|
||||
subscriptions["test"] = subscription
|
||||
|
||||
listenSwitch.mount(proto)
|
||||
|
||||
for wakuMsg in msgList:
|
||||
await subscriptions.notify("foo", wakuMsg)
|
||||
var completionFut = newFuture[bool]()
|
||||
|
||||
proc handler(response: HistoryResponse) {.gcsafe, closure.} =
|
||||
check:
|
||||
response.messages.len() == 8
|
||||
response.pagingInfo == PagingInfo()
|
||||
completionFut.complete(true)
|
||||
|
||||
let rpc = HistoryQuery(topics: @[ContentTopic(1)] )
|
||||
|
||||
await proto.query(rpc, handler)
|
||||
|
||||
check:
|
||||
(await completionFut.withTimeout(5.seconds)) == true
|
||||
|
||||
test "Index Protobuf encoder/decoder test":
|
||||
let
|
||||
|
||||
@ -12,9 +12,9 @@ import
|
||||
libp2p/stream/connection,
|
||||
libp2p/protocols/pubsub/[pubsub, gossipsub],
|
||||
nimcrypto/sha2
|
||||
|
||||
# constants required for pagination -------------------------------------------
|
||||
const MaxPageSize* = 100 # Maximum number of waku messages in each page
|
||||
# Common data types -----------------------------------------------------------
|
||||
|
||||
type
|
||||
ContentTopic* = uint32
|
||||
|
||||
@ -61,11 +61,11 @@ type
|
||||
|
||||
HistoryQuery* = object
|
||||
topics*: seq[ContentTopic]
|
||||
pagingInfo*: PagingInfo
|
||||
pagingInfo*: PagingInfo # used for pagination
|
||||
|
||||
HistoryResponse* = object
|
||||
messages*: seq[WakuMessage]
|
||||
pagingInfo*: PagingInfo
|
||||
pagingInfo*: PagingInfo # used for pagination
|
||||
|
||||
HistoryRPC* = object
|
||||
requestId*: string
|
||||
@ -79,7 +79,8 @@ type
|
||||
switch*: Switch
|
||||
rng*: ref BrHmacDrbgContext
|
||||
peers*: seq[HistoryPeer]
|
||||
messages*: seq[WakuMessage]
|
||||
messages*: seq[IndexedWakuMessage]
|
||||
|
||||
|
||||
FilterRequest* = object
|
||||
contentFilters*: seq[ContentFilter]
|
||||
@ -145,8 +146,7 @@ type
|
||||
#multiaddrStrings*: seq[string]
|
||||
|
||||
WakuResult*[T] = Result[T, cstring]
|
||||
|
||||
# Encoding and decoding -------------------------------------------------------
|
||||
# Encoding and decoding -------------------------------------------------------
|
||||
|
||||
proc init*(T: type WakuMessage, buffer: seq[byte]): ProtoResult[T] =
|
||||
var msg = WakuMessage()
|
||||
@ -189,15 +189,13 @@ proc generateRequestId*(rng: ref BrHmacDrbgContext): string =
|
||||
toHex(bytes)
|
||||
|
||||
proc computeIndex*(msg: WakuMessage): Index =
|
||||
## Takes a WakuMessage and returns its index
|
||||
## Takes a WakuMessage and returns its Index
|
||||
var ctx: sha256
|
||||
ctx.init()
|
||||
if msg.contentTopic != 0: # checks for non-empty contentTopic
|
||||
ctx.update(msg.contentTopic.toBytes()) # converts the topic to bytes
|
||||
ctx.update(msg.contentTopic.toBytes()) # converts the contentTopic to bytes
|
||||
ctx.update(msg.payload)
|
||||
let digest = ctx.finish() # computes the hash
|
||||
ctx.clear()
|
||||
|
||||
result.digest = digest
|
||||
result.receivedTime = epochTime() # gets the unix timestamp
|
||||
|
||||
result.receivedTime = epochTime() # gets the unix timestamp
|
||||
@ -1,7 +1,7 @@
|
||||
import
|
||||
std/tables,
|
||||
std/[tables, sequtils, future, algorithm, options],
|
||||
bearssl,
|
||||
chronos, chronicles, metrics, stew/results,
|
||||
chronos, chronicles, metrics, stew/[results,byteutils],
|
||||
libp2p/switch,
|
||||
libp2p/crypto/crypto,
|
||||
libp2p/protocols/protocol,
|
||||
@ -173,11 +173,106 @@ proc encode*(rpc: HistoryRPC): ProtoBuffer =
|
||||
result.write(2, rpc.query.encode())
|
||||
result.write(3, rpc.response.encode())
|
||||
|
||||
proc indexComparison* (x, y: Index): int =
|
||||
## compares x and y
|
||||
## returns 0 if they are equal
|
||||
## returns -1 if x < y
|
||||
## returns 1 if x > y
|
||||
let
|
||||
timecmp = system.cmp(x.receivedTime, y.receivedTime)
|
||||
digestcm = system.cmp(x.digest.data, y.digest.data)
|
||||
if timecmp != 0: # timestamp has a higher priority for comparison
|
||||
return timecmp
|
||||
return digestcm
|
||||
|
||||
proc indexedWakuMessageComparison*(x, y: IndexedWakuMessage): int =
|
||||
## compares x and y
|
||||
## returns 0 if they are equal
|
||||
## returns -1 if x < y
|
||||
## returns 1 if x > y
|
||||
result = indexComparison(x.index, y.index)
|
||||
|
||||
proc findIndex*(msgList: seq[IndexedWakuMessage], index: Index): Option[int] =
|
||||
## returns the position of an IndexedWakuMessage in msgList whose index value matches the given index
|
||||
## returns none if no match is found
|
||||
for i, indexedWakuMessage in msgList:
|
||||
if indexedWakuMessage.index == index:
|
||||
return some(i)
|
||||
return none(int)
|
||||
|
||||
proc paginateWithIndex*(list: seq[IndexedWakuMessage], pinfo: PagingInfo): (seq[IndexedWakuMessage], PagingInfo) =
|
||||
## takes list, and performs paging based on pinfo
|
||||
## returns the page i.e, a sequence of IndexedWakuMessage and the new paging info to be used for the next paging request
|
||||
var
|
||||
cursor = pinfo.cursor
|
||||
pageSize = pinfo.pageSize
|
||||
dir = pinfo.direction
|
||||
|
||||
if pageSize == 0: # pageSize being zero indicates that no pagination is required
|
||||
return (list, pinfo)
|
||||
|
||||
if list.len == 0: # no pagination is needed for an empty list
|
||||
return (list, PagingInfo(pageSize: 0, cursor:pinfo.cursor, direction: pinfo.direction))
|
||||
|
||||
var msgList = list # makes a copy of the list
|
||||
# sorts msgList based on the custom comparison proc indexedWakuMessageComparison
|
||||
msgList.sort(indexedWakuMessageComparison)
|
||||
|
||||
var initQuery = false
|
||||
if cursor == Index():
|
||||
initQuery = true # an empty cursor means it is an intial query
|
||||
case dir
|
||||
of PagingDirection.FORWARD:
|
||||
cursor = list[0].index # perform paging from the begining of the list
|
||||
of PagingDirection.BACKWARD:
|
||||
cursor = list[list.len - 1].index # perform paging from the end of the list
|
||||
var foundIndexOption = msgList.findIndex(cursor)
|
||||
if foundIndexOption.isNone: # the cursor is not valid
|
||||
return (@[], PagingInfo(pageSize: 0, cursor:pinfo.cursor, direction: pinfo.direction))
|
||||
var foundIndex = foundIndexOption.get()
|
||||
var retrievedPageSize, s, e: int
|
||||
var newCursor: Index # to be returned as part of the new paging info
|
||||
case dir
|
||||
of PagingDirection.FORWARD: # forward pagination
|
||||
let remainingMessages= msgList.len - foundIndex - 1
|
||||
# the number of queried messages cannot exceed the MaxPageSize and the total remaining messages i.e., msgList.len-foundIndex
|
||||
retrievedPageSize = min(int(pageSize), MaxPageSize).min(remainingMessages)
|
||||
if initQuery : foundIndex = foundIndex - 1
|
||||
s = foundIndex + 1 # non inclusive
|
||||
e = foundIndex + retrievedPageSize
|
||||
newCursor = msgList[e].index # the new cursor points to the end of the page
|
||||
of PagingDirection.BACKWARD: # backward pagination
|
||||
let remainingMessages=foundIndex
|
||||
# the number of queried messages cannot exceed the MaxPageSize and the total remaining messages i.e., foundIndex-0
|
||||
retrievedPageSize = min(int(pageSize), MaxPageSize).min(remainingMessages)
|
||||
if initQuery : foundIndex = foundIndex + 1
|
||||
s = foundIndex - retrievedPageSize
|
||||
e = foundIndex - 1
|
||||
newCursor = msgList[s].index # the new cursor points to the begining of the page
|
||||
|
||||
# retrieve the messages
|
||||
for i in s..e:
|
||||
result[0].add(msgList[i])
|
||||
|
||||
result[1] = PagingInfo(pageSize : uint64(retrievedPageSize), cursor : newCursor, direction : pinfo.direction)
|
||||
|
||||
|
||||
proc paginateWithoutIndex(list: seq[IndexedWakuMessage], pinfo: PagingInfo): (seq[WakuMessage], PagingInfo) =
|
||||
## takes list, and perfomrs paging based on pinfo
|
||||
## returns the page i.e, a sequence of WakuMessage and the new paging info to be used for the next paging request
|
||||
var (indexedData, updatedPagingInfo) = paginateWithIndex(list,pinfo)
|
||||
for indexedMsg in indexedData:
|
||||
result[0].add(indexedMsg.msg)
|
||||
result[1] = updatedPagingInfo
|
||||
|
||||
proc findMessages(w: WakuStore, query: HistoryQuery): HistoryResponse =
|
||||
result = HistoryResponse(messages: newSeq[WakuMessage]())
|
||||
for msg in w.messages:
|
||||
if msg.contentTopic in query.topics:
|
||||
result.messages.insert(msg)
|
||||
# data holds IndexedWakuMessage whose topics match the query
|
||||
var data = w.messages.filterIt(it.msg.contentTopic in query.topics)
|
||||
|
||||
# perform pagination
|
||||
(result.messages, result.pagingInfo)= paginateWithoutIndex(data, query.pagingInfo)
|
||||
|
||||
|
||||
method init*(ws: WakuStore) =
|
||||
proc handle(conn: Connection, proto: string) {.async, gcsafe, closure.} =
|
||||
@ -191,7 +286,8 @@ method init*(ws: WakuStore) =
|
||||
|
||||
let value = res.value
|
||||
let response = ws.findMessages(res.value.query)
|
||||
await conn.writeLp(HistoryRPC(requestId: value.requestId, response: response).encode().buffer)
|
||||
await conn.writeLp(HistoryRPC(requestId: value.requestId,
|
||||
response: response).encode().buffer)
|
||||
|
||||
ws.handler = handle
|
||||
ws.codec = WakuStoreCodec
|
||||
@ -212,7 +308,9 @@ proc subscription*(proto: WakuStore): MessageNotificationSubscription =
|
||||
## the filter should be used by the component that receives
|
||||
## new messages.
|
||||
proc handle(topic: string, msg: WakuMessage) {.async.} =
|
||||
proto.messages.add(msg)
|
||||
let index = msg.computeIndex()
|
||||
proto.messages.add(IndexedWakuMessage(msg: msg, index: index))
|
||||
|
||||
|
||||
MessageNotificationSubscription.init(@[], handle)
|
||||
|
||||
@ -227,7 +325,8 @@ proc query*(w: WakuStore, query: HistoryQuery, handler: QueryHandlerFunc) {.asyn
|
||||
let peer = w.peers[0]
|
||||
let conn = await w.switch.dial(peer.peerInfo.peerId, peer.peerInfo.addrs, WakuStoreCodec)
|
||||
|
||||
await conn.writeLP(HistoryRPC(requestId: generateRequestId(w.rng), query: query).encode().buffer)
|
||||
await conn.writeLP(HistoryRPC(requestId: generateRequestId(w.rng),
|
||||
query: query).encode().buffer)
|
||||
|
||||
var message = await conn.readLp(64*1024)
|
||||
let response = HistoryRPC.init(message)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user