nim-datastore/datastore/tieredds.nim

124 lines
2.5 KiB
Nim
Raw Normal View History

import std/sequtils
import pkg/chronos
2022-05-11 10:50:05 -05:00
import pkg/questionable
import pkg/questionable/results
from pkg/stew/results as stewResults import get, isErr
2022-05-11 10:50:05 -05:00
import pkg/upraises
import ./datastore
export datastore
push: {.upraises: [].}
type
TieredDatastore* = ref object of Datastore
stores: seq[Datastore]
proc new*(
T: type TieredDatastore,
stores: varargs[Datastore]): ?!T =
if stores.len == 0:
failure "stores must contain at least one Datastore"
else:
success T(stores: @stores)
proc stores*(self: TieredDatastore): seq[Datastore] =
self.stores
2022-12-02 16:25:44 -06:00
method has*(
2022-05-11 10:50:05 -05:00
self: TieredDatastore,
2022-12-02 16:25:44 -06:00
key: Key): Future[?!bool] {.async.} =
2022-05-11 10:50:05 -05:00
for store in self.stores:
2022-12-02 16:25:44 -06:00
without res =? (await store.has(key)), err:
return failure(err)
2022-05-11 10:50:05 -05:00
2022-12-02 16:25:44 -06:00
if res:
return success true
return success false
2022-05-11 10:50:05 -05:00
method delete*(
self: TieredDatastore,
2022-12-02 16:25:44 -06:00
key: Key): Future[?!void] {.async.} =
2022-05-11 10:50:05 -05:00
let
pending = await allFinished(self.stores.mapIt(it.delete(key)))
for fut in pending:
if fut.read().isErr: return fut.read()
2022-05-11 10:50:05 -05:00
return success()
2022-05-11 10:50:05 -05:00
method delete*(
self: TieredDatastore,
2022-12-02 16:25:44 -06:00
keys: seq[Key]): Future[?!void] {.async.} =
for key in keys:
let
pending = await allFinished(self.stores.mapIt(it.delete(key)))
for fut in pending:
if fut.read().isErr: return fut.read()
return success()
2022-05-11 10:50:05 -05:00
method get*(
self: TieredDatastore,
2022-12-02 16:25:44 -06:00
key: Key): Future[?!seq[byte]] {.async.} =
2022-05-11 10:50:05 -05:00
var
2022-09-19 17:13:34 -06:00
bytes: seq[byte]
2022-05-11 10:50:05 -05:00
for store in self.stores:
2022-09-19 17:13:34 -06:00
without bytes =? (await store.get(key)):
continue
2022-09-19 17:13:34 -06:00
if bytes.len <= 0:
continue
2022-05-11 10:50:05 -05:00
# put found data into stores logically in front of the current store
2022-09-19 17:13:34 -06:00
for s in self.stores:
if s == store: break
if(
2022-09-19 22:31:24 -06:00
let res = (await s.put(key, bytes));
res.isErr):
2022-09-19 17:13:34 -06:00
return failure res.error
2022-05-11 10:50:05 -05:00
2022-09-19 17:13:34 -06:00
return success bytes
2022-05-11 10:50:05 -05:00
method put*(
self: TieredDatastore,
key: Key,
2022-12-02 16:25:44 -06:00
data: seq[byte]): Future[?!void] {.async.} =
2022-05-11 10:50:05 -05:00
let
pending = await allFinished(self.stores.mapIt(it.put(key, data)))
for fut in pending:
if fut.read().isErr: return fut.read()
2022-05-11 10:50:05 -05:00
return success()
2022-05-11 10:50:05 -05:00
method put*(
self: TieredDatastore,
2022-12-02 16:25:44 -06:00
batch: seq[BatchEntry]): Future[?!void] {.async.} =
for entry in batch:
let
pending = await allFinished(self.stores.mapIt(it.put(entry.key, entry.data)))
for fut in pending:
if fut.read().isErr: return fut.read()
return success()
2022-05-11 10:50:05 -05:00
# method query*(
# self: TieredDatastore,
2022-12-02 16:25:44 -06:00
# query: ...): Future[?!(?...)] {.async.} =
2022-05-11 10:50:05 -05:00
#
# return success ....some