Extract MessageStore interface (#350)

This commit is contained in:
Hanno Cornelius 2021-01-22 11:39:16 +02:00 committed by GitHub
parent 76c0d2b839
commit e8dd014079
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 44 additions and 27 deletions

View File

@ -2,16 +2,15 @@
import
std/[unittest, options, tables, sets],
chronos, chronicles,
../../waku/v2/node/message_store,
../../waku/v2/node/message_store/waku_message_store,
../../waku/v2/protocol/waku_store/waku_store,
./utils,
../../waku/v2/node/sqlite
./utils
suite "Message Store":
test "set and get works":
let
database = SqliteDatabase.init("", inMemory = true)[]
store = MessageStore.init(database)[]
store = WakuMessageStore.init(database)[]
topic = ContentTopic(1)
var msgs = @[

View File

@ -10,7 +10,7 @@ import
libp2p/protocols/pubsub/rpc/message,
../../waku/v2/protocol/[waku_message, message_notifier],
../../waku/v2/protocol/waku_store/waku_store,
../../waku/v2/node/[message_store, sqlite],
../../waku/v2/node/message_store/waku_message_store,
../test_helpers, ./utils
procSuite "Waku Store":
@ -62,7 +62,7 @@ procSuite "Waku Store":
peer = PeerInfo.init(key)
topic = ContentTopic(1)
database = SqliteDatabase.init("", inMemory = true)[]
store = MessageStore.init(database)[]
store = WakuMessageStore.init(database)[]
msg = WakuMessage(payload: @[byte 1, 2, 3], contentTopic: topic)
msg2 = WakuMessage(payload: @[byte 1, 2, 3], contentTopic: ContentTopic(2))

View File

@ -3,7 +3,7 @@ import
eth/keys,
../../../v1/node/rpc/hexstrings,
../../protocol/waku_store/waku_store_types,
../wakunode2, ../waku_payload,
../waku_payload,
./jsonrpc_types
export hexstrings

View File

@ -0,0 +1,19 @@
import
stew/results,
../../protocol/waku_message,
../../utils/pagination
## This module defines a message store interface. Implementations of
## MessageStore are used by the `WakuStore` protocol to store and re-
## trieve historical messages
type
DataProc* = proc(timestamp: uint64, msg: WakuMessage) {.closure.}
MessageStoreResult*[T] = Result[T, string]
MessageStore* = ref object of RootObj
# MessageStore interface
method put*(db: MessageStore, cursor: Index, message: WakuMessage): MessageStoreResult[void] {.base.} = discard
method getAll*(db: MessageStore, onData: DataProc): MessageStoreResult[bool] {.base.} = discard

View File

@ -1,14 +1,17 @@
import
sqlite3_abi,
chronos, metrics, stew/results,
chronos, metrics,
libp2p/crypto/crypto,
libp2p/protocols/protocol,
libp2p/protobuf/minprotobuf,
libp2p/stream/connection,
stew/results, metrics,
./sqlite,
../protocol/waku_message,
../utils/pagination
./message_store,
../../protocol/waku_message,
../../utils/pagination
export sqlite
# The code in this file is an adaptation of the Sqlite KV Store found in nim-eth.
# https://github.com/status-im/nim-eth/blob/master/eth/db/kvstore_sqlite3.nim
@ -16,14 +19,10 @@ import
# Most of it is a direct copy, the only unique functions being `get` and `put`.
type
DataProc* = proc(timestamp: uint64, msg: WakuMessage) {.closure.}
MessageStoreResult*[T] = Result[T, string]
MessageStore* = ref object of RootObj
WakuMessageStore* = ref object of MessageStore
database*: SqliteDatabase
proc init*(T: type MessageStore, db: SqliteDatabase): MessageStoreResult[T] =
proc init*(T: type WakuMessageStore, db: SqliteDatabase): MessageStoreResult[T] =
## Table is the SQL query for creating the messages Table.
## It contains:
## - 4-Byte ContentTopic stored as an Integer
@ -44,9 +43,9 @@ proc init*(T: type MessageStore, db: SqliteDatabase): MessageStoreResult[T] =
if res.isErr:
return err("failed to exec")
ok(MessageStore(database: db))
ok(WakuMessageStore(database: db))
proc put*(db: MessageStore, cursor: Index, message: WakuMessage): MessageStoreResult[void] =
method put*(db: WakuMessageStore, cursor: Index, message: WakuMessage): MessageStoreResult[void] =
## Adds a message to the storage.
##
## **Example:**
@ -71,7 +70,7 @@ proc put*(db: MessageStore, cursor: Index, message: WakuMessage): MessageStoreRe
ok()
proc getAll*(db: MessageStore, onData: DataProc): MessageStoreResult[bool] =
method getAll*(db: WakuMessageStore, onData: message_store.DataProc): MessageStoreResult[bool] =
## Retreives all messages from the storage.
##
## **Example:**
@ -100,6 +99,6 @@ proc getAll*(db: MessageStore, onData: DataProc): MessageStoreResult[bool] =
ok gotMessages
proc close*(db: MessageStore) =
proc close*(db: WakuMessageStore) =
## Closes the database.
db.database.close()

View File

@ -14,8 +14,7 @@ import
../protocol/waku_store/waku_store,
../protocol/waku_swap/waku_swap,
../protocol/waku_filter/waku_filter,
./message_store,
./sqlite,
./message_store/message_store,
../utils/requests
logScope:
@ -406,6 +405,7 @@ when isMainModule:
private_api,
relay_api,
store_api],
./message_store/waku_message_store,
../../common/utils/nat
proc startRpc(node: WakuNode, rpcIp: ValidIpAddress, rpcPort: Port, conf: WakuNodeConf) =
@ -473,16 +473,16 @@ when isMainModule:
# TODO Set swap peer, for now should be same as store peer
if conf.store:
var store: MessageStore
var store: WakuMessageStore
if conf.dbpath != "":
let dbRes = SqliteDatabase.init(conf.dbpath)
if dbRes.isErr:
warn "failed to init database", err = dbRes.error
let res = MessageStore.init(dbRes.value)
let res = WakuMessageStore.init(dbRes.value)
if res.isErr:
warn "failed to init MessageStore", err = res.error
warn "failed to init WakuMessageStore", err = res.error
else:
store = res.value

View File

@ -12,7 +12,7 @@ import
libp2p/protobuf/minprotobuf,
libp2p/stream/connection,
../message_notifier,
../../node/message_store,
../../node/message_store/message_store,
../waku_swap/waku_swap,
./waku_store_types,
../../utils/requests

View File

@ -6,7 +6,7 @@ import
libp2p/protocols/protocol,
../waku_swap/waku_swap_types,
../waku_message,
../../node/message_store,
../../node/message_store/message_store,
../../utils/pagination
export waku_message