in-memory storage #32
5 changed files with 66 additions and 21 deletions
|
@ -443,7 +443,7 @@ impl<S: BayouState> Bayou<S> {
|
||||||
struct K2vWatch {
|
struct K2vWatch {
|
||||||
pk: String,
|
pk: String,
|
||||||
sk: String,
|
sk: String,
|
||||||
rx: watch::Receiver<storage::RowRef>,
|
rx: watch::Receiver<storage::OrphanRowRef>,
|
||||||
notify: Notify,
|
notify: Notify,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -454,7 +454,7 @@ impl K2vWatch {
|
||||||
fn new(creds: &Credentials, pk: String, sk: String) -> Result<Arc<Self>> {
|
fn new(creds: &Credentials, pk: String, sk: String) -> Result<Arc<Self>> {
|
||||||
let row_client = creds.row_client()?;
|
let row_client = creds.row_client()?;
|
||||||
|
|
||||||
let (tx, rx) = watch::channel::<storage::RowRef>(row_client.row(&pk, &sk));
|
let (tx, rx) = watch::channel::<storage::OrphanRowRef>(row_client.row(&pk, &sk).to_orphan());
|
||||||
let notify = Notify::new();
|
let notify = Notify::new();
|
||||||
|
|
||||||
let watch = Arc::new(K2vWatch { pk, sk, rx, notify });
|
let watch = Arc::new(K2vWatch { pk, sk, rx, notify });
|
||||||
|
@ -471,7 +471,7 @@ impl K2vWatch {
|
||||||
async fn background_task(
|
async fn background_task(
|
||||||
self_weak: Weak<Self>,
|
self_weak: Weak<Self>,
|
||||||
k2v: storage::RowStore,
|
k2v: storage::RowStore,
|
||||||
tx: watch::Sender<storage::RowRef>,
|
tx: watch::Sender<storage::OrphanRowRef>,
|
||||||
) {
|
) {
|
||||||
let mut row = match Weak::upgrade(&self_weak) {
|
let mut row = match Weak::upgrade(&self_weak) {
|
||||||
Some(this) => k2v.row(&this.pk, &this.sk),
|
Some(this) => k2v.row(&this.pk, &this.sk),
|
||||||
|
@ -497,7 +497,7 @@ impl K2vWatch {
|
||||||
}
|
}
|
||||||
Ok(new_value) => {
|
Ok(new_value) => {
|
||||||
row = new_value.to_ref();
|
row = new_value.to_ref();
|
||||||
if tx.send(XXX).is_err() {
|
if tx.send(row.to_orphan()).is_err() {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -81,7 +81,11 @@ impl User {
|
||||||
let mb_uidvalidity = mb.current_uid_index().await.uidvalidity;
|
let mb_uidvalidity = mb.current_uid_index().await.uidvalidity;
|
||||||
if mb_uidvalidity > uidvalidity {
|
if mb_uidvalidity > uidvalidity {
|
||||||
list.update_uidvalidity(name, mb_uidvalidity);
|
list.update_uidvalidity(name, mb_uidvalidity);
|
||||||
self.save_mailbox_list(&list, ct).await?;
|
let orphan = match ct {
|
||||||
|
Some(x) => Some(x.to_orphan()),
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
self.save_mailbox_list(&list, orphan).await?;
|
||||||
}
|
}
|
||||||
Ok(Some(mb))
|
Ok(Some(mb))
|
||||||
} else {
|
} else {
|
||||||
|
@ -104,7 +108,11 @@ impl User {
|
||||||
let (mut list, ct) = self.load_mailbox_list().await?;
|
let (mut list, ct) = self.load_mailbox_list().await?;
|
||||||
match list.create_mailbox(name) {
|
match list.create_mailbox(name) {
|
||||||
CreatedMailbox::Created(_, _) => {
|
CreatedMailbox::Created(_, _) => {
|
||||||
self.save_mailbox_list(&list, ct).await?;
|
let orphan = match ct {
|
||||||
|
Some(x) => Some(x.to_orphan()),
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
self.save_mailbox_list(&list, orphan).await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
CreatedMailbox::Existed(_, _) => Err(anyhow!("Mailbox {} already exists", name)),
|
CreatedMailbox::Existed(_, _) => Err(anyhow!("Mailbox {} already exists", name)),
|
||||||
|
@ -121,7 +129,11 @@ impl User {
|
||||||
if list.has_mailbox(name) {
|
if list.has_mailbox(name) {
|
||||||
// TODO: actually delete mailbox contents
|
// TODO: actually delete mailbox contents
|
||||||
list.set_mailbox(name, None);
|
list.set_mailbox(name, None);
|
||||||
self.save_mailbox_list(&list, ct).await?;
|
let orphan = match ct {
|
||||||
|
Some(x) => Some(x.to_orphan()),
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
self.save_mailbox_list(&list, orphan).await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
} else {
|
} else {
|
||||||
bail!("Mailbox {} does not exist", name);
|
bail!("Mailbox {} does not exist", name);
|
||||||
|
@ -142,7 +154,11 @@ impl User {
|
||||||
if old_name == INBOX {
|
if old_name == INBOX {
|
||||||
list.rename_mailbox(old_name, new_name)?;
|
list.rename_mailbox(old_name, new_name)?;
|
||||||
if !self.ensure_inbox_exists(&mut list, &ct).await? {
|
if !self.ensure_inbox_exists(&mut list, &ct).await? {
|
||||||
self.save_mailbox_list(&list, ct).await?;
|
let orphan = match ct {
|
||||||
|
Some(x) => Some(x.to_orphan()),
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
self.save_mailbox_list(&list, orphan).await?;
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
let names = list.existing_mailbox_names();
|
let names = list.existing_mailbox_names();
|
||||||
|
@ -165,7 +181,12 @@ impl User {
|
||||||
list.rename_mailbox(name, &nnew)?;
|
list.rename_mailbox(name, &nnew)?;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
self.save_mailbox_list(&list, ct).await?;
|
|
||||||
|
let orphan = match ct {
|
||||||
|
Some(x) => Some(x.to_orphan()),
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
self.save_mailbox_list(&list, orphan).await?;
|
||||||
}
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
@ -257,7 +278,11 @@ impl User {
|
||||||
let saved;
|
let saved;
|
||||||
let (inbox_id, inbox_uidvalidity) = match list.create_mailbox(INBOX) {
|
let (inbox_id, inbox_uidvalidity) = match list.create_mailbox(INBOX) {
|
||||||
CreatedMailbox::Created(i, v) => {
|
CreatedMailbox::Created(i, v) => {
|
||||||
self.save_mailbox_list(list, ct.clone()).await?;
|
let orphan = match ct {
|
||||||
|
Some(x) => Some(x.to_orphan()),
|
||||||
|
None => None,
|
||||||
|
};
|
||||||
|
self.save_mailbox_list(list, orphan).await?;
|
||||||
saved = true;
|
saved = true;
|
||||||
(i, v)
|
(i, v)
|
||||||
}
|
}
|
||||||
|
@ -277,11 +302,11 @@ impl User {
|
||||||
async fn save_mailbox_list(
|
async fn save_mailbox_list(
|
||||||
&self,
|
&self,
|
||||||
list: &MailboxList,
|
list: &MailboxList,
|
||||||
ct: Option<storage::RowRef>,
|
ct: Option<storage::OrphanRowRef>,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let list_blob = seal_serialize(list, &self.creds.keys.master)?;
|
let list_blob = seal_serialize(list, &self.creds.keys.master)?;
|
||||||
let rref = match ct {
|
let rref = match ct {
|
||||||
Some(x) => x,
|
Some(x) => self.k2v.from_orphan(x),
|
||||||
None => self.k2v.row(MAILBOX_LIST_PK, MAILBOX_LIST_SK),
|
None => self.k2v.row(MAILBOX_LIST_PK, MAILBOX_LIST_SK),
|
||||||
};
|
};
|
||||||
rref.set_value(list_blob).push().await?;
|
rref.set_value(list_blob).push().await?;
|
||||||
|
|
|
@ -6,6 +6,9 @@ pub struct GrgStore {}
|
||||||
pub struct GrgRef {}
|
pub struct GrgRef {}
|
||||||
pub struct GrgValue {}
|
pub struct GrgValue {}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug)]
|
||||||
|
pub struct GrgOrphanRowRef {}
|
||||||
|
|
||||||
impl IBuilders for GrgCreds {
|
impl IBuilders for GrgCreds {
|
||||||
fn row_store(&self) -> Result<RowStore, StorageError> {
|
fn row_store(&self) -> Result<RowStore, StorageError> {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
|
@ -32,13 +35,17 @@ impl IRowStore for GrgStore {
|
||||||
fn rm(&self, selector: Selector) -> AsyncResult<()> {
|
fn rm(&self, selector: Selector) -> AsyncResult<()> {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn from_orphan(&self, orphan: OrphanRowRef) -> RowRef {
|
||||||
|
unimplemented!();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl IRowRef for GrgRef {
|
impl IRowRef for GrgRef {
|
||||||
/*fn clone_boxed(&self) -> RowRef {
|
/*fn clone_boxed(&self) -> RowRef {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
}*/
|
}*/
|
||||||
fn to_orphan(&self) -> RowRefOrphan {
|
fn to_orphan(&self) -> OrphanRowRef {
|
||||||
unimplemented!()
|
unimplemented!()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -7,6 +7,9 @@ pub struct MemStore {}
|
||||||
pub struct MemRef {}
|
pub struct MemRef {}
|
||||||
pub struct MemValue {}
|
pub struct MemValue {}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug)]
|
||||||
|
pub struct MemOrphanRowRef {}
|
||||||
|
|
||||||
impl IBuilders for FullMem {
|
impl IBuilders for FullMem {
|
||||||
fn row_store(&self) -> Result<RowStore, StorageError> {
|
fn row_store(&self) -> Result<RowStore, StorageError> {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
|
@ -33,16 +36,24 @@ impl IRowStore for MemStore {
|
||||||
fn rm(&self, selector: Selector) -> AsyncResult<()> {
|
fn rm(&self, selector: Selector) -> AsyncResult<()> {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn from_orphan(&self, orphan: OrphanRowRef) -> RowRef {
|
||||||
|
unimplemented!();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl IRowRef for MemRef {
|
impl IRowRef for MemRef {
|
||||||
|
fn to_orphan(&self) -> OrphanRowRef {
|
||||||
|
unimplemented!()
|
||||||
|
}
|
||||||
|
|
||||||
fn key(&self) -> (&str, &str) {
|
fn key(&self) -> (&str, &str) {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
}
|
}
|
||||||
|
|
||||||
fn clone_boxed(&self) -> RowRef {
|
/*fn clone_boxed(&self) -> RowRef {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
}
|
}*/
|
||||||
|
|
||||||
fn set_value(&self, content: Vec<u8>) -> RowValue {
|
fn set_value(&self, content: Vec<u8>) -> RowValue {
|
||||||
unimplemented!();
|
unimplemented!();
|
||||||
|
|
|
@ -20,6 +20,12 @@ pub enum Alternative {
|
||||||
}
|
}
|
||||||
type ConcurrentValues = Vec<Alternative>;
|
type ConcurrentValues = Vec<Alternative>;
|
||||||
|
|
||||||
|
#[derive(Clone, Debug)]
|
||||||
|
pub enum OrphanRowRef {
|
||||||
|
Garage(garage::GrgOrphanRowRef),
|
||||||
|
Memory(in_memory::MemOrphanRowRef),
|
||||||
|
}
|
||||||
|
|
||||||
pub enum Selector<'a> {
|
pub enum Selector<'a> {
|
||||||
Range { shard_key: &'a str, begin: &'a str, end: &'a str },
|
Range { shard_key: &'a str, begin: &'a str, end: &'a str },
|
||||||
List (Vec<(&'a str, &'a str)>), // list of (shard_key, sort_key)
|
List (Vec<(&'a str, &'a str)>), // list of (shard_key, sort_key)
|
||||||
|
@ -81,13 +87,14 @@ pub trait IRowStore
|
||||||
fn row(&self, partition: &str, sort: &str) -> RowRef;
|
fn row(&self, partition: &str, sort: &str) -> RowRef;
|
||||||
fn select(&self, selector: Selector) -> AsyncResult<Vec<RowValue>>;
|
fn select(&self, selector: Selector) -> AsyncResult<Vec<RowValue>>;
|
||||||
fn rm(&self, selector: Selector) -> AsyncResult<()>;
|
fn rm(&self, selector: Selector) -> AsyncResult<()>;
|
||||||
|
fn from_orphan(&self, orphan: OrphanRowRef) -> RowRef;
|
||||||
}
|
}
|
||||||
pub type RowStore = Box<dyn IRowStore + Sync + Send>;
|
pub type RowStore = Box<dyn IRowStore + Sync + Send>;
|
||||||
|
|
||||||
pub trait IRowRef
|
pub trait IRowRef
|
||||||
{
|
{
|
||||||
/*fn clone_boxed(&self) -> RowRef;*/
|
/*fn clone_boxed(&self) -> RowRef;*/
|
||||||
fn to_orphan(&self) -> RowRefOrphan;
|
fn to_orphan(&self) -> OrphanRowRef;
|
||||||
fn key(&self) -> (&str, &str);
|
fn key(&self) -> (&str, &str);
|
||||||
fn set_value(&self, content: Vec<u8>) -> RowValue;
|
fn set_value(&self, content: Vec<u8>) -> RowValue;
|
||||||
fn fetch(&self) -> AsyncResult<RowValue>;
|
fn fetch(&self) -> AsyncResult<RowValue>;
|
||||||
|
@ -101,11 +108,6 @@ pub type RowRef = Box<dyn IRowRef + Send + Sync>;
|
||||||
}
|
}
|
||||||
}*/
|
}*/
|
||||||
|
|
||||||
pub trait IRowRefOrphan
|
|
||||||
{
|
|
||||||
fn attach(&self, store: &RowStore) -> RowRef;
|
|
||||||
}
|
|
||||||
pub type RowRefOrphan = Box<dyn IRowRefOrphan + Send + Sync>;
|
|
||||||
|
|
||||||
pub trait IRowValue
|
pub trait IRowValue
|
||||||
{
|
{
|
||||||
|
|
Loading…
Reference in a new issue