mirror of
https://github.com/logos-storage/logos-storage-network-crawler.git
synced 2026-01-05 23:13:11 +00:00
formatting
This commit is contained in:
parent
5fa90c5c2f
commit
c1f25f10cc
@ -12,8 +12,7 @@ import ../state
|
||||
import ../utils/datastoreutils
|
||||
import ../utils/asyncdataevent
|
||||
|
||||
const
|
||||
nodestoreName = "nodestore"
|
||||
const nodestoreName = "nodestore"
|
||||
|
||||
type
|
||||
NodeEntry* = object
|
||||
@ -66,12 +65,9 @@ proc storeNodeIsNew(s: NodeStore, nid: Nid): Future[?!bool] {.async.} =
|
||||
return failure(err)
|
||||
|
||||
if not exists:
|
||||
let entry = NodeEntry(
|
||||
id: nid,
|
||||
lastVisit: 0
|
||||
)
|
||||
let entry = NodeEntry(id: nid, lastVisit: 0)
|
||||
?await s.store.put(key, entry)
|
||||
|
||||
|
||||
return success(not exists)
|
||||
|
||||
proc fireNewNodesDiscovered(s: NodeStore, nids: seq[Nid]): Future[?!void] {.async.} =
|
||||
@ -83,12 +79,12 @@ proc processFoundNodes(s: NodeStore, nids: seq[Nid]): Future[?!void] {.async.} =
|
||||
for nid in nids:
|
||||
without isNew =? (await s.storeNodeIsNew(nid)), err:
|
||||
return failure(err)
|
||||
|
||||
|
||||
if isNew:
|
||||
newNodes.add(nid)
|
||||
|
||||
if newNodes.len > 0:
|
||||
? await s.fireNewNodesDiscovered(newNodes)
|
||||
?await s.fireNewNodesDiscovered(newNodes)
|
||||
return success()
|
||||
|
||||
proc iterateAll*(s: NodeStore, onNode: OnNodeEntry): Future[?!void] {.async.} =
|
||||
@ -102,10 +98,10 @@ proc iterateAll*(s: NodeStore, onNode: OnNodeEntry): Future[?!void] {.async.} =
|
||||
return failure(err)
|
||||
without value =? item.value, err:
|
||||
return failure(err)
|
||||
|
||||
|
||||
?await onNode(value)
|
||||
return success()
|
||||
|
||||
|
||||
method start*(s: NodeStore): Future[?!void] {.async.} =
|
||||
info "Starting nodestore..."
|
||||
|
||||
@ -119,15 +115,8 @@ method stop*(s: NodeStore): Future[?!void] {.async.} =
|
||||
await s.state.events.nodesFound.unsubscribe(s.sub)
|
||||
return success()
|
||||
|
||||
proc new*(
|
||||
T: type NodeStore,
|
||||
state: State,
|
||||
store: TypedDatastore
|
||||
): NodeStore =
|
||||
NodeStore(
|
||||
state: state,
|
||||
store: store
|
||||
)
|
||||
proc new*(T: type NodeStore, state: State, store: TypedDatastore): NodeStore =
|
||||
NodeStore(state: state, store: store)
|
||||
|
||||
proc createNodeStore*(state: State): ?!NodeStore =
|
||||
without ds =? createTypedDatastore(state.config.dataDir / "nodestore"), err:
|
||||
|
||||
@ -113,4 +113,3 @@ proc pop*(this: List): Future[?!Nid] {.async.} =
|
||||
|
||||
proc len*(this: List): int =
|
||||
this.items.len
|
||||
|
||||
|
||||
@ -22,7 +22,9 @@ proc newAsyncDataEvent*[T](): AsyncDataEvent[T] =
|
||||
queue: newAsyncEventQueue[?T](), subscriptions: newSeq[AsyncDataEventSubscription]()
|
||||
)
|
||||
|
||||
proc performUnsubscribe[T](event: AsyncDataEvent[T], subscription: AsyncDataEventSubscription) {.async.} =
|
||||
proc performUnsubscribe[T](
|
||||
event: AsyncDataEvent[T], subscription: AsyncDataEventSubscription
|
||||
) {.async.} =
|
||||
if subscription in event.subscriptions:
|
||||
await subscription.listenFuture.cancelAndWait()
|
||||
event.subscriptions.delete(event.subscriptions.find(subscription))
|
||||
@ -35,7 +37,7 @@ proc subscribe*[T](
|
||||
listenFuture: newFuture[void](),
|
||||
fireEvent: newAsyncEvent(),
|
||||
inHandler: false,
|
||||
delayedUnsubscribe: false
|
||||
delayedUnsubscribe: false,
|
||||
)
|
||||
|
||||
proc listener() {.async.} =
|
||||
@ -62,7 +64,7 @@ proc fire*[T](event: AsyncDataEvent[T], data: T): Future[?!void] {.async.} =
|
||||
return failure(err)
|
||||
if sub.delayedUnsubscribe:
|
||||
toUnsubscribe.add(sub)
|
||||
|
||||
|
||||
for sub in toUnsubscribe:
|
||||
await event.unsubscribe(sub)
|
||||
|
||||
|
||||
@ -16,7 +16,7 @@ suite "Nodestore":
|
||||
dsPath = getTempDir() / "testds"
|
||||
nodestoreName = "nodestore"
|
||||
|
||||
var
|
||||
var
|
||||
ds: TypedDatastore
|
||||
state: MockState
|
||||
store: NodeStore
|
||||
@ -25,9 +25,7 @@ suite "Nodestore":
|
||||
ds = createTypedDatastore(dsPath).tryGet()
|
||||
state = createMockState()
|
||||
|
||||
store = NodeStore.new(
|
||||
state, ds
|
||||
)
|
||||
store = NodeStore.new(state, ds)
|
||||
|
||||
(await store.start()).tryGet()
|
||||
|
||||
@ -38,10 +36,7 @@ suite "Nodestore":
|
||||
removeDir(dsPath)
|
||||
|
||||
test "nodeEntry encoding":
|
||||
let entry = NodeEntry(
|
||||
id: genNid(),
|
||||
lastVisit: 123.uint64
|
||||
)
|
||||
let entry = NodeEntry(id: genNid(), lastVisit: 123.uint64)
|
||||
|
||||
let
|
||||
bytes = entry.encode()
|
||||
@ -52,7 +47,7 @@ suite "Nodestore":
|
||||
entry.lastVisit == decoded.lastVisit
|
||||
|
||||
test "nodesFound event should store nodes":
|
||||
let
|
||||
let
|
||||
nid = genNid()
|
||||
expectedKey = Key.init(nodestoreName / $nid).tryGet()
|
||||
|
||||
@ -60,7 +55,7 @@ suite "Nodestore":
|
||||
|
||||
check:
|
||||
(await ds.has(expectedKey)).tryGet()
|
||||
|
||||
|
||||
let entry = (await get[NodeEntry](ds, expectedKey)).tryGet()
|
||||
check:
|
||||
entry.id == nid
|
||||
@ -71,7 +66,7 @@ suite "Nodestore":
|
||||
newNodes = nids
|
||||
return success()
|
||||
|
||||
let
|
||||
let
|
||||
sub = state.events.newNodesDiscovered.subscribe(onNewNodes)
|
||||
nid = genNid()
|
||||
|
||||
@ -81,10 +76,9 @@ suite "Nodestore":
|
||||
newNodes == @[nid]
|
||||
|
||||
await state.events.newNodesDiscovered.unsubscribe(sub)
|
||||
|
||||
|
||||
test "nodesFound event should not fire newNodesDiscovered for previously seen nodes":
|
||||
let
|
||||
nid = genNid()
|
||||
let nid = genNid()
|
||||
|
||||
# Make nid known first. Then subscribe.
|
||||
(await state.events.nodesFound.fire(@[nid])).tryGet()
|
||||
@ -97,9 +91,8 @@ suite "Nodestore":
|
||||
inc count
|
||||
return success()
|
||||
|
||||
let
|
||||
sub = state.events.newNodesDiscovered.subscribe(onNewNodes)
|
||||
|
||||
let sub = state.events.newNodesDiscovered.subscribe(onNewNodes)
|
||||
|
||||
# Firing the event again should not trigger newNodesDiscovered for nid
|
||||
(await state.events.nodesFound.fire(@[nid])).tryGet()
|
||||
|
||||
@ -110,11 +103,11 @@ suite "Nodestore":
|
||||
await state.events.newNodesDiscovered.unsubscribe(sub)
|
||||
|
||||
test "iterateAll yields all known nids":
|
||||
let
|
||||
let
|
||||
nid1 = genNid()
|
||||
nid2 = genNid()
|
||||
nid3 = genNid()
|
||||
|
||||
|
||||
(await state.events.nodesFound.fire(@[nid1, nid2, nid3])).tryGet()
|
||||
|
||||
var iterNodes = newSeq[Nid]()
|
||||
|
||||
@ -4,11 +4,7 @@ import ../../codexcrawler/utils/asyncdataevent
|
||||
import ../../codexcrawler/types
|
||||
import ../../codexcrawler/config
|
||||
|
||||
type
|
||||
MockState* = ref object of State
|
||||
# config*: Config
|
||||
# events*: Events
|
||||
|
||||
type MockState* = ref object of State
|
||||
|
||||
proc createMockState*(): MockState =
|
||||
MockState(
|
||||
|
||||
@ -21,5 +21,3 @@ suite "Types":
|
||||
|
||||
check:
|
||||
nid == Nid.fromBytes(bytes).tryGet()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user