leyoda/rqbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

HTTP API: move functions out to global scope

Browse source
This commit is contained in:
Igor Katson 2025-01-14 09:50:18 +00:00
parent e79b51ea1f
commit c8d4c8d713
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
1 changed files with 426 additions and 428 deletions
Download patch file Download diff file Expand all files Collapse all files

274
crates/librqbit/src/http_api.rs
View file

@ -144,92 +144,31 @@ mod timeout {
use timeout::Timeout;
impl HttpApi {
pub fn new(api: Api, opts: Option<HttpApiOptions>) -> Self {
Self {
inner: api,
opts: opts.unwrap_or_default(),
}
}
/// Run the HTTP server forever on the given address.
/// If read_only is passed, no state-modifying methods will be exposed.
#[inline(never)]
pub fn make_http_api_and_run(
mut self,
listener: TcpListener,
upnp_router: Option<Router>,
) -> BoxFuture<'static, anyhow::Result<()>> {
let state = self.inner;
let api_root = move |parts: Parts| async move {
// If browser, and webui enabled, redirect to web
#[cfg(feature = "webui")]
{
if parts.headers.get("Accept").map_or(false, |h| {
std::str::from_utf8(h.as_bytes()).map_or(false, |s| s.contains("text/html"))
}) {
return Redirect::temporary("./web/").into_response();
}
}
(
[("Content-Type", "application/json")],
axum::Json(serde_json::json!({
"apis": {
"GET /": "list all available APIs",
"GET /dht/stats": "DHT stats",
"GET /dht/table": "DHT routing table",
"GET /torrents": "List torrents",
"GET /torrents/playlist": "Generate M3U8 playlist for all files in all torrents",
"GET /stats": "Global session stats",
"POST /torrents/resolve_magnet": "Resolve a magnet to torrent file bytes",
"GET /torrents/{id_or_infohash}": "Torrent details",
"GET /torrents/{id_or_infohash}/haves": "The bitfield of have pieces",
"GET /torrents/{id_or_infohash}/playlist": "Generate M3U8 playlist for this torrent",
"GET /torrents/{id_or_infohash}/stats/v1": "Torrent stats",
"GET /torrents/{id_or_infohash}/peer_stats": "Per peer stats",
"GET /torrents/{id_or_infohash}/stream/{file_idx}": "Stream a file. Accepts Range header to seek.",
"POST /torrents/{id_or_infohash}/pause": "Pause torrent",
"POST /torrents/{id_or_infohash}/start": "Resume torrent",
"POST /torrents/{id_or_infohash}/forget": "Forget about the torrent, keep the files",
"POST /torrents/{id_or_infohash}/delete": "Forget about the torrent, remove the files",
"POST /torrents/{id_or_infohash}/update_only_files": "Change the selection of files to download. You need to POST json of the following form {\"only_files\": [0, 1, 2]}",
"POST /torrents": "Add a torrent here. magnet: or http:// or a local file.",
"POST /rust_log": "Set RUST_LOG to this post launch (for debugging)",
"GET /web/": "Web UI",
},
"server": "rqbit",
"version": env!("CARGO_PKG_VERSION"),
})),
).into_response()
};
async fn dht_stats(State(state): State<ApiState>) -> Result<impl IntoResponse> {
async fn dht_stats(State(state): State<ApiState>) -> Result<impl IntoResponse> {
state.api_dht_stats().map(axum::Json)
}
}
async fn dht_table(State(state): State<ApiState>) -> Result<impl IntoResponse> {
async fn dht_table(State(state): State<ApiState>) -> Result<impl IntoResponse> {
state.api_dht_table().map(axum::Json)
}
}
async fn session_stats(State(state): State<ApiState>) -> impl IntoResponse {
async fn session_stats(State(state): State<ApiState>) -> impl IntoResponse {
axum::Json(state.api_session_stats())
}
}
async fn torrents_list(
async fn torrents_list(
State(state): State<ApiState>,
Query(opts): Query<ApiTorrentListOpts>,
) -> impl IntoResponse {
) -> impl IntoResponse {
axum::Json(state.api_torrent_list_ext(opts))
}
}
async fn torrents_post(
async fn torrents_post(
State(state): State<ApiState>,
Query(params): Query<TorrentAddQueryParams>,
Timeout(timeout): Timeout<600_000, 3_600_000>,
data: Bytes,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
let is_url = params.is_url;
let opts = params.into_add_torrent_options();
let data = data.to_vec();
@ -265,16 +204,16 @@ impl HttpApi {
.await
.context("timeout")?
.map(axum::Json)
}
}
async fn torrent_details(
async fn torrent_details(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_torrent_details(idx).map(axum::Json)
}
}
fn torrent_playlist_items(handle: &ManagedTorrent) -> Result<Vec<(usize, String)>> {
fn torrent_playlist_items(handle: &ManagedTorrent) -> Result<Vec<(usize, String)>> {
let mut playlist_items = handle
.metadata
.load()
@ -302,22 +241,20 @@ impl HttpApi {
.collect::<Vec<_>>();
playlist_items.sort_by(|left, right| left.1.cmp(&right.1));
Ok(playlist_items)
}
}
fn get_host(headers: &HeaderMap) -> Result<&str> {
fn get_host(headers: &HeaderMap) -> Result<&str> {
Ok(headers
.get("host")
.ok_or_else(|| {
ApiError::new_from_text(StatusCode::BAD_REQUEST, "Missing host header")
})?
.ok_or_else(|| ApiError::new_from_text(StatusCode::BAD_REQUEST, "Missing host header"))?
.to_str()
.context("hostname is not string")?)
}
}
fn build_playlist_content(
fn build_playlist_content(
host: &str,
it: impl IntoIterator<Item = (TorrentIdOrHash, usize, String)>,
) -> impl IntoResponse {
) -> impl IntoResponse {
let body = it
.into_iter()
.map(|(torrent_idx, file_idx, filename)| {
@ -335,14 +272,14 @@ impl HttpApi {
],
format!("#EXTM3U\r\n{body}"), // https://en.wikipedia.org/wiki/M3U
)
}
}
async fn resolve_magnet(
async fn resolve_magnet(
State(state): State<ApiState>,
Timeout(timeout): Timeout<600_000, 3_600_000>,
inp_headers: HeaderMap,
url: String,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
let added = tokio::time::timeout(
timeout,
state.session().add_torrent(
@ -402,13 +339,13 @@ impl HttpApi {
}
}
Ok((headers, content).into_response())
}
}
async fn torrent_playlist(
async fn torrent_playlist(
State(state): State<ApiState>,
headers: HeaderMap,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
let host = get_host(&headers)?;
let playlist_items = torrent_playlist_items(&*state.mgr_handle(idx)?)?;
Ok(build_playlist_content(
@ -417,12 +354,12 @@ impl HttpApi {
.into_iter()
.map(move |(file_idx, filename)| (idx, file_idx, filename)),
))
}
}
async fn global_playlist(
async fn global_playlist(
State(state): State<ApiState>,
headers: HeaderMap,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
let host = get_host(&headers)?;
let all_items = state.session().with_torrents(|torrents| {
torrents
@ -439,50 +376,50 @@ impl HttpApi {
.collect::<Vec<_>>()
});
Ok(build_playlist_content(host, all_items))
}
}
async fn torrent_haves(
async fn torrent_haves(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_dump_haves(idx)
}
}
async fn torrent_stats_v0(
async fn torrent_stats_v0(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_stats_v0(idx).map(axum::Json)
}
}
async fn torrent_stats_v1(
async fn torrent_stats_v1(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_stats_v1(idx).map(axum::Json)
}
}
async fn peer_stats(
async fn peer_stats(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
Query(filter): Query<PeerStatsFilter>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_peer_stats(idx, filter).map(axum::Json)
}
}
#[derive(Deserialize)]
struct StreamPathParams {
#[derive(Deserialize)]
struct StreamPathParams {
id: TorrentIdOrHash,
file_id: usize,
#[serde(rename = "filename")]
_filename: Option<Arc<str>>,
}
}
async fn torrent_stream_file(
async fn torrent_stream_file(
State(state): State<ApiState>,
Path(StreamPathParams { id, file_id, .. }): Path<StreamPathParams>,
headers: http::HeaderMap,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
let mut stream = state.api_stream(id, file_id)?;
let mut status = StatusCode::OK;
let mut output_headers = HeaderMap::new();
@ -560,78 +497,139 @@ impl HttpApi {
let s = tokio_util::io::ReaderStream::with_capacity(stream, 65536);
Ok((status, (output_headers, axum::body::Body::from_stream(s))))
}
}
async fn torrent_action_pause(
async fn torrent_action_pause(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_torrent_action_pause(idx).await.map(axum::Json)
}
}
async fn torrent_action_start(
async fn torrent_action_start(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_torrent_action_start(idx).await.map(axum::Json)
}
}
async fn torrent_action_forget(
async fn torrent_action_forget(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_torrent_action_forget(idx).await.map(axum::Json)
}
}
async fn torrent_action_delete(
async fn torrent_action_delete(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_torrent_action_delete(idx).await.map(axum::Json)
}
}
#[derive(Deserialize)]
struct UpdateOnlyFilesRequest {
#[derive(Deserialize)]
struct UpdateOnlyFilesRequest {
only_files: Vec<usize>,
}
}
async fn torrent_action_update_only_files(
async fn torrent_action_update_only_files(
State(state): State<ApiState>,
Path(idx): Path<TorrentIdOrHash>,
axum::Json(req): axum::Json<UpdateOnlyFilesRequest>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state
.api_torrent_action_update_only_files(idx, &req.only_files.into_iter().collect())
.await
.map(axum::Json)
}
}
async fn set_rust_log(
async fn set_rust_log(
State(state): State<ApiState>,
new_value: String,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.api_set_rust_log(new_value).map(axum::Json)
}
}
async fn stream_logs(State(state): State<ApiState>) -> Result<impl IntoResponse> {
async fn stream_logs(State(state): State<ApiState>) -> Result<impl IntoResponse> {
let s = state.api_log_lines_stream()?.map_err(|e| {
debug!(error=%e, "stream_logs");
e
});
Ok(axum::body::Body::from_stream(s))
}
}
async fn update_session_ratelimits(
async fn update_session_ratelimits(
State(state): State<ApiState>,
Json(limits): Json<LimitsConfig>,
) -> Result<impl IntoResponse> {
) -> Result<impl IntoResponse> {
state.session().ratelimits.set_upload_bps(limits.upload_bps);
state
.session()
.ratelimits
.set_download_bps(limits.download_bps);
Ok(Json(EmptyJsonResponse {}))
}
impl HttpApi {
pub fn new(api: Api, opts: Option<HttpApiOptions>) -> Self {
Self {
inner: api,
opts: opts.unwrap_or_default(),
}
}
/// Run the HTTP server forever on the given address.
/// If read_only is passed, no state-modifying methods will be exposed.
#[inline(never)]
pub fn make_http_api_and_run(
mut self,
listener: TcpListener,
upnp_router: Option<Router>,
) -> BoxFuture<'static, anyhow::Result<()>> {
let state = self.inner;
let api_root = move |parts: Parts| async move {
// If browser, and webui enabled, redirect to web
#[cfg(feature = "webui")]
{
if parts.headers.get("Accept").map_or(false, |h| {
std::str::from_utf8(h.as_bytes()).map_or(false, |s| s.contains("text/html"))
}) {
return Redirect::temporary("./web/").into_response();
}
}
(
[("Content-Type", "application/json")],
axum::Json(serde_json::json!({
"apis": {
"GET /": "list all available APIs",
"GET /dht/stats": "DHT stats",
"GET /dht/table": "DHT routing table",
"GET /torrents": "List torrents",
"GET /torrents/playlist": "Generate M3U8 playlist for all files in all torrents",
"GET /stats": "Global session stats",
"POST /torrents/resolve_magnet": "Resolve a magnet to torrent file bytes",
"GET /torrents/{id_or_infohash}": "Torrent details",
"GET /torrents/{id_or_infohash}/haves": "The bitfield of have pieces",
"GET /torrents/{id_or_infohash}/playlist": "Generate M3U8 playlist for this torrent",
"GET /torrents/{id_or_infohash}/stats/v1": "Torrent stats",
"GET /torrents/{id_or_infohash}/peer_stats": "Per peer stats",
"GET /torrents/{id_or_infohash}/stream/{file_idx}": "Stream a file. Accepts Range header to seek.",
"POST /torrents/{id_or_infohash}/pause": "Pause torrent",
"POST /torrents/{id_or_infohash}/start": "Resume torrent",
"POST /torrents/{id_or_infohash}/forget": "Forget about the torrent, keep the files",
"POST /torrents/{id_or_infohash}/delete": "Forget about the torrent, remove the files",
"POST /torrents/{id_or_infohash}/update_only_files": "Change the selection of files to download. You need to POST json of the following form {\"only_files\": [0, 1, 2]}",
"POST /torrents": "Add a torrent here. magnet: or http:// or a local file.",
"POST /rust_log": "Set RUST_LOG to this post launch (for debugging)",
"GET /web/": "Web UI",
},
"server": "rqbit",
"version": env!("CARGO_PKG_VERSION"),
})),
).into_response()
};
let mut app = Router::new()
.route("/", get(api_root))