mirror of
https://github.com/meilisearch/MeiliSearch
synced 2025-07-04 20:37:15 +02:00
fix clippy warnings
This commit is contained in:
parent
01479dcf99
commit
abbea59732
17 changed files with 124 additions and 131 deletions
|
@ -92,7 +92,7 @@ enum IndexMsg {
|
|||
},
|
||||
GetMeta {
|
||||
uuid: Uuid,
|
||||
ret: oneshot::Sender<Result<Option<IndexMeta>>>,
|
||||
ret: oneshot::Sender<Result<IndexMeta>>,
|
||||
},
|
||||
UpdateIndex {
|
||||
uuid: Uuid,
|
||||
|
@ -137,7 +137,7 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
|
|||
) -> Result<Self> {
|
||||
let options = IndexerOpts::default();
|
||||
let update_handler =
|
||||
UpdateHandler::new(&options).map_err(|e| IndexError::Error(e.into()))?;
|
||||
UpdateHandler::new(&options).map_err(IndexError::Error)?;
|
||||
let update_handler = Arc::new(update_handler);
|
||||
let read_receiver = Some(read_receiver);
|
||||
let write_receiver = Some(write_receiver);
|
||||
|
@ -274,11 +274,11 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
|
|||
data: File,
|
||||
) -> Result<UpdateResult> {
|
||||
debug!("Processing update {}", meta.id());
|
||||
let uuid = meta.index_uuid().clone();
|
||||
let uuid = meta.index_uuid();
|
||||
let update_handler = self.update_handler.clone();
|
||||
let index = match self.store.get(uuid.clone()).await? {
|
||||
let index = match self.store.get(*uuid).await? {
|
||||
Some(index) => index,
|
||||
None => self.store.create(uuid, None).await?,
|
||||
None => self.store.create(*uuid, None).await?,
|
||||
};
|
||||
spawn_blocking(move || update_handler.handle_update(meta, data, index))
|
||||
.await
|
||||
|
@ -291,7 +291,7 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
|
|||
.get(uuid)
|
||||
.await?
|
||||
.ok_or(IndexError::UnexistingIndex)?;
|
||||
spawn_blocking(move || index.settings().map_err(|e| IndexError::Error(e)))
|
||||
spawn_blocking(move || index.settings().map_err(IndexError::Error))
|
||||
.await
|
||||
.map_err(|e| IndexError::Error(e.into()))?
|
||||
}
|
||||
|
@ -311,7 +311,7 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
|
|||
spawn_blocking(move || {
|
||||
index
|
||||
.retrieve_documents(offset, limit, attributes_to_retrieve)
|
||||
.map_err(|e| IndexError::Error(e))
|
||||
.map_err(IndexError::Error)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| IndexError::Error(e.into()))?
|
||||
|
@ -331,7 +331,7 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
|
|||
spawn_blocking(move || {
|
||||
index
|
||||
.retrieve_document(doc_id, attributes_to_retrieve)
|
||||
.map_err(|e| IndexError::Error(e))
|
||||
.map_err(IndexError::Error)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| IndexError::Error(e.into()))?
|
||||
|
@ -354,15 +354,15 @@ impl<S: IndexStore + Sync + Send> IndexActor<S> {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
async fn handle_get_meta(&self, uuid: Uuid) -> Result<Option<IndexMeta>> {
|
||||
async fn handle_get_meta(&self, uuid: Uuid) -> Result<IndexMeta> {
|
||||
match self.store.get(uuid).await? {
|
||||
Some(index) => {
|
||||
let meta = spawn_blocking(move || IndexMeta::new(&index))
|
||||
.await
|
||||
.map_err(|e| IndexError::Error(e.into()))??;
|
||||
Ok(Some(meta))
|
||||
Ok(meta)
|
||||
}
|
||||
None => Ok(None),
|
||||
None => Err(IndexError::UnexistingIndex),
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -405,7 +405,7 @@ impl IndexActorHandle {
|
|||
let (read_sender, read_receiver) = mpsc::channel(100);
|
||||
let (write_sender, write_receiver) = mpsc::channel(100);
|
||||
|
||||
let store = HeedIndexStore::new(path, index_size)?;
|
||||
let store = HeedIndexStore::new(path, index_size);
|
||||
let actor = IndexActor::new(read_receiver, write_receiver, store)?;
|
||||
tokio::task::spawn(actor.run());
|
||||
Ok(Self {
|
||||
|
@ -492,7 +492,7 @@ impl IndexActorHandle {
|
|||
Ok(receiver.await.expect("IndexActor has been killed")?)
|
||||
}
|
||||
|
||||
pub async fn get_index_meta(&self, uuid: Uuid) -> Result<Option<IndexMeta>> {
|
||||
pub async fn get_index_meta(&self, uuid: Uuid) -> Result<IndexMeta> {
|
||||
let (ret, receiver) = oneshot::channel();
|
||||
let msg = IndexMsg::GetMeta { uuid, ret };
|
||||
let _ = self.read_sender.send(msg).await;
|
||||
|
@ -518,14 +518,14 @@ struct HeedIndexStore {
|
|||
}
|
||||
|
||||
impl HeedIndexStore {
|
||||
fn new(path: impl AsRef<Path>, index_size: usize) -> anyhow::Result<Self> {
|
||||
fn new(path: impl AsRef<Path>, index_size: usize) -> Self {
|
||||
let path = path.as_ref().join("indexes/");
|
||||
let index_store = Arc::new(RwLock::new(HashMap::new()));
|
||||
Ok(Self {
|
||||
Self {
|
||||
index_store,
|
||||
path,
|
||||
index_size,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -550,7 +550,7 @@ impl IndexStore for HeedIndexStore {
|
|||
.await
|
||||
.map_err(|e| IndexError::Error(e.into()))??;
|
||||
|
||||
self.index_store.write().await.insert(uuid.clone(), index.clone());
|
||||
self.index_store.write().await.insert(uuid, index.clone());
|
||||
|
||||
Ok(index)
|
||||
}
|
||||
|
@ -574,7 +574,7 @@ impl IndexStore for HeedIndexStore {
|
|||
self.index_store
|
||||
.write()
|
||||
.await
|
||||
.insert(uuid.clone(), index.clone());
|
||||
.insert(uuid, index.clone());
|
||||
Ok(Some(index))
|
||||
}
|
||||
}
|
||||
|
@ -595,6 +595,6 @@ fn open_index(path: impl AsRef<Path>, size: usize) -> Result<Index> {
|
|||
let mut options = EnvOpenOptions::new();
|
||||
options.map_size(size);
|
||||
let index = milli::Index::new(options, &path)
|
||||
.map_err(|e| IndexError::Error(e))?;
|
||||
.map_err(IndexError::Error)?;
|
||||
Ok(Index(Arc::new(index)))
|
||||
}
|
||||
|
|
|
@ -127,7 +127,7 @@ impl IndexController {
|
|||
// the update is processed. This would make calls to GET index to fail until the update
|
||||
// is complete. Since this is get or create, we ignore the error when the index already
|
||||
// exists.
|
||||
match self.index_handle.create_index(uuid.clone(), None).await {
|
||||
match self.index_handle.create_index(uuid, None).await {
|
||||
Ok(_) | Err(index_actor::IndexError::IndexAlreadyExists) => (),
|
||||
Err(e) => return Err(e.into()),
|
||||
}
|
||||
|
@ -158,12 +158,12 @@ impl IndexController {
|
|||
let uuid = self.uuid_resolver
|
||||
.delete(uid)
|
||||
.await?;
|
||||
self.update_handle.delete(uuid.clone()).await?;
|
||||
self.update_handle.delete(uuid).await?;
|
||||
self.index_handle.delete(uuid).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn update_status(&self, uid: String, id: u64) -> anyhow::Result<Option<UpdateStatus>> {
|
||||
pub async fn update_status(&self, uid: String, id: u64) -> anyhow::Result<UpdateStatus> {
|
||||
let uuid = self.uuid_resolver
|
||||
.resolve(uid)
|
||||
.await?;
|
||||
|
@ -184,10 +184,9 @@ impl IndexController {
|
|||
let mut ret = Vec::new();
|
||||
|
||||
for (uid, uuid) in uuids {
|
||||
if let Some(meta) = self.index_handle.get_index_meta(uuid).await? {
|
||||
let meta = IndexMetadata { uid, meta };
|
||||
ret.push(meta);
|
||||
}
|
||||
let meta = self.index_handle.get_index_meta(uuid).await?;
|
||||
let meta = IndexMetadata { uid, meta };
|
||||
ret.push(meta);
|
||||
}
|
||||
|
||||
Ok(ret)
|
||||
|
@ -247,13 +246,13 @@ impl IndexController {
|
|||
Ok(result)
|
||||
}
|
||||
|
||||
pub async fn get_index(&self, uid: String) -> anyhow::Result<Option<IndexMetadata>> {
|
||||
pub async fn get_index(&self, uid: String) -> anyhow::Result<IndexMetadata> {
|
||||
let uuid = self.uuid_resolver.resolve(uid.clone()).await?;
|
||||
let result = self.index_handle
|
||||
let meta = self.index_handle
|
||||
.get_index_meta(uuid)
|
||||
.await?
|
||||
.map(|meta| IndexMetadata { uid, meta });
|
||||
Ok(result)
|
||||
.await?;
|
||||
let meta = IndexMetadata { uid, meta };
|
||||
Ok(meta)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -25,6 +25,8 @@ pub enum UpdateError {
|
|||
Error(Box<dyn std::error::Error + Sync + Send + 'static>),
|
||||
#[error("Index {0} doesn't exist.")]
|
||||
UnexistingIndex(Uuid),
|
||||
#[error("Update {0} doesn't exist.")]
|
||||
UnexistingUpdate(u64),
|
||||
}
|
||||
|
||||
enum UpdateMsg<D> {
|
||||
|
@ -40,7 +42,7 @@ enum UpdateMsg<D> {
|
|||
},
|
||||
GetUpdate {
|
||||
uuid: Uuid,
|
||||
ret: oneshot::Sender<Result<Option<UpdateStatus>>>,
|
||||
ret: oneshot::Sender<Result<UpdateStatus>>,
|
||||
id: u64,
|
||||
},
|
||||
Delete {
|
||||
|
@ -62,8 +64,8 @@ struct UpdateActor<D, S> {
|
|||
#[async_trait::async_trait]
|
||||
trait UpdateStoreStore {
|
||||
async fn get_or_create(&self, uuid: Uuid) -> Result<Arc<UpdateStore>>;
|
||||
async fn delete(&self, uuid: &Uuid) -> Result<Option<Arc<UpdateStore>>>;
|
||||
async fn get(&self, uuid: &Uuid) -> Result<Option<Arc<UpdateStore>>>;
|
||||
async fn delete(&self, uuid: Uuid) -> Result<Option<Arc<UpdateStore>>>;
|
||||
async fn get(&self, uuid: Uuid) -> Result<Option<Arc<UpdateStore>>>;
|
||||
}
|
||||
|
||||
impl<D, S> UpdateActor<D, S>
|
||||
|
@ -145,18 +147,17 @@ where
|
|||
.map_err(|e| UpdateError::Error(Box::new(e)))?;
|
||||
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let result = update_store
|
||||
update_store
|
||||
.register_update(meta, path, uuid)
|
||||
.map(|pending| UpdateStatus::Pending(pending))
|
||||
.map_err(|e| UpdateError::Error(Box::new(e)));
|
||||
result
|
||||
.map(UpdateStatus::Pending)
|
||||
.map_err(|e| UpdateError::Error(Box::new(e)))
|
||||
})
|
||||
.await
|
||||
.map_err(|e| UpdateError::Error(Box::new(e)))?
|
||||
}
|
||||
|
||||
async fn handle_list_updates(&self, uuid: Uuid) -> Result<Vec<UpdateStatus>> {
|
||||
let update_store = self.store.get(&uuid).await?;
|
||||
let update_store = self.store.get(uuid).await?;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let result = update_store
|
||||
.ok_or(UpdateError::UnexistingIndex(uuid))?
|
||||
|
@ -168,20 +169,21 @@ where
|
|||
.map_err(|e| UpdateError::Error(Box::new(e)))?
|
||||
}
|
||||
|
||||
async fn handle_get_update(&self, uuid: Uuid, id: u64) -> Result<Option<UpdateStatus>> {
|
||||
async fn handle_get_update(&self, uuid: Uuid, id: u64) -> Result<UpdateStatus> {
|
||||
let store = self
|
||||
.store
|
||||
.get(&uuid)
|
||||
.get(uuid)
|
||||
.await?
|
||||
.ok_or(UpdateError::UnexistingIndex(uuid))?;
|
||||
let result = store
|
||||
.meta(id)
|
||||
.map_err(|e| UpdateError::Error(Box::new(e)))?;
|
||||
.map_err(|e| UpdateError::Error(Box::new(e)))?
|
||||
.ok_or(UpdateError::UnexistingUpdate(id))?;
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn handle_delete(&self, uuid: Uuid) -> Result<()> {
|
||||
let store = self.store.delete(&uuid).await?;
|
||||
let store = self.store.delete(uuid).await?;
|
||||
|
||||
if let Some(store) = store {
|
||||
tokio::task::spawn(async move {
|
||||
|
@ -246,7 +248,7 @@ where
|
|||
receiver.await.expect("update actor killed.")
|
||||
}
|
||||
|
||||
pub async fn update_status(&self, uuid: Uuid, id: u64) -> Result<Option<UpdateStatus>> {
|
||||
pub async fn update_status(&self, uuid: Uuid, id: u64) -> Result<UpdateStatus> {
|
||||
let (ret, receiver) = oneshot::channel();
|
||||
let msg = UpdateMsg::GetUpdate { uuid, id, ret };
|
||||
let _ = self.sender.send(msg).await;
|
||||
|
@ -310,9 +312,9 @@ impl UpdateStoreStore for MapUpdateStoreStore {
|
|||
}
|
||||
}
|
||||
|
||||
async fn get(&self, uuid: &Uuid) -> Result<Option<Arc<UpdateStore>>> {
|
||||
async fn get(&self, uuid: Uuid) -> Result<Option<Arc<UpdateStore>>> {
|
||||
let guard = self.db.read().await;
|
||||
match guard.get(uuid) {
|
||||
match guard.get(&uuid) {
|
||||
Some(uuid) => Ok(Some(uuid.clone())),
|
||||
None => {
|
||||
// The index is not found in the found in the loaded indexes, so we attempt to load
|
||||
|
@ -322,7 +324,7 @@ impl UpdateStoreStore for MapUpdateStoreStore {
|
|||
let path = self.path.clone().join(format!("updates-{}", uuid));
|
||||
if path.exists() {
|
||||
let mut guard = self.db.write().await;
|
||||
match guard.entry(uuid.clone()) {
|
||||
match guard.entry(uuid) {
|
||||
Entry::Vacant(entry) => {
|
||||
// We can safely load the index
|
||||
let index_handle = self.index_handle.clone();
|
||||
|
@ -333,7 +335,7 @@ impl UpdateStoreStore for MapUpdateStoreStore {
|
|||
futures::executor::block_on(index_handle.update(meta, file))
|
||||
})
|
||||
.map_err(|e| UpdateError::Error(e.into()))?;
|
||||
let store = entry.insert(store.clone());
|
||||
let store = entry.insert(store);
|
||||
Ok(Some(store.clone()))
|
||||
}
|
||||
Entry::Occupied(entry) => {
|
||||
|
@ -348,7 +350,7 @@ impl UpdateStoreStore for MapUpdateStoreStore {
|
|||
}
|
||||
}
|
||||
|
||||
async fn delete(&self, uuid: &Uuid) -> Result<Option<Arc<UpdateStore>>> {
|
||||
async fn delete(&self, uuid: Uuid) -> Result<Option<Arc<UpdateStore>>> {
|
||||
let store = self.db.write().await.remove(&uuid);
|
||||
let path = self.path.clone().join(format!("updates-{}", uuid));
|
||||
if store.is_some() || path.exists() {
|
||||
|
|
|
@ -92,7 +92,7 @@ where
|
|||
let update_store_weak = Arc::downgrade(&update_store);
|
||||
tokio::task::spawn(async move {
|
||||
// Block and wait for something to process.
|
||||
'outer: while let Some(_) = notification_receiver.recv().await {
|
||||
'outer: while notification_receiver.recv().await.is_some() {
|
||||
loop {
|
||||
match update_store_weak.upgrade() {
|
||||
Some(update_store) => {
|
||||
|
@ -276,7 +276,7 @@ where
|
|||
|
||||
updates.extend(failed);
|
||||
|
||||
updates.sort_unstable_by(|a, b| a.id().cmp(&b.id()));
|
||||
updates.sort_by_key(|u| u.id());
|
||||
|
||||
Ok(updates)
|
||||
}
|
||||
|
|
|
@ -216,7 +216,7 @@ impl HeedUuidStore {
|
|||
impl UuidStore for HeedUuidStore {
|
||||
async fn create_uuid(&self, name: String, err: bool) -> Result<Uuid> {
|
||||
let env = self.env.clone();
|
||||
let db = self.db.clone();
|
||||
let db = self.db;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let mut txn = env.write_txn()?;
|
||||
match db.get(&txn, &name)? {
|
||||
|
@ -240,7 +240,7 @@ impl UuidStore for HeedUuidStore {
|
|||
|
||||
async fn get_uuid(&self, name: String) -> Result<Option<Uuid>> {
|
||||
let env = self.env.clone();
|
||||
let db = self.db.clone();
|
||||
let db = self.db;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let txn = env.read_txn()?;
|
||||
match db.get(&txn, &name)? {
|
||||
|
@ -255,7 +255,7 @@ impl UuidStore for HeedUuidStore {
|
|||
|
||||
async fn delete(&self, uid: String) -> Result<Option<Uuid>> {
|
||||
let env = self.env.clone();
|
||||
let db = self.db.clone();
|
||||
let db = self.db;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let mut txn = env.write_txn()?;
|
||||
match db.get(&txn, &uid)? {
|
||||
|
@ -272,7 +272,7 @@ impl UuidStore for HeedUuidStore {
|
|||
|
||||
async fn list(&self) -> Result<Vec<(String, Uuid)>> {
|
||||
let env = self.env.clone();
|
||||
let db = self.db.clone();
|
||||
let db = self.db;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let txn = env.read_txn()?;
|
||||
let mut entries = Vec::new();
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue