653 lines
23 KiB
Rust
653 lines
23 KiB
Rust
//! Soulseek network client: server connection, peer connections, search,
|
|
//! downloads.
|
|
//!
|
|
//! Behavioral reference: Soulseek.NET `SoulseekClient.cs` and `src/Network/**`.
|
|
//!
|
|
//! M1: connect + login, peer listener, indirect (PierceFirewall)
|
|
//! connections both directions, live search.
|
|
//! M2: downloads — TransferRequest/TransferResponse dance, F-type transfer
|
|
//! connections (both directions), resume via offset, download manager.
|
|
|
|
mod conn;
|
|
mod distributed;
|
|
mod responder;
|
|
mod share;
|
|
mod transfer;
|
|
mod upload;
|
|
mod upnp;
|
|
|
|
pub use share::ShareIndex;
|
|
pub use transfer::{DownloadState, DownloadUpdate};
|
|
pub use upload::{UploadState, UploadUpdate};
|
|
|
|
use std::collections::{HashMap, HashSet, VecDeque};
|
|
use std::net::{Ipv4Addr, SocketAddrV4};
|
|
use std::path::PathBuf;
|
|
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
|
|
use std::sync::{Arc, Mutex, RwLock};
|
|
use tokio::net::TcpStream;
|
|
use tokio::sync::{mpsc, oneshot};
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
use soulseek_proto::{
|
|
peer, server,
|
|
types::{Directory, SearchResponse},
|
|
};
|
|
|
|
pub use soulseek_proto::server::RoomUser;
|
|
pub use soulseek_proto::types::Directory as BrowseDirectory;
|
|
|
|
pub const DEFAULT_SERVER: (&str, u16) = ("server.slsknet.org", 2242);
|
|
pub const DEFAULT_LISTEN_PORT: u16 = 50300;
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct Config {
|
|
pub server_host: String,
|
|
pub server_port: u16,
|
|
pub listen_port: u16,
|
|
pub downloads_dir: PathBuf,
|
|
/// Folders we share with the network.
|
|
pub shared_folders: Vec<PathBuf>,
|
|
/// Concurrent upload slots offered to peers.
|
|
pub upload_slots: u32,
|
|
/// User-info card description.
|
|
pub description: String,
|
|
}
|
|
|
|
impl Default for Config {
|
|
fn default() -> Self {
|
|
Config {
|
|
server_host: DEFAULT_SERVER.0.into(),
|
|
server_port: DEFAULT_SERVER.1,
|
|
listen_port: DEFAULT_LISTEN_PORT,
|
|
downloads_dir: PathBuf::from("downloads"),
|
|
shared_folders: Vec::new(),
|
|
upload_slots: 2,
|
|
description: "SoulStream user".into(),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub enum Event {
|
|
LoggedIn {
|
|
greeting: String,
|
|
own_ip: Ipv4Addr,
|
|
},
|
|
LoginFailed {
|
|
reason: String,
|
|
},
|
|
Disconnected {
|
|
reason: String,
|
|
},
|
|
SearchResult {
|
|
token: u32,
|
|
response: SearchResponse,
|
|
},
|
|
ExcludedSearchPhrases(Vec<String>),
|
|
Download(DownloadUpdate),
|
|
Upload(UploadUpdate),
|
|
/// We answered a peer's search with N files (for activity display).
|
|
RespondedToSearch {
|
|
username: String,
|
|
files: usize,
|
|
},
|
|
// ---- rooms / chat / users (the Soulseek hub's seats) ----
|
|
RoomList {
|
|
rooms: Vec<(String, u32)>,
|
|
},
|
|
RoomJoined {
|
|
room: String,
|
|
users: Vec<RoomUser>,
|
|
},
|
|
RoomLeft {
|
|
room: String,
|
|
},
|
|
RoomUserJoined {
|
|
room: String,
|
|
user: RoomUser,
|
|
},
|
|
RoomUserLeft {
|
|
room: String,
|
|
username: String,
|
|
},
|
|
RoomMessage {
|
|
room: String,
|
|
username: String,
|
|
message: String,
|
|
},
|
|
/// Already acked to the server by the time it is emitted.
|
|
PrivateMessage {
|
|
id: u32,
|
|
timestamp: u32,
|
|
username: String,
|
|
message: String,
|
|
},
|
|
UserWatched {
|
|
username: String,
|
|
exists: bool,
|
|
status: u32,
|
|
avg_speed: u32,
|
|
file_count: u32,
|
|
dir_count: u32,
|
|
},
|
|
UserStatus {
|
|
username: String,
|
|
status: u32,
|
|
privileged: bool,
|
|
},
|
|
UserStats {
|
|
username: String,
|
|
avg_speed: u32,
|
|
file_count: u32,
|
|
dir_count: u32,
|
|
},
|
|
}
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum ClientError {
|
|
#[error("io: {0}")]
|
|
Io(#[from] std::io::Error),
|
|
#[error("protocol: {0}")]
|
|
Proto(#[from] soulseek_proto::ProtoError),
|
|
#[error("not connected to server")]
|
|
NotConnected,
|
|
#[error("login failed: {0}")]
|
|
LoginFailed(String),
|
|
#[error("user {0} is offline or unknown")]
|
|
UserOffline(String),
|
|
#[error("could not reach peer {0}")]
|
|
PeerUnreachable(String),
|
|
#[error("transfer rejected: {0}")]
|
|
TransferRejected(String),
|
|
#[error("timed out")]
|
|
Timeout,
|
|
}
|
|
|
|
pub type Result<T> = std::result::Result<T, ClientError>;
|
|
|
|
/// Write handle for an established peer message ("P") connection.
|
|
#[derive(Clone)]
|
|
pub(crate) struct PeerHandle {
|
|
pub write_tx: mpsc::UnboundedSender<Vec<u8>>,
|
|
}
|
|
|
|
/// Peer's answer to our TransferRequest(download).
|
|
#[derive(Debug, Clone)]
|
|
pub(crate) struct TransferAck {
|
|
pub allowed: bool,
|
|
pub size: u64,
|
|
pub message: String,
|
|
}
|
|
|
|
/// Peer's TransferRequest(upload) once we reach their queue front —
|
|
/// or the denial that arrived instead.
|
|
pub(crate) type UploadOffer = std::result::Result<(u32, u64), String>; // (remote token, size)
|
|
|
|
/// A peer's user-info card as they sent it. The picture is untouched bytes:
|
|
/// the app decides whether and how to decode it.
|
|
#[derive(Debug, Clone)]
|
|
pub struct UserInfo {
|
|
pub description: String,
|
|
pub picture: Option<Vec<u8>>,
|
|
pub total_uploads: u32,
|
|
pub queue_size: u32,
|
|
pub has_free_slot: bool,
|
|
}
|
|
|
|
pub(crate) struct Inner {
|
|
pub server_tx: mpsc::UnboundedSender<Vec<u8>>,
|
|
pub event_tx: mpsc::UnboundedSender<Event>,
|
|
pub username: Mutex<String>,
|
|
pub active_searches: Mutex<HashSet<u32>>,
|
|
pub pending_login: Mutex<Option<oneshot::Sender<std::result::Result<String, String>>>>,
|
|
pub token_counter: AtomicU32,
|
|
pub listen_port: u16,
|
|
pub downloads_dir: PathBuf,
|
|
|
|
/// GetPeerAddress requests awaiting a server answer, by username.
|
|
pub pending_addresses: Mutex<HashMap<String, Vec<oneshot::Sender<SocketAddrV4>>>>,
|
|
/// Last known endpoint per user.
|
|
pub address_cache: Mutex<HashMap<String, SocketAddrV4>>,
|
|
/// Our indirect-connection solicitations: token → deliver the pierced stream.
|
|
pub pending_pierce: Mutex<HashMap<u32, oneshot::Sender<TcpStream>>>,
|
|
/// Inbound transfer connections we expect: (username, remote token) → stream.
|
|
pub expected_transfers: Mutex<HashMap<(String, u32), oneshot::Sender<TcpStream>>>,
|
|
/// Live peer message connections.
|
|
pub peers: Mutex<HashMap<String, PeerHandle>>,
|
|
/// Browse requests awaiting a peer's share listing, by username.
|
|
pub pending_browses: Mutex<HashMap<String, Vec<oneshot::Sender<Vec<Directory>>>>>,
|
|
pub pending_user_infos: Mutex<HashMap<String, Vec<oneshot::Sender<UserInfo>>>>,
|
|
pub pending_interests: Mutex<HashMap<String, Vec<oneshot::Sender<(Vec<String>, Vec<String>)>>>>,
|
|
pub pending_similar: Mutex<Vec<oneshot::Sender<Vec<(String, u32)>>>>,
|
|
/// TransferResponse waits: (username, our token).
|
|
pub transfer_waits: Mutex<HashMap<(String, u32), oneshot::Sender<TransferAck>>>,
|
|
/// TransferRequest(upload) waits: (username, remote filename).
|
|
pub upload_offers: Mutex<HashMap<(String, String), oneshot::Sender<UploadOffer>>>,
|
|
/// Active downloads by (username, remote filename) → (id, cancel).
|
|
pub downloads_by_file: Mutex<HashMap<(String, String), (u64, CancellationToken)>>,
|
|
pub download_counter: AtomicU64,
|
|
|
|
// ---- M5: sharing / uploads / distributed ----
|
|
/// Files we share with the network.
|
|
pub shares: Arc<ShareIndex>,
|
|
/// Excluded phrases from the server; outgoing search results filter these.
|
|
pub excluded_phrases: RwLock<Vec<String>>,
|
|
/// Our user-info description.
|
|
pub description: String,
|
|
/// Total upload slots and how many are in use.
|
|
pub upload_slots: u32,
|
|
pub uploads_active: AtomicU32,
|
|
/// Pending upload queue (username, remote path), FIFO.
|
|
pub upload_queue: Mutex<VecDeque<(String, String)>>,
|
|
/// Rolling average upload speed reported to peers (bytes/sec).
|
|
pub upload_speed: AtomicU32,
|
|
pub upload_counter: AtomicU64,
|
|
/// Whether we currently have a distributed parent.
|
|
pub has_parent: AtomicBool,
|
|
/// De-dupe relayed searches so we don't answer the same token twice.
|
|
pub answered_searches: Mutex<HashSet<u32>>,
|
|
/// Count of relayed searches we've seen (instrumentation).
|
|
pub searches_seen: AtomicU64,
|
|
}
|
|
|
|
impl Inner {
|
|
pub fn next_token(&self) -> u32 {
|
|
self.token_counter.fetch_add(1, Ordering::Relaxed)
|
|
}
|
|
|
|
pub fn own_username(&self) -> String {
|
|
self.username.lock().unwrap().clone()
|
|
}
|
|
|
|
pub fn send_server(&self, frame: Vec<u8>) -> Result<()> {
|
|
self.server_tx
|
|
.send(frame)
|
|
.map_err(|_| ClientError::NotConnected)
|
|
}
|
|
|
|
pub fn excluded(&self) -> Vec<String> {
|
|
self.excluded_phrases.read().unwrap().clone()
|
|
}
|
|
|
|
pub fn has_free_upload_slot(&self) -> bool {
|
|
self.uploads_active.load(Ordering::Relaxed) < self.upload_slots
|
|
}
|
|
}
|
|
|
|
/// Handle to a running client. Dropping it does not tear down the
|
|
/// connection; call [`Client::shutdown`].
|
|
pub struct Client {
|
|
inner: Arc<Inner>,
|
|
shutdown_tx: mpsc::UnboundedSender<()>,
|
|
listener_shutdown_tx: mpsc::UnboundedSender<()>,
|
|
}
|
|
|
|
impl Client {
|
|
/// Connect the server socket and start the peer listener.
|
|
/// Events (search results, download progress) arrive on the receiver.
|
|
pub async fn connect(config: Config) -> Result<(Client, mpsc::UnboundedReceiver<Event>)> {
|
|
let (event_tx, event_rx) = mpsc::unbounded_channel();
|
|
let (server_tx, server_rx) = mpsc::unbounded_channel::<Vec<u8>>();
|
|
let (shutdown_tx, shutdown_rx) = mpsc::unbounded_channel::<()>();
|
|
let (listener_shutdown_tx, listener_shutdown_rx) = mpsc::unbounded_channel::<()>();
|
|
|
|
let stream = TcpStream::connect((config.server_host.as_str(), config.server_port)).await?;
|
|
stream.set_nodelay(true).ok();
|
|
|
|
// Bind the listener before login so SetListenPort is accurate.
|
|
let listener = conn::bind_listener(config.listen_port).await?;
|
|
let listen_port = listener.local_addr()?.port();
|
|
|
|
// Seed tokens off the clock so restarts don't reuse recent tokens.
|
|
let seed = std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|d| d.subsec_nanos())
|
|
.unwrap_or(1);
|
|
|
|
// Build the share index up front (blocking walk on a worker thread).
|
|
let shares = Arc::new(ShareIndex::new());
|
|
if !config.shared_folders.is_empty() {
|
|
let shares2 = shares.clone();
|
|
let folders = config.shared_folders.clone();
|
|
let (dirs, files) = tokio::task::spawn_blocking(move || shares2.rebuild(&folders))
|
|
.await
|
|
.unwrap_or((0, 0));
|
|
log::info!("share index: {files} files in {dirs} folders");
|
|
}
|
|
|
|
let inner = Arc::new(Inner {
|
|
server_tx,
|
|
event_tx,
|
|
username: Mutex::new(String::new()),
|
|
active_searches: Mutex::new(HashSet::new()),
|
|
pending_login: Mutex::new(None),
|
|
token_counter: AtomicU32::new(seed),
|
|
listen_port,
|
|
downloads_dir: config.downloads_dir.clone(),
|
|
pending_addresses: Mutex::new(HashMap::new()),
|
|
address_cache: Mutex::new(HashMap::new()),
|
|
pending_pierce: Mutex::new(HashMap::new()),
|
|
expected_transfers: Mutex::new(HashMap::new()),
|
|
peers: Mutex::new(HashMap::new()),
|
|
pending_browses: Mutex::new(HashMap::new()),
|
|
pending_user_infos: Mutex::new(HashMap::new()),
|
|
pending_interests: Mutex::new(HashMap::new()),
|
|
pending_similar: Mutex::new(Vec::new()),
|
|
transfer_waits: Mutex::new(HashMap::new()),
|
|
upload_offers: Mutex::new(HashMap::new()),
|
|
downloads_by_file: Mutex::new(HashMap::new()),
|
|
download_counter: AtomicU64::new(1),
|
|
shares,
|
|
excluded_phrases: RwLock::new(Vec::new()),
|
|
description: config.description.clone(),
|
|
upload_slots: config.upload_slots,
|
|
uploads_active: AtomicU32::new(0),
|
|
upload_queue: Mutex::new(VecDeque::new()),
|
|
upload_speed: AtomicU32::new(0),
|
|
upload_counter: AtomicU64::new(1),
|
|
has_parent: AtomicBool::new(false),
|
|
answered_searches: Mutex::new(HashSet::new()),
|
|
searches_seen: AtomicU64::new(0),
|
|
});
|
|
|
|
conn::spawn_server_io(stream, server_rx, shutdown_rx, inner.clone());
|
|
conn::spawn_listener(listener, inner.clone(), listener_shutdown_rx);
|
|
|
|
// Best-effort UPnP mapping for direct connectivity (non-fatal).
|
|
tokio::spawn(upnp::try_map_port(listen_port));
|
|
|
|
Ok((
|
|
Client {
|
|
inner,
|
|
shutdown_tx,
|
|
listener_shutdown_tx,
|
|
},
|
|
event_rx,
|
|
))
|
|
}
|
|
|
|
/// Log in; resolves when the server answers.
|
|
pub async fn login(&self, username: &str, password: &str) -> Result<String> {
|
|
let (tx, rx) = oneshot::channel();
|
|
*self.inner.pending_login.lock().unwrap() = Some(tx);
|
|
*self.inner.username.lock().unwrap() = username.to_string();
|
|
|
|
self.inner.send_server(server::login(username, password))?;
|
|
|
|
let outcome = tokio::time::timeout(std::time::Duration::from_secs(15), rx)
|
|
.await
|
|
.map_err(|_| ClientError::Timeout)?
|
|
.map_err(|_| ClientError::NotConnected)?;
|
|
|
|
match outcome {
|
|
Ok(greeting) => {
|
|
// Post-login bookkeeping the server expects from clients.
|
|
self.inner
|
|
.send_server(server::set_listen_port(self.inner.listen_port as u32))?;
|
|
let (dirs, files) = self.inner.shares.counts();
|
|
self.inner
|
|
.send_server(server::shared_folders_and_files(dirs, files))?;
|
|
// Join the distributed network as a child so we receive
|
|
// relayed searches to answer.
|
|
self.inner.send_server(server::have_no_parents(true))?;
|
|
self.inner
|
|
.send_server(server::branch_root(&self.inner.own_username()))?;
|
|
self.inner.send_server(server::branch_level(0))?;
|
|
self.inner.send_server(server::accept_children(false))?;
|
|
Ok(greeting)
|
|
}
|
|
Err(reason) => Err(ClientError::LoginFailed(reason)),
|
|
}
|
|
}
|
|
|
|
/// Re-index the shared folders and report the new counts to the server.
|
|
pub async fn update_shares(&self, folders: Vec<PathBuf>) -> (u32, u32) {
|
|
let shares = self.inner.shares.clone();
|
|
let (dirs, files) = tokio::task::spawn_blocking(move || shares.rebuild(&folders))
|
|
.await
|
|
.unwrap_or((0, 0));
|
|
let _ = self
|
|
.inner
|
|
.send_server(server::shared_folders_and_files(dirs, files));
|
|
(dirs, files)
|
|
}
|
|
|
|
/// Current (folders, files) share counts.
|
|
pub fn share_counts(&self) -> (u32, u32) {
|
|
self.inner.shares.counts()
|
|
}
|
|
|
|
/// How many relayed searches we've received (whether or not matched).
|
|
pub fn searches_seen(&self) -> u64 {
|
|
self.inner.searches_seen.load(Ordering::Relaxed)
|
|
}
|
|
|
|
/// Seed a peer's endpoint, bypassing server address resolution. Useful
|
|
/// for LAN peers and for tests where two instances share one public IP.
|
|
pub fn set_peer_address(&self, username: &str, addr: SocketAddrV4) {
|
|
self.inner
|
|
.address_cache
|
|
.lock()
|
|
.unwrap()
|
|
.insert(username.to_string(), addr);
|
|
}
|
|
|
|
/// Whether we currently hold a distributed parent connection.
|
|
pub fn has_parent(&self) -> bool {
|
|
self.inner.has_parent.load(Ordering::Relaxed)
|
|
}
|
|
|
|
/// Start a network-wide file search. Results stream out as
|
|
/// [`Event::SearchResult`] carrying this token.
|
|
pub fn search(&self, query: &str) -> Result<u32> {
|
|
let token = self.inner.next_token();
|
|
self.inner.active_searches.lock().unwrap().insert(token);
|
|
self.inner.send_server(server::file_search(token, query))?;
|
|
Ok(token)
|
|
}
|
|
|
|
/// Stop routing results for a search token.
|
|
pub fn stop_search(&self, token: u32) {
|
|
self.inner.active_searches.lock().unwrap().remove(&token);
|
|
}
|
|
|
|
/// Start downloading `filename` (full remote path) from `username`.
|
|
/// Progress streams out as [`Event::Download`]. Returns the download id.
|
|
pub fn download(&self, username: &str, filename: &str, size: u64) -> u64 {
|
|
transfer::start_download(self.inner.clone(), username, filename, size, None, |_| {})
|
|
}
|
|
|
|
/// Like [`download`](Self::download) but writing to an explicit local
|
|
/// path (used for stream-to-cache).
|
|
pub fn download_to(&self, username: &str, filename: &str, size: u64, target: PathBuf) -> u64 {
|
|
transfer::start_download(self.inner.clone(), username, filename, size, Some(target), |_| {})
|
|
}
|
|
|
|
/// `download_to`, with `claim` run on the new id BEFORE the transfer task
|
|
/// is spawned. A download emits `Connecting` the instant it starts, so a
|
|
/// caller that must recognise its own transfers — the import tool's
|
|
/// audition pen — has to be able to register the id first. Doing it on the
|
|
/// returned value is a race the emitter usually wins.
|
|
pub fn download_to_claimed(
|
|
&self,
|
|
username: &str,
|
|
filename: &str,
|
|
size: u64,
|
|
target: PathBuf,
|
|
claim: impl FnOnce(u64),
|
|
) -> u64 {
|
|
transfer::start_download(self.inner.clone(), username, filename, size, Some(target), claim)
|
|
}
|
|
|
|
/// Cancel an active download by id.
|
|
pub fn cancel_download(&self, id: u64) {
|
|
let map = self.inner.downloads_by_file.lock().unwrap();
|
|
for (did, cancel) in map.values() {
|
|
if *did == id {
|
|
cancel.cancel();
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---- rooms / chat / users ----
|
|
|
|
/// Join (or create) a public room. The roster arrives as
|
|
/// [`Event::RoomJoined`]; traffic as [`Event::RoomMessage`].
|
|
pub fn join_room(&self, room: &str) -> Result<()> {
|
|
self.inner.send_server(server::join_room(room))
|
|
}
|
|
|
|
pub fn leave_room(&self, room: &str) -> Result<()> {
|
|
self.inner.send_server(server::leave_room(room))
|
|
}
|
|
|
|
pub fn say_room(&self, room: &str, message: &str) -> Result<()> {
|
|
self.inner.send_server(server::say_chatroom(room, message))
|
|
}
|
|
|
|
/// Send a private message. Delivery/acks are the server's business;
|
|
/// incoming replies arrive as [`Event::PrivateMessage`].
|
|
pub fn message_user(&self, username: &str, message: &str) -> Result<()> {
|
|
self.inner.send_server(server::message_user(username, message))
|
|
}
|
|
|
|
/// Watch a user: stats now, status pushes until unwatched.
|
|
pub fn watch_user(&self, username: &str) -> Result<()> {
|
|
self.inner.send_server(server::watch_user(username))
|
|
}
|
|
|
|
pub fn unwatch_user(&self, username: &str) -> Result<()> {
|
|
self.inner.send_server(server::unwatch_user(username))
|
|
}
|
|
|
|
pub fn get_user_stats(&self, username: &str) -> Result<()> {
|
|
self.inner.send_server(server::get_user_stats(username))
|
|
}
|
|
|
|
/// Ask for the public room list (also pushed unprompted after login).
|
|
pub fn request_room_list(&self) -> Result<()> {
|
|
self.inner.send_server(server::room_list())
|
|
}
|
|
|
|
/// Fetch a user's full shared listing over a peer connection. Slow for
|
|
/// big shares — the reply is one large compressed frame — so this is
|
|
/// spawn-shaped: the returned future owns everything it needs and the
|
|
/// caller can drop whatever lock guarded the `Client` before awaiting.
|
|
pub fn browse(
|
|
&self,
|
|
username: &str,
|
|
) -> impl std::future::Future<Output = Result<Vec<Directory>>> + Send + 'static {
|
|
let inner = self.inner.clone();
|
|
let username = username.to_string();
|
|
async move {
|
|
let handle = conn::get_or_connect_peer(&inner, &username).await?;
|
|
let (tx, rx) = oneshot::channel();
|
|
inner
|
|
.pending_browses
|
|
.lock()
|
|
.unwrap()
|
|
.entry(username.clone())
|
|
.or_default()
|
|
.push(tx);
|
|
handle
|
|
.write_tx
|
|
.send(peer::browse_request())
|
|
.map_err(|_| ClientError::PeerUnreachable(username.clone()))?;
|
|
|
|
match tokio::time::timeout(std::time::Duration::from_secs(45), rx).await {
|
|
Ok(Ok(dirs)) => Ok(dirs),
|
|
_ => {
|
|
inner.pending_browses.lock().unwrap().remove(&username);
|
|
Err(ClientError::Timeout)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A user's info card over a peer connection — spawn-shaped like
|
|
/// `browse`: drop whatever lock guarded the `Client` before awaiting.
|
|
pub fn user_info(
|
|
&self,
|
|
username: &str,
|
|
) -> impl std::future::Future<Output = Result<UserInfo>> + Send + 'static {
|
|
let inner = self.inner.clone();
|
|
let username = username.to_string();
|
|
async move {
|
|
let handle = conn::get_or_connect_peer(&inner, &username).await?;
|
|
let (tx, rx) = oneshot::channel();
|
|
inner
|
|
.pending_user_infos
|
|
.lock()
|
|
.unwrap()
|
|
.entry(username.clone())
|
|
.or_default()
|
|
.push(tx);
|
|
handle
|
|
.write_tx
|
|
.send(peer::user_info_request())
|
|
.map_err(|_| ClientError::PeerUnreachable(username.clone()))?;
|
|
match tokio::time::timeout(std::time::Duration::from_secs(30), rx).await {
|
|
Ok(Ok(info)) => Ok(info),
|
|
_ => {
|
|
inner.pending_user_infos.lock().unwrap().remove(&username);
|
|
Err(ClientError::Timeout)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A user's declared likes and hates, from the server. Spawn-shaped.
|
|
pub fn user_interests(
|
|
&self,
|
|
username: &str,
|
|
) -> impl std::future::Future<Output = Result<(Vec<String>, Vec<String>)>> + Send + 'static {
|
|
let inner = self.inner.clone();
|
|
let username = username.to_string();
|
|
async move {
|
|
let (tx, rx) = oneshot::channel();
|
|
inner
|
|
.pending_interests
|
|
.lock()
|
|
.unwrap()
|
|
.entry(username.clone())
|
|
.or_default()
|
|
.push(tx);
|
|
inner.send_server(server::user_interests(&username))?;
|
|
match tokio::time::timeout(std::time::Duration::from_secs(20), rx).await {
|
|
Ok(Ok(v)) => Ok(v),
|
|
_ => {
|
|
inner.pending_interests.lock().unwrap().remove(&username);
|
|
Err(ClientError::Timeout)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Users the server ranks as similar to us by shared interests. Spawn-shaped.
|
|
pub fn similar_users(
|
|
&self,
|
|
) -> impl std::future::Future<Output = Result<Vec<(String, u32)>>> + Send + 'static {
|
|
let inner = self.inner.clone();
|
|
async move {
|
|
let (tx, rx) = oneshot::channel();
|
|
inner.pending_similar.lock().unwrap().push(tx);
|
|
inner.send_server(server::similar_users())?;
|
|
match tokio::time::timeout(std::time::Duration::from_secs(20), rx).await {
|
|
Ok(Ok(v)) => Ok(v),
|
|
_ => Err(ClientError::Timeout),
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn listen_port(&self) -> u16 {
|
|
self.inner.listen_port
|
|
}
|
|
|
|
pub fn shutdown(&self) {
|
|
let _ = self.shutdown_tx.send(());
|
|
let _ = self.listener_shutdown_tx.send(());
|
|
}
|
|
}
|