refactor set command
This commit is contained in:
@@ -1,24 +1,18 @@
|
|||||||
use std::{io::Cursor, time::Duration};
|
use std::io::Cursor;
|
||||||
|
|
||||||
|
use crate::{Result, connection::Connection, database::Database, errors::AppError};
|
||||||
use bytes::{Buf as _, Bytes};
|
use bytes::{Buf as _, Bytes};
|
||||||
use tokio::time::Instant;
|
|
||||||
|
|
||||||
use crate::{
|
|
||||||
Result,
|
|
||||||
connection::Connection,
|
|
||||||
database::{Database, Value},
|
|
||||||
errors::AppError,
|
|
||||||
};
|
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct Set {
|
pub struct Set {
|
||||||
key: String,
|
key: String,
|
||||||
value: Value,
|
data: Box<[u8]>,
|
||||||
|
expiration: Option<u64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Set {
|
impl Set {
|
||||||
pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
|
pub async fn execute(self, db: &Database, connection: &mut Connection) -> Result<()> {
|
||||||
db.set(self.key, self.value).await?;
|
db.set(self.key, self.data, self.expiration).await?;
|
||||||
|
|
||||||
connection.write(Bytes::from_static(&[1])).await?;
|
connection.write(Bytes::from_static(&[1])).await?;
|
||||||
|
|
||||||
@@ -42,15 +36,16 @@ impl Set {
|
|||||||
|
|
||||||
let data = bytes.copy_to_bytes(value_length);
|
let data = bytes.copy_to_bytes(value_length);
|
||||||
|
|
||||||
let expiration: Option<Instant> = match bytes.try_get_u8()? {
|
let expiration: Option<u64> = match bytes.try_get_u8()? {
|
||||||
1 => Some(Instant::now() + Duration::from_secs(bytes.try_get_u64()?)),
|
1 => Some(bytes.try_get_u64()?),
|
||||||
0 => None,
|
0 => None,
|
||||||
_ => return Err(AppError::UnexpectedCommandData),
|
_ => return Err(AppError::UnexpectedCommandData),
|
||||||
};
|
};
|
||||||
|
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
key,
|
key,
|
||||||
value: Value::new(data, expiration),
|
data: (*data).into(),
|
||||||
|
expiration,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ use std::{
|
|||||||
time::Duration,
|
time::Duration,
|
||||||
};
|
};
|
||||||
|
|
||||||
use bytes::{BufMut, Bytes, BytesMut};
|
use bytes::{BufMut, BytesMut};
|
||||||
use byteyarn::Yarn;
|
use byteyarn::Yarn;
|
||||||
use tokio::{
|
use tokio::{
|
||||||
sync::{Mutex, Notify},
|
sync::{Mutex, Notify},
|
||||||
@@ -33,11 +33,8 @@ pub struct Value {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl Value {
|
impl Value {
|
||||||
pub fn new(data: Bytes, expiration: Option<Instant>) -> Self {
|
pub fn new(data: Box<[u8]>, expiration: Option<Instant>) -> Self {
|
||||||
Self {
|
Self { data, expiration }
|
||||||
data: data.into_iter().collect(),
|
|
||||||
expiration,
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn from_string(data: String, expiration: Option<Instant>) -> Self {
|
pub fn from_string(data: String, expiration: Option<Instant>) -> Self {
|
||||||
@@ -69,13 +66,15 @@ impl Database {
|
|||||||
state.entries.get(key).map(|v| v.data.clone())
|
state.entries.get(key).map(|v| v.data.clone())
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn set(&self, key: String, value: Value) -> Result<()> {
|
pub async fn set(&self, key: String, data: Box<[u8]>, expiration: Option<u64>) -> Result<()> {
|
||||||
let mut state = self.state.lock().await;
|
let mut state = self.state.lock().await;
|
||||||
|
|
||||||
let expiration = value.expiration.clone();
|
let expiration = expiration.map(|seconds| Instant::now() + Duration::from_secs(seconds));
|
||||||
|
|
||||||
let key = Yarn::from(key.clone());
|
let key = Yarn::from(key.clone());
|
||||||
|
|
||||||
|
let value = Value::new(data, expiration);
|
||||||
|
|
||||||
let previous = state.entries.insert(key.clone(), value);
|
let previous = state.entries.insert(key.clone(), value);
|
||||||
|
|
||||||
let mut notify = false;
|
let mut notify = false;
|
||||||
|
|||||||
Reference in New Issue
Block a user