diff --git a/src/data/mod.rs b/src/data/mod.rs index 1dec00766..cdb2127a8 100644 --- a/src/data/mod.rs +++ b/src/data/mod.rs @@ -82,8 +82,8 @@ impl Data { self.index_controller.settings(index_uid.as_ref().to_string()).await } - pub fn list_indexes(&self) -> anyhow::Result> { - self.index_controller.list_indexes() + pub async fn list_indexes(&self) -> anyhow::Result> { + self.index_controller.list_indexes().await } pub fn index(&self, name: impl AsRef) -> anyhow::Result> { diff --git a/src/index_controller/index_actor.rs b/src/index_controller/index_actor.rs index c77793a5b..360f3a3c0 100644 --- a/src/index_controller/index_actor.rs +++ b/src/index_controller/index_actor.rs @@ -1,18 +1,19 @@ use std::collections::{hash_map::Entry, HashMap}; -use std::fs::{create_dir_all, File, remove_dir_all}; +use std::fs::{create_dir_all, remove_dir_all, File}; +use std::future::Future; use std::path::{Path, PathBuf}; use std::sync::Arc; use async_stream::stream; -use chrono::Utc; +use chrono::{Utc, DateTime}; use futures::pin_mut; use futures::stream::StreamExt; use heed::EnvOpenOptions; use log::info; -use std::future::Future; use thiserror::Error; use tokio::sync::{mpsc, oneshot, RwLock}; use uuid::Uuid; +use serde::{Serialize, Deserialize}; use super::get_arc_ownership_blocking; use super::update_handler::UpdateHandler; @@ -20,7 +21,7 @@ use crate::index::UpdateResult as UResult; use crate::index::{Document, Index, SearchQuery, SearchResult, Settings}; use crate::index_controller::{ updates::{Failed, Processed, Processing}, - IndexMetadata, UpdateMeta, + UpdateMeta, }; use crate::option::IndexerOpts; @@ -28,11 +29,20 @@ pub type Result = std::result::Result; type AsyncMap = Arc>>; type UpdateResult = std::result::Result, Failed>; +#[derive(Debug, Serialize, Deserialize, Clone)] +#[serde(rename_all = "camelCase")] +pub struct IndexMeta{ + uuid: Uuid, + created_at: DateTime, + updated_at: DateTime, + primary_key: Option, +} + enum IndexMsg { CreateIndex { uuid: Uuid, primary_key: Option, - ret: oneshot::Sender>, + ret: oneshot::Sender>, }, Update { meta: Processing, @@ -65,6 +75,10 @@ enum IndexMsg { uuid: Uuid, ret: oneshot::Sender>, }, + GetMeta { + uuid: Uuid, + ret: oneshot::Sender>>, + }, } struct IndexActor { @@ -84,10 +98,11 @@ pub enum IndexError { #[async_trait::async_trait] trait IndexStore { - async fn create_index(&self, uuid: Uuid, primary_key: Option) -> Result; + async fn create_index(&self, uuid: Uuid, primary_key: Option) -> Result; async fn get_or_create(&self, uuid: Uuid) -> Result; async fn get(&self, uuid: Uuid) -> Result>; async fn delete(&self, uuid: &Uuid) -> Result>; + async fn get_meta(&self, uuid: &Uuid) -> Result>; } impl IndexActor { @@ -150,10 +165,6 @@ impl IndexActor { let fut1: Box + Unpin + Send> = Box::new(fut1); let fut2: Box + Unpin + Send> = Box::new(fut2); - //let futures = futures::stream::futures_unordered::FuturesUnordered::new(); - //futures.push(fut1); - //futures.push(fut2); - //futures.for_each(f) tokio::join!(fut1, fut2); } @@ -189,7 +200,10 @@ impl IndexActor { } Delete { uuid, ret } => { let _ = ret.send(self.handle_delete(uuid).await); - }, + } + GetMeta { uuid, ret } => { + let _ = ret.send(self.handle_get_meta(uuid).await); + } } } @@ -210,7 +224,7 @@ impl IndexActor { &self, uuid: Uuid, primary_key: Option, - ret: oneshot::Sender>, + ret: oneshot::Sender>, ) { let result = self.store.create_index(uuid, primary_key).await; let _ = ret.send(result); @@ -293,6 +307,11 @@ impl IndexActor { Ok(()) } + + async fn handle_get_meta(&self, uuid: Uuid) -> Result> { + let result = self.store.get_meta(&uuid).await?; + Ok(result) + } } #[derive(Clone)] @@ -319,7 +338,7 @@ impl IndexActorHandle { &self, uuid: Uuid, primary_key: Option, - ) -> Result { + ) -> Result { let (ret, receiver) = oneshot::channel(); let msg = IndexMsg::CreateIndex { ret, @@ -393,20 +412,27 @@ impl IndexActorHandle { let _ = self.read_sender.send(msg).await; Ok(receiver.await.expect("IndexActor has been killed")?) } + + pub async fn get_index_meta(&self, uuid: Uuid) -> Result> { + let (ret, receiver) = oneshot::channel(); + let msg = IndexMsg::GetMeta { uuid, ret }; + let _ = self.read_sender.send(msg).await; + Ok(receiver.await.expect("IndexActor has been killed")?) + } } struct MapIndexStore { root: PathBuf, - meta_store: AsyncMap, + meta_store: AsyncMap, index_store: AsyncMap, } #[async_trait::async_trait] impl IndexStore for MapIndexStore { - async fn create_index(&self, uuid: Uuid, primary_key: Option) -> Result { + async fn create_index(&self, uuid: Uuid, primary_key: Option) -> Result { let meta = match self.meta_store.write().await.entry(uuid.clone()) { Entry::Vacant(entry) => { - let meta = IndexMetadata { + let meta = IndexMeta{ uuid, created_at: Utc::now(), updated_at: Utc::now(), @@ -464,6 +490,10 @@ impl IndexStore for MapIndexStore { } Ok(index) } + + async fn get_meta(&self, uuid: &Uuid) -> Result> { + Ok(self.meta_store.read().await.get(uuid).cloned()) + } } impl MapIndexStore { diff --git a/src/index_controller/mod.rs b/src/index_controller/mod.rs index 19fe62f4d..09a979cbd 100644 --- a/src/index_controller/mod.rs +++ b/src/index_controller/mod.rs @@ -11,7 +11,6 @@ use std::time::Duration; use actix_web::web::{Bytes, Payload}; use anyhow::Context; -use chrono::{DateTime, Utc}; use futures::stream::StreamExt; use milli::update::{IndexDocumentsMethod, UpdateFormat}; use serde::{Serialize, Deserialize}; @@ -28,10 +27,9 @@ pub type UpdateStatus = updates::UpdateStatus; #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct IndexMetadata { - uuid: Uuid, - created_at: DateTime, - updated_at: DateTime, - primary_key: Option, + name: String, + #[serde(flatten)] + meta: index_actor::IndexMeta, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -144,9 +142,12 @@ impl IndexController { pub async fn create_index(&self, index_settings: IndexSettings) -> anyhow::Result { let IndexSettings { name, primary_key } = index_settings; - let uuid = self.uuid_resolver.create(name.unwrap()).await?; - let index_meta = self.index_handle.create_index(uuid, primary_key).await?; - Ok(index_meta) + let name = name.unwrap(); + let uuid = self.uuid_resolver.create(name.clone()).await?; + let meta = self.index_handle.create_index(uuid, primary_key).await?; + let meta = IndexMetadata { name, meta }; + + Ok(meta) } pub async fn delete_index(&self, index_uid: String) -> anyhow::Result<()> { @@ -176,8 +177,19 @@ impl IndexController { Ok(result) } - pub fn list_indexes(&self) -> anyhow::Result> { - todo!() + pub async fn list_indexes(&self) -> anyhow::Result> { + let uuids = self.uuid_resolver.list().await?; + + let mut ret = Vec::new(); + + for (name, uuid) in uuids { + if let Some(meta) = self.index_handle.get_index_meta(uuid).await? { + let meta = IndexMetadata { name, meta }; + ret.push(meta); + } + } + + Ok(ret) } pub async fn settings(&self, index: String) -> anyhow::Result { diff --git a/src/index_controller/uuid_resolver.rs b/src/index_controller/uuid_resolver.rs index 50740f30f..e2539fdb2 100644 --- a/src/index_controller/uuid_resolver.rs +++ b/src/index_controller/uuid_resolver.rs @@ -26,6 +26,9 @@ enum UuidResolveMsg { name: String, ret: oneshot::Sender>>, }, + List { + ret: oneshot::Sender>>, + }, } struct UuidResolverActor { @@ -50,6 +53,9 @@ impl UuidResolverActor { Some(GetOrCreate { name, ret }) => self.handle_get_or_create(name, ret).await, Some(Resolve { name, ret }) => self.handle_resolve(name, ret).await, Some(Delete { name, ret }) => self.handle_delete(name, ret).await, + Some(List { ret }) => { + let _ = ret.send(self.handle_list().await); + } // all senders have been dropped, need to quit. None => break, } @@ -77,6 +83,11 @@ impl UuidResolverActor { let result = self.store.delete(&name).await; let _ = ret.send(result); } + + async fn handle_list(&self) -> Result> { + let result = self.store.list().await?; + Ok(result) + } } #[derive(Clone)] @@ -120,6 +131,13 @@ impl UuidResolverHandle { let _ = self.sender.send(msg).await; Ok(receiver.await.expect("Uuid resolver actor has been killed")?) } + + pub async fn list(&self) -> anyhow::Result> { + let (ret, receiver) = oneshot::channel(); + let msg = UuidResolveMsg::List { ret }; + let _ = self.sender.send(msg).await; + Ok(receiver.await.expect("Uuid resolver actor has been killed")?) + } } #[derive(Clone, Debug, Error)] @@ -135,6 +153,7 @@ trait UuidStore { async fn create_uuid(&self, name: String, err: bool) -> Result; async fn get_uuid(&self, name: &str) -> Result>; async fn delete(&self, name: &str) -> Result>; + async fn list(&self) -> Result>; } struct MapUuidStore(Arc>>); @@ -165,4 +184,9 @@ impl UuidStore for MapUuidStore { async fn delete(&self, name: &str) -> Result> { Ok(self.0.write().await.remove(name)) } + + async fn list(&self) -> Result> { + let list = self.0.read().await.iter().map(|(name, uuid)| (name.to_owned(), uuid.clone())).collect(); + Ok(list) + } } diff --git a/src/routes/index.rs b/src/routes/index.rs index c9200f085..752493a3b 100644 --- a/src/routes/index.rs +++ b/src/routes/index.rs @@ -21,7 +21,7 @@ pub fn services(cfg: &mut web::ServiceConfig) { #[get("/indexes", wrap = "Authentication::Private")] async fn list_indexes(data: web::Data) -> Result { - match data.list_indexes() { + match data.list_indexes().await { Ok(indexes) => { let json = serde_json::to_string(&indexes).unwrap(); Ok(HttpResponse::Ok().body(&json))