mirror of https://github.com/waku-org/nwaku.git
fix/move-protocols-to-node (#172)
This commit is contained in:
parent
24e2f7dcdc
commit
b25c1e10f2
|
@ -7,6 +7,8 @@ import
|
||||||
chronos,
|
chronos,
|
||||||
libp2p/[switch, peerinfo, multiaddress, crypto/crypto],
|
libp2p/[switch, peerinfo, multiaddress, crypto/crypto],
|
||||||
libp2p/protobuf/minprotobuf,
|
libp2p/protobuf/minprotobuf,
|
||||||
|
libp2p/protocols/protocol,
|
||||||
|
libp2p/stream/connection,
|
||||||
libp2p/protocols/pubsub/[pubsub, floodsub, gossipsub]
|
libp2p/protocols/pubsub/[pubsub, floodsub, gossipsub]
|
||||||
|
|
||||||
# Common data types -----------------------------------------------------------
|
# Common data types -----------------------------------------------------------
|
||||||
|
@ -41,10 +43,37 @@ type
|
||||||
topics*: seq[string] # @TODO TOPIC
|
topics*: seq[string] # @TODO TOPIC
|
||||||
handler*: MessageNotificationHandler
|
handler*: MessageNotificationHandler
|
||||||
|
|
||||||
|
HistoryQuery* = object
|
||||||
|
uuid*: string
|
||||||
|
topics*: seq[string]
|
||||||
|
|
||||||
|
HistoryResponse* = object
|
||||||
|
uuid*: string
|
||||||
|
messages*: seq[WakuMessage]
|
||||||
|
|
||||||
|
WakuStore* = ref object of LPProtocol
|
||||||
|
messages*: seq[WakuMessage]
|
||||||
|
|
||||||
|
FilterRequest* = object
|
||||||
|
contentFilter*: seq[ContentFilter]
|
||||||
|
topic*: string
|
||||||
|
|
||||||
|
MessagePush* = object
|
||||||
|
messages*: seq[WakuMessage]
|
||||||
|
|
||||||
|
Subscriber* = object
|
||||||
|
connection*: Connection
|
||||||
|
filter*: FilterRequest # @TODO MAKE THIS A SEQUENCE AGAIN?
|
||||||
|
|
||||||
|
WakuFilter* = ref object of LPProtocol
|
||||||
|
subscribers*: seq[Subscriber]
|
||||||
|
|
||||||
# NOTE based on Eth2Node in NBC eth2_network.nim
|
# NOTE based on Eth2Node in NBC eth2_network.nim
|
||||||
WakuNode* = ref object of RootObj
|
WakuNode* = ref object of RootObj
|
||||||
switch*: Switch
|
switch*: Switch
|
||||||
wakuRelay*: WakuRelay
|
wakuRelay*: WakuRelay
|
||||||
|
wakuStore*: WakuStore
|
||||||
|
wakuFilter*: WakuFilter
|
||||||
peerInfo*: PeerInfo
|
peerInfo*: PeerInfo
|
||||||
libp2pTransportLoops*: seq[Future[void]]
|
libp2pTransportLoops*: seq[Future[void]]
|
||||||
# TODO Revist messages field indexing as well as if this should be Message or WakuMessage
|
# TODO Revist messages field indexing as well as if this should be Message or WakuMessage
|
||||||
|
|
|
@ -29,12 +29,6 @@ type
|
||||||
Topic* = waku_types.Topic
|
Topic* = waku_types.Topic
|
||||||
Message* = seq[byte]
|
Message* = seq[byte]
|
||||||
|
|
||||||
HistoryQuery* = object
|
|
||||||
topics*: seq[string]
|
|
||||||
|
|
||||||
HistoryResponse* = object
|
|
||||||
messages*: seq[Message]
|
|
||||||
|
|
||||||
# NOTE Any difference here in Waku vs Eth2?
|
# NOTE Any difference here in Waku vs Eth2?
|
||||||
# E.g. Devp2p/Libp2p support, etc.
|
# E.g. Devp2p/Libp2p support, etc.
|
||||||
#func asLibp2pKey*(key: keys.PublicKey): PublicKey =
|
#func asLibp2pKey*(key: keys.PublicKey): PublicKey =
|
||||||
|
@ -108,13 +102,13 @@ proc start*(node: WakuNode) {.async.} =
|
||||||
node.libp2pTransportLoops = await node.switch.start()
|
node.libp2pTransportLoops = await node.switch.start()
|
||||||
|
|
||||||
# NOTE WakuRelay is being instantiated as part of initing node
|
# NOTE WakuRelay is being instantiated as part of initing node
|
||||||
let storeProto = WakuStore.init()
|
node.wakuStore = WakuStore.init()
|
||||||
node.switch.mount(storeProto)
|
node.switch.mount(node.wakuStore)
|
||||||
node.subscriptions.subscribe(WakuStoreCodec, storeProto.subscription())
|
node.subscriptions.subscribe(WakuStoreCodec, node.wakuStore.subscription())
|
||||||
|
|
||||||
let filterProto = WakuFilter.init()
|
node.wakuFilter = WakuFilter.init()
|
||||||
node.switch.mount(filterProto)
|
node.switch.mount(node.wakuFilter)
|
||||||
node.subscriptions.subscribe(WakuFilterCodec, filterProto.subscription())
|
node.subscriptions.subscribe(WakuFilterCodec, node.wakuFilter.subscription())
|
||||||
|
|
||||||
proc relayHandler(topic: string, data: seq[byte]) {.async, gcsafe.} =
|
proc relayHandler(topic: string, data: seq[byte]) {.async, gcsafe.} =
|
||||||
let msg = WakuMessage.init(data)
|
let msg = WakuMessage.init(data)
|
||||||
|
@ -193,14 +187,10 @@ proc query*(w: WakuNode, query: HistoryQuery): HistoryResponse =
|
||||||
##
|
##
|
||||||
## Status: Not yet implemented.
|
## Status: Not yet implemented.
|
||||||
## TODO Implement as wrapper around `waku_store` and send RPC.
|
## TODO Implement as wrapper around `waku_store` and send RPC.
|
||||||
result.messages = newSeq[Message]()
|
|
||||||
|
# XXX Unclear how this should be hooked up, Message or WakuMessage?
|
||||||
for msg in w.messages:
|
# result.messages.insert(msg[1])
|
||||||
if msg[0] notin query.topics:
|
discard
|
||||||
continue
|
|
||||||
|
|
||||||
# XXX Unclear how this should be hooked up, Message or WakuMessage?
|
|
||||||
# result.messages.insert(msg[1])
|
|
||||||
|
|
||||||
when isMainModule:
|
when isMainModule:
|
||||||
import
|
import
|
||||||
|
|
|
@ -20,24 +20,6 @@ logScope:
|
||||||
const
|
const
|
||||||
WakuFilterCodec* = "/vac/waku/filter/2.0.0-alpha5"
|
WakuFilterCodec* = "/vac/waku/filter/2.0.0-alpha5"
|
||||||
|
|
||||||
type
|
|
||||||
ContentFilter* = object
|
|
||||||
topics*: seq[string]
|
|
||||||
|
|
||||||
FilterRequest* = object
|
|
||||||
contentFilter*: seq[ContentFilter]
|
|
||||||
topic*: string
|
|
||||||
|
|
||||||
MessagePush* = object
|
|
||||||
messages*: seq[WakuMessage]
|
|
||||||
|
|
||||||
Subscriber = object
|
|
||||||
connection: Connection
|
|
||||||
filter: FilterRequest # @TODO MAKE THIS A SEQUENCE AGAIN?
|
|
||||||
|
|
||||||
WakuFilter* = ref object of LPProtocol
|
|
||||||
subscribers*: seq[Subscriber]
|
|
||||||
|
|
||||||
proc encode*(filter: ContentFilter): ProtoBuffer =
|
proc encode*(filter: ContentFilter): ProtoBuffer =
|
||||||
result = initProtoBuffer()
|
result = initProtoBuffer()
|
||||||
|
|
||||||
|
|
|
@ -13,18 +13,6 @@ logScope:
|
||||||
const
|
const
|
||||||
WakuStoreCodec* = "/vac/waku/store/2.0.0-alpha5"
|
WakuStoreCodec* = "/vac/waku/store/2.0.0-alpha5"
|
||||||
|
|
||||||
type
|
|
||||||
HistoryQuery* = object
|
|
||||||
uuid*: string
|
|
||||||
topics*: seq[string]
|
|
||||||
|
|
||||||
HistoryResponse* = object
|
|
||||||
uuid*: string
|
|
||||||
messages*: seq[WakuMessage]
|
|
||||||
|
|
||||||
WakuStore* = ref object of LPProtocol
|
|
||||||
messages*: seq[WakuMessage]
|
|
||||||
|
|
||||||
proc init*(T: type HistoryQuery, buffer: seq[byte]): ProtoResult[T] =
|
proc init*(T: type HistoryQuery, buffer: seq[byte]): ProtoResult[T] =
|
||||||
var msg = HistoryQuery()
|
var msg = HistoryQuery()
|
||||||
let pb = initProtoBuffer(buffer)
|
let pb = initProtoBuffer(buffer)
|
||||||
|
|
Loading…
Reference in New Issue