230 lines
8.1 KiB
Rust
230 lines
8.1 KiB
Rust
// 负责组合 HTTP 路由和共享接口状态但不直接访问数据库实现
|
|
|
|
mod handlers;
|
|
mod request;
|
|
mod response;
|
|
|
|
use std::{net::SocketAddr, path::PathBuf, sync::Arc};
|
|
|
|
use axum::{Router, routing::get};
|
|
use dht_crawler::DhtRuntimeStats;
|
|
use dht_search::{search::SearchEngine, storage::TorrentRepository};
|
|
use tokio_util::sync::CancellationToken;
|
|
use tower_http::services::{ServeDir, ServeFile};
|
|
|
|
use crate::{crawler::pipeline::PersistenceIngress, verification::VerificationIngress};
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct ApiState {
|
|
pub(crate) repository: Arc<dyn TorrentRepository>,
|
|
pub(crate) search: SearchEngine,
|
|
pub(crate) dht_stats: DhtRuntimeStats,
|
|
pub(crate) persistence: PersistenceIngress,
|
|
pub(crate) verification: Option<VerificationIngress>,
|
|
}
|
|
|
|
pub(crate) async fn serve(
|
|
listen: SocketAddr,
|
|
web_dir: PathBuf,
|
|
state: ApiState,
|
|
cancel: CancellationToken,
|
|
) -> std::io::Result<()> {
|
|
let router = router(state, web_dir);
|
|
let listener = tokio::net::TcpListener::bind(listen).await?;
|
|
tracing::info!(%listen, "HTTP 服务启动");
|
|
axum::serve(listener, router)
|
|
.with_graceful_shutdown(cancel.cancelled_owned())
|
|
.await
|
|
}
|
|
|
|
fn router(state: ApiState, web_dir: PathBuf) -> Router {
|
|
let index = web_dir.join("index.html");
|
|
Router::new()
|
|
.route("/health", get(handlers::health))
|
|
.route("/ready", get(handlers::ready))
|
|
.route("/stats", get(handlers::stats))
|
|
.route("/search", get(handlers::search))
|
|
.route("/contents/{content_key}", get(handlers::content_variants))
|
|
.route("/torrents/{info_hash}", get(handlers::torrent))
|
|
.with_state(state)
|
|
.fallback_service(ServeDir::new(web_dir).fallback(ServeFile::new(index)))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::sync::Arc;
|
|
|
|
use axum::{
|
|
body::{Body, to_bytes},
|
|
http::{Request, StatusCode},
|
|
};
|
|
use dht_crawler::{DhtRuntimeStats, FileInfo, TorrentInfo};
|
|
use dht_search::{
|
|
domain::{InfoHash, TorrentRecord},
|
|
search::SearchEngine,
|
|
storage::{RocksTorrentRepository, TorrentRepository},
|
|
};
|
|
use tempfile::TempDir;
|
|
use tower::ServiceExt;
|
|
|
|
use crate::{crawler::pipeline::PersistencePipeline, verification::VerificationIngress};
|
|
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn search_and_detail_return_user_fields_and_enqueue_verification() {
|
|
let directory = TempDir::new().unwrap();
|
|
let repository =
|
|
Arc::new(RocksTorrentRepository::open(directory.path().join("rocksdb")).unwrap());
|
|
let mut record = TorrentRecord::try_from(TorrentInfo {
|
|
info_hash: "0101010101010101010101010101010101010101".into(),
|
|
magnet_link: String::new(),
|
|
name: "Example Movie".into(),
|
|
total_size: 42,
|
|
files: vec![FileInfo {
|
|
path: "movie.mkv".into(),
|
|
size: 42,
|
|
}],
|
|
piece_length: 16_384,
|
|
peers: vec!["127.0.0.1:6881".into()],
|
|
timestamp: 10,
|
|
})
|
|
.unwrap();
|
|
record.availability = dht_search::domain::Availability::default();
|
|
repository.upsert(record.clone()).unwrap();
|
|
let mut variant = record.clone();
|
|
variant.info_hash = InfoHash::from_bytes([2; 20]);
|
|
variant.name = "Example Movie Alternate".into();
|
|
repository.upsert(variant).unwrap();
|
|
let search = SearchEngine::open(directory.path().join("tantivy")).unwrap();
|
|
search.index_pending(repository.as_ref(), 10, 20).unwrap();
|
|
let repository_trait: Arc<dyn TorrentRepository> = repository.clone();
|
|
let persistence = PersistencePipeline::start(repository_trait.clone(), 4);
|
|
let verification = VerificationIngress::for_test(repository.clone(), 10);
|
|
let web_dir = directory.path().join("web");
|
|
std::fs::create_dir_all(&web_dir).unwrap();
|
|
std::fs::write(web_dir.join("index.html"), "<main>DHT Search</main>").unwrap();
|
|
let app = router(
|
|
ApiState {
|
|
repository: repository_trait,
|
|
search,
|
|
dht_stats: DhtRuntimeStats::default(),
|
|
persistence: persistence.ingress.clone(),
|
|
verification: Some(verification),
|
|
},
|
|
web_dir,
|
|
);
|
|
|
|
let response = app
|
|
.clone()
|
|
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::OK);
|
|
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
|
|
assert_eq!(body.as_ref(), b"<main>DHT Search</main>");
|
|
|
|
let response = app
|
|
.clone()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri("/search?q=Example")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::OK);
|
|
let json: serde_json::Value =
|
|
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
|
|
.unwrap();
|
|
assert_eq!(json["hits"][0]["name"], "Example Movie");
|
|
assert!(json["hits"][0]["heat"]["score"].is_number());
|
|
assert_eq!(json["hits"][0]["availability"]["status"], "unknown");
|
|
assert_eq!(json["hits"][0]["variant_count"], 2);
|
|
assert_eq!(repository.verification_queue_len().unwrap(), 1);
|
|
|
|
let response = app
|
|
.clone()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri(format!("/search?q={}", record.info_hash))
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::OK);
|
|
let json: serde_json::Value =
|
|
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
|
|
.unwrap();
|
|
assert_eq!(json["total"], 1);
|
|
|
|
let response = app
|
|
.clone()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri(format!("/contents/{}", hex::encode(record.content_key)))
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::OK);
|
|
let json: serde_json::Value =
|
|
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
|
|
.unwrap();
|
|
assert_eq!(json["total"], 2);
|
|
assert_eq!(json["variants"].as_array().unwrap().len(), 2);
|
|
|
|
let response = app
|
|
.clone()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri("/search?min_files=3&max_files=1")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
|
|
|
|
let response = app
|
|
.clone()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri("/search?sort=not_a_sort")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
|
|
|
|
let response = app
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri(format!("/torrents/{}", InfoHash::from_bytes([1; 20])))
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.status(), StatusCode::OK);
|
|
let json: serde_json::Value =
|
|
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
|
|
.unwrap();
|
|
assert_eq!(json["name"], "Example Movie");
|
|
assert!(
|
|
json["magnet_link"]
|
|
.as_str()
|
|
.unwrap()
|
|
.starts_with("magnet:?xt=")
|
|
);
|
|
assert_eq!(json["files"][0]["path"], "movie.mkv");
|
|
assert_eq!(repository.verification_queue_len().unwrap(), 1);
|
|
|
|
persistence.close_and_join().await.unwrap();
|
|
}
|
|
}
|