chore: refactor server as an actor
This commit is contained in:
Michel de Graaf
+149
-54
@@ -1,9 +1,18 @@
|
||||
name: Release
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
release_type:
|
||||
description: 'Release type (patch, minor, major)'
|
||||
required: false
|
||||
default: 'auto'
|
||||
type: choice
|
||||
options:
|
||||
- auto
|
||||
- patch
|
||||
- minor
|
||||
- major
|
||||
|
||||
permissions:
|
||||
contents: write
|
||||
@@ -42,30 +51,59 @@ jobs:
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
run: |
|
||||
cat > .releaserc.json << 'EOF'
|
||||
{
|
||||
"branches": ["master"],
|
||||
"plugins": [
|
||||
"@semantic-release/commit-analyzer",
|
||||
"@semantic-release/release-notes-generator"
|
||||
],
|
||||
"dryRun": true
|
||||
}
|
||||
# If a specific release type is requested (not 'auto'), always proceed
|
||||
if [ "${{ github.event.inputs.release_type }}" != "auto" ] && [ -n "${{ github.event.inputs.release_type }}" ]; then
|
||||
echo "Manual release type specified: ${{ github.event.inputs.release_type }}"
|
||||
echo "should_release=true" >> $GITHUB_OUTPUT
|
||||
|
||||
# Get current version from Cargo.toml
|
||||
CURRENT_VERSION=$(grep '^version = ' Cargo.toml | head -1 | cut -d'"' -f2)
|
||||
echo "Current version: $CURRENT_VERSION"
|
||||
|
||||
# Calculate next version based on release type
|
||||
IFS='.' read -r major minor patch <<< "$CURRENT_VERSION"
|
||||
case "${{ github.event.inputs.release_type }}" in
|
||||
major)
|
||||
VERSION="$((major + 1)).0.0"
|
||||
;;
|
||||
minor)
|
||||
VERSION="${major}.$((minor + 1)).0"
|
||||
;;
|
||||
patch)
|
||||
VERSION="${major}.${minor}.$((patch + 1))"
|
||||
;;
|
||||
esac
|
||||
echo "version=$VERSION" >> $GITHUB_OUTPUT
|
||||
echo "Manual version to release: $VERSION"
|
||||
else
|
||||
# Auto mode - use semantic-release to determine if release is needed
|
||||
cat > .releaserc.json << 'EOF'
|
||||
{
|
||||
"branches": ["master"],
|
||||
"plugins": [
|
||||
"@semantic-release/commit-analyzer",
|
||||
"@semantic-release/release-notes-generator"
|
||||
],
|
||||
"dryRun": true
|
||||
}
|
||||
EOF
|
||||
|
||||
OUTPUT=$(npx semantic-release --dry-run 2>&1 || true)
|
||||
echo "$OUTPUT"
|
||||
OUTPUT=$(npx semantic-release --dry-run 2>&1 || true)
|
||||
echo "$OUTPUT"
|
||||
|
||||
if echo "$OUTPUT" | grep -q "There is no relevant changes"; then
|
||||
echo "should_release=false" >> $GITHUB_OUTPUT
|
||||
else
|
||||
VERSION=$(echo "$OUTPUT" | grep -oP "next release version is \K[0-9]+\.[0-9]+\.[0-9]+" | head -1)
|
||||
if [ -n "$VERSION" ]; then
|
||||
echo "should_release=true" >> $GITHUB_OUTPUT
|
||||
echo "version=$VERSION" >> $GITHUB_OUTPUT
|
||||
echo "Version to release: $VERSION"
|
||||
else
|
||||
if echo "$OUTPUT" | grep -q "There is no relevant changes"; then
|
||||
echo "should_release=false" >> $GITHUB_OUTPUT
|
||||
echo "No relevant changes found for automatic release"
|
||||
else
|
||||
VERSION=$(echo "$OUTPUT" | grep -oP "next release version is \K[0-9]+\.[0-9]+\.[0-9]+" | head -1)
|
||||
if [ -n "$VERSION" ]; then
|
||||
echo "should_release=true" >> $GITHUB_OUTPUT
|
||||
echo "version=$VERSION" >> $GITHUB_OUTPUT
|
||||
echo "Auto version to release: $VERSION"
|
||||
else
|
||||
echo "should_release=false" >> $GITHUB_OUTPUT
|
||||
echo "Could not determine version for automatic release"
|
||||
fi
|
||||
fi
|
||||
fi
|
||||
|
||||
@@ -191,40 +229,97 @@ jobs:
|
||||
|
||||
- name: Create semantic-release config
|
||||
run: |
|
||||
cat > .releaserc.json << 'EOF'
|
||||
{
|
||||
"branches": ["master"],
|
||||
"plugins": [
|
||||
"@semantic-release/commit-analyzer",
|
||||
"@semantic-release/release-notes-generator",
|
||||
"@semantic-release/changelog",
|
||||
[
|
||||
"@semantic-release/exec",
|
||||
{
|
||||
"prepareCmd": "sed -i 's/^version = \".*\"/version = \"${nextRelease.version}\"/' Cargo.toml"
|
||||
}
|
||||
],
|
||||
[
|
||||
"@semantic-release/git",
|
||||
{
|
||||
"assets": ["CHANGELOG.md", "Cargo.toml"],
|
||||
"message": "chore(release): ${nextRelease.version} [skip ci]\n\n${nextRelease.notes}"
|
||||
}
|
||||
],
|
||||
[
|
||||
"@semantic-release/github",
|
||||
{
|
||||
"assets": [
|
||||
{"path": "release-assets/*.tar.gz"},
|
||||
{"path": "release-assets/*.zip"}
|
||||
]
|
||||
}
|
||||
# Check if we need to force a specific version for manual releases
|
||||
if [ "${{ github.event.inputs.release_type }}" != "auto" ] && [ -n "${{ github.event.inputs.release_type }}" ]; then
|
||||
# For manual releases, we'll bypass semantic-release version detection
|
||||
cat > .releaserc.json << 'EOF'
|
||||
{
|
||||
"branches": ["master"],
|
||||
"plugins": [
|
||||
[
|
||||
"@semantic-release/commit-analyzer",
|
||||
{
|
||||
"releaseRules": [
|
||||
{"breaking": true, "release": "${{ github.event.inputs.release_type }}"},
|
||||
{"revert": true, "release": "patch"},
|
||||
{"type": "feat", "release": "${{ github.event.inputs.release_type }}"},
|
||||
{"type": "fix", "release": "${{ github.event.inputs.release_type }}"},
|
||||
{"type": "perf", "release": "${{ github.event.inputs.release_type }}"},
|
||||
{"scope": "no-release", "release": false}
|
||||
]
|
||||
}
|
||||
],
|
||||
"@semantic-release/release-notes-generator",
|
||||
"@semantic-release/changelog",
|
||||
[
|
||||
"@semantic-release/exec",
|
||||
{
|
||||
"prepareCmd": "sed -i 's/^version = \".*\"/version = \"${{ needs.check-release.outputs.version }}\"/' Cargo.toml"
|
||||
}
|
||||
],
|
||||
[
|
||||
"@semantic-release/git",
|
||||
{
|
||||
"assets": ["CHANGELOG.md", "Cargo.toml"],
|
||||
"message": "chore(release): ${{ needs.check-release.outputs.version }} [skip ci]\n\n${nextRelease.notes}"
|
||||
}
|
||||
],
|
||||
[
|
||||
"@semantic-release/github",
|
||||
{
|
||||
"assets": [
|
||||
{"path": "release-assets/*.tar.gz"},
|
||||
{"path": "release-assets/*.zip"}
|
||||
]
|
||||
}
|
||||
]
|
||||
]
|
||||
]
|
||||
}
|
||||
}
|
||||
EOF
|
||||
else
|
||||
# For automatic releases, use standard semantic-release
|
||||
cat > .releaserc.json << 'EOF'
|
||||
{
|
||||
"branches": ["master"],
|
||||
"plugins": [
|
||||
"@semantic-release/commit-analyzer",
|
||||
"@semantic-release/release-notes-generator",
|
||||
"@semantic-release/changelog",
|
||||
[
|
||||
"@semantic-release/exec",
|
||||
{
|
||||
"prepareCmd": "sed -i 's/^version = \".*\"/version = \"${nextRelease.version}\"/' Cargo.toml"
|
||||
}
|
||||
],
|
||||
[
|
||||
"@semantic-release/git",
|
||||
{
|
||||
"assets": ["CHANGELOG.md", "Cargo.toml"],
|
||||
"message": "chore(release): ${nextRelease.version} [skip ci]\n\n${nextRelease.notes}"
|
||||
}
|
||||
],
|
||||
[
|
||||
"@semantic-release/github",
|
||||
{
|
||||
"assets": [
|
||||
{"path": "release-assets/*.tar.gz"},
|
||||
{"path": "release-assets/*.zip"}
|
||||
]
|
||||
}
|
||||
]
|
||||
]
|
||||
}
|
||||
EOF
|
||||
fi
|
||||
|
||||
- name: Run semantic-release
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
run: npx semantic-release
|
||||
run: |
|
||||
if [ "${{ github.event.inputs.release_type }}" != "auto" ] && [ -n "${{ github.event.inputs.release_type }}" ]; then
|
||||
# For manual releases, we need to create the release directly
|
||||
echo "Creating manual release version ${{ needs.check-release.outputs.version }}"
|
||||
npx semantic-release --no-ci
|
||||
else
|
||||
npx semantic-release
|
||||
fi
|
||||
|
||||
Generated
+1
-1
@@ -4,4 +4,4 @@ version = 4
|
||||
|
||||
[[package]]
|
||||
name = "soulseek-rs"
|
||||
version = "1.0.0"
|
||||
version = "0.1.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Soulseek-rs 🦀
|
||||
|
||||
# ⚠️WARNING THIS IS UNDER DEVELOPMENT AND NOT READY FOR USE ⚠️
|
||||
## ⚠️WARNING THIS IS UNDER DEVELOPMENT AND NOT READY FOR USE ⚠️
|
||||
|
||||
Souleek-rs is an experimental Soulseek client built in Rust. It's under
|
||||
development and not yet ready for use. Soulseek is a closed-source P2P
|
||||
|
||||
+7
-5
@@ -1,6 +1,3 @@
|
||||
// src/actor/mod.rs
|
||||
// Lightweight actor system using the existing thread pool
|
||||
|
||||
use std::sync::mpsc::{channel, Receiver, Sender};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
@@ -10,7 +7,14 @@ use crate::utils::thread_pool::ThreadPool;
|
||||
|
||||
pub mod peer_actor;
|
||||
pub mod peer_registry;
|
||||
pub mod server_actor;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum ConnectionState {
|
||||
Disconnected,
|
||||
Connecting { since: Instant },
|
||||
Connected,
|
||||
}
|
||||
/// Core actor trait - each actor processes messages
|
||||
pub trait Actor: Send + 'static {
|
||||
type Message: Send + Clone + 'static;
|
||||
@@ -28,14 +32,12 @@ pub trait Actor: Send + 'static {
|
||||
fn tick(&mut self) {}
|
||||
}
|
||||
|
||||
/// Handle to send messages to an actor
|
||||
#[derive(Clone)]
|
||||
pub struct ActorHandle<M: Send> {
|
||||
pub(crate) sender: Sender<ActorMessage<M>>,
|
||||
}
|
||||
|
||||
impl<M: Send> ActorHandle<M> {
|
||||
/// Send a message to the actor (non-blocking)
|
||||
pub fn send(&self, msg: M) -> Result<(), String> {
|
||||
self.sender
|
||||
.send(ActorMessage::UserMessage(msg))
|
||||
|
||||
+2
-13
@@ -1,4 +1,4 @@
|
||||
use crate::actor::{Actor, ActorHandle};
|
||||
use crate::actor::{Actor, ActorHandle, ConnectionState};
|
||||
use crate::client::ClientOperation;
|
||||
use crate::dispatcher::MessageDispatcher;
|
||||
use crate::message::peer::{
|
||||
@@ -17,13 +17,6 @@ use std::sync::mpsc::{Receiver, Sender};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
enum ConnectionState {
|
||||
Disconnected,
|
||||
Connecting { since: Instant },
|
||||
Connected,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum PeerMessage {
|
||||
SendMessage(Message),
|
||||
@@ -552,18 +545,14 @@ impl Actor for PeerActor {
|
||||
fn tick(&mut self) {
|
||||
match self.connection_state {
|
||||
ConnectionState::Connecting { .. } => {
|
||||
// Check if connection has been established
|
||||
self.check_connection_status();
|
||||
}
|
||||
ConnectionState::Connected => {
|
||||
// Normal operation - process reads
|
||||
if self.stream.is_some() {
|
||||
self.process_read();
|
||||
}
|
||||
}
|
||||
ConnectionState::Disconnected => {
|
||||
// Do nothing
|
||||
}
|
||||
ConnectionState::Disconnected => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,758 @@
|
||||
use crate::actor::{Actor, ActorHandle, ConnectionState};
|
||||
use crate::client::ClientOperation;
|
||||
use crate::dispatcher::MessageDispatcher;
|
||||
use crate::message::server::ConnectToPeerHandler;
|
||||
use crate::message::server::ExcludedSearchPhrasesHandler;
|
||||
use crate::message::server::FileSearchHandler;
|
||||
use crate::message::server::GetPeerAddressHandler;
|
||||
use crate::message::server::LoginHandler;
|
||||
use crate::message::server::MessageFactory;
|
||||
use crate::message::server::MessageUser;
|
||||
use crate::message::server::ParentMinSpeedHandler;
|
||||
use crate::message::server::ParentSpeedRatioHandler;
|
||||
use crate::message::server::PrivilegedUsersHandler;
|
||||
use crate::message::server::RoomListHandler;
|
||||
use crate::message::server::WishListIntervalHandler;
|
||||
use crate::message::{Handlers, MessageType};
|
||||
use crate::message::{Message, MessageReader};
|
||||
use crate::peer::ConnectionType;
|
||||
use crate::peer::Peer;
|
||||
|
||||
use std::io::{self, Error, Write};
|
||||
use std::net::{TcpStream, ToSocketAddrs};
|
||||
use std::sync::mpsc::{Receiver, Sender};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::{debug, error, info, trace, warn, SoulseekRs};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct PeerAddress {
|
||||
host: String,
|
||||
port: u16,
|
||||
}
|
||||
|
||||
impl PeerAddress {
|
||||
pub fn new(host: String, port: u16) -> Self {
|
||||
Self { host, port }
|
||||
}
|
||||
|
||||
pub fn get_host(&self) -> &str {
|
||||
&self.host
|
||||
}
|
||||
|
||||
pub fn get_port(&self) -> u16 {
|
||||
self.port
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for PeerAddress {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
|
||||
write!(f, "{}:{}", self.host, self.port)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Context {
|
||||
pub logged_in: Option<bool>,
|
||||
#[allow(dead_code)]
|
||||
rooms: Rooms,
|
||||
}
|
||||
|
||||
impl Default for Context {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Context {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
#[allow(dead_code)]
|
||||
rooms: Rooms::new(),
|
||||
logged_in: Option::None,
|
||||
}
|
||||
}
|
||||
}
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct UserMessage {
|
||||
id: u32,
|
||||
timestamp: u32,
|
||||
username: String,
|
||||
message: String,
|
||||
new_message: bool,
|
||||
}
|
||||
impl UserMessage {
|
||||
pub fn new(
|
||||
id: u32,
|
||||
timestamp: u32,
|
||||
username: String,
|
||||
message: String,
|
||||
new_message: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
id,
|
||||
timestamp,
|
||||
username,
|
||||
message,
|
||||
new_message,
|
||||
}
|
||||
}
|
||||
pub fn print(&self) {
|
||||
debug!(
|
||||
"Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}",
|
||||
self.timestamp,
|
||||
self.username,
|
||||
self.id,
|
||||
self.new_message,
|
||||
self.message
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug)]
|
||||
pub struct Rooms {
|
||||
pub public_rooms: Vec<Room>,
|
||||
pub owned_private_rooms: Vec<Room>,
|
||||
pub private_rooms: Vec<Room>,
|
||||
pub operated_private_rooms: Vec<Room>,
|
||||
}
|
||||
impl Rooms {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
public_rooms: Vec::new(),
|
||||
owned_private_rooms: Vec::new(),
|
||||
private_rooms: Vec::new(),
|
||||
operated_private_rooms: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn print(&self) {
|
||||
info!("Public rooms ({}):", self.public_rooms.len());
|
||||
for room in &self.public_rooms {
|
||||
room.print();
|
||||
}
|
||||
info!("Owned private rooms ({}):", self.owned_private_rooms.len());
|
||||
for room in &self.owned_private_rooms {
|
||||
room.print();
|
||||
}
|
||||
info!("Private rooms ({}):", self.private_rooms.len());
|
||||
for room in &self.private_rooms {
|
||||
room.print();
|
||||
}
|
||||
info!(
|
||||
"Operated private rooms ({}):",
|
||||
self.operated_private_rooms.len()
|
||||
);
|
||||
for room in &self.operated_private_rooms {
|
||||
room.print();
|
||||
}
|
||||
}
|
||||
}
|
||||
#[derive(Debug)]
|
||||
pub struct Room {
|
||||
name: String,
|
||||
number_of_users: i32,
|
||||
}
|
||||
impl Room {
|
||||
#[allow(dead_code)]
|
||||
pub fn new(name: String, number_of_users: i32) -> Self {
|
||||
Self {
|
||||
name,
|
||||
number_of_users,
|
||||
}
|
||||
}
|
||||
#[allow(dead_code)]
|
||||
pub fn set_number_of_users(&mut self, number_of_users: i32) {
|
||||
self.number_of_users = number_of_users;
|
||||
}
|
||||
pub fn print(&self) {
|
||||
debug!(
|
||||
"Room: {}, Number of users: {}",
|
||||
self.name, self.number_of_users
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum ServerOperation {
|
||||
ProcessRead,
|
||||
LoginStatus(bool),
|
||||
SendMessage(Message),
|
||||
Login {
|
||||
username: String,
|
||||
password: String,
|
||||
response: std::sync::mpsc::Sender<Result<bool, SoulseekRs>>,
|
||||
},
|
||||
FileSearch {
|
||||
token: u32,
|
||||
query: String,
|
||||
},
|
||||
#[allow(dead_code)]
|
||||
ConnectToPeer(Peer),
|
||||
PierceFirewall(u32),
|
||||
GetPeerAddress(String),
|
||||
GetPeerAddressResponse {
|
||||
username: String,
|
||||
host: String,
|
||||
port: u32,
|
||||
obfuscation_type: u32,
|
||||
obfuscated_port: u16,
|
||||
},
|
||||
}
|
||||
|
||||
pub struct ServerActor {
|
||||
address: PeerAddress,
|
||||
context: Arc<RwLock<Context>>,
|
||||
listen_port: u32,
|
||||
enable_listen: bool,
|
||||
stream: Option<TcpStream>,
|
||||
connection_state: ConnectionState,
|
||||
reader: MessageReader,
|
||||
client_channel: Sender<ClientOperation>,
|
||||
self_handle: Option<ActorHandle<ServerOperation>>,
|
||||
dispatcher: Option<MessageDispatcher<ServerOperation>>,
|
||||
dispatcher_receiver: Option<Receiver<ServerOperation>>,
|
||||
dispatcher_sender: Option<Sender<ServerOperation>>,
|
||||
queued_messages: Vec<ServerOperation>,
|
||||
}
|
||||
|
||||
impl ServerActor {
|
||||
pub fn new(
|
||||
address: PeerAddress,
|
||||
client_channel: Sender<ClientOperation>,
|
||||
listen_port: u32,
|
||||
enable_listen: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
address,
|
||||
context: Arc::new(RwLock::new(Context::new())),
|
||||
listen_port,
|
||||
enable_listen,
|
||||
stream: None,
|
||||
connection_state: ConnectionState::Disconnected,
|
||||
dispatcher: None,
|
||||
dispatcher_receiver: None,
|
||||
dispatcher_sender: None,
|
||||
reader: MessageReader::new(),
|
||||
client_channel,
|
||||
self_handle: None,
|
||||
queued_messages: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn get_address(&self) -> &PeerAddress {
|
||||
&self.address
|
||||
}
|
||||
|
||||
pub fn get_sender(&self) -> Option<&Sender<ServerOperation>> {
|
||||
self.dispatcher_sender.as_ref()
|
||||
}
|
||||
|
||||
fn initiate_connection(&mut self) -> bool {
|
||||
let host = self.address.host.clone();
|
||||
let port = self.address.port;
|
||||
|
||||
let addr_str = format!("{}:{}", host, port);
|
||||
|
||||
let mut socket_addrs = match addr_str.to_socket_addrs() {
|
||||
Ok(addrs) => addrs,
|
||||
Err(e) => {
|
||||
error!("[server_actor] Failed to resolve address: {}", e);
|
||||
|
||||
self.disconnect_with_error(io::Error::new(
|
||||
io::ErrorKind::InvalidInput,
|
||||
e,
|
||||
));
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
let socket_addr = socket_addrs.next();
|
||||
|
||||
match socket_addr {
|
||||
Some(addr) => {
|
||||
if let Ok(stream) = TcpStream::connect(addr) {
|
||||
if let Err(e) = stream.set_nonblocking(true) {
|
||||
error!(
|
||||
"[server_actor] Failed to set non-blocking: {}",
|
||||
e
|
||||
);
|
||||
self.disconnect_with_error(e);
|
||||
return false;
|
||||
}
|
||||
|
||||
stream.set_nodelay(true).ok();
|
||||
|
||||
self.stream = Some(stream);
|
||||
self.connection_state = ConnectionState::Connecting {
|
||||
since: Instant::now(),
|
||||
};
|
||||
true
|
||||
} else {
|
||||
match TcpStream::connect(addr) {
|
||||
Ok(stream) => {
|
||||
if let Err(e) = stream.set_nonblocking(true) {
|
||||
error!("[server_actor] Failed to set non-blocking: {}", e);
|
||||
self.disconnect_with_error(e);
|
||||
return false;
|
||||
}
|
||||
stream.set_nodelay(true).ok();
|
||||
self.stream = Some(stream);
|
||||
self.connection_state =
|
||||
ConnectionState::Connecting {
|
||||
since: Instant::now(),
|
||||
};
|
||||
true
|
||||
}
|
||||
Err(e) => {
|
||||
self.disconnect_with_error(e);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
let error_msg =
|
||||
format!("No socket addresses found for {}:{}", host, port);
|
||||
error!("[server_actor] {}", error_msg);
|
||||
self.disconnect_with_error(io::Error::new(
|
||||
io::ErrorKind::InvalidInput,
|
||||
error_msg,
|
||||
));
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn set_self_handle(&mut self, handle: ActorHandle<ServerOperation>) {
|
||||
self.self_handle = Some(handle);
|
||||
}
|
||||
|
||||
fn initialize_dispatcher(&mut self) {
|
||||
let (dispatcher_sender, dispatcher_receiver) =
|
||||
std::sync::mpsc::channel::<ServerOperation>();
|
||||
|
||||
self.dispatcher_receiver = Some(dispatcher_receiver);
|
||||
self.dispatcher_sender = Some(dispatcher_sender.clone());
|
||||
|
||||
self.client_channel
|
||||
.send(ClientOperation::SetServerSender(dispatcher_sender.clone()))
|
||||
.unwrap();
|
||||
|
||||
let mut handlers = Handlers::new();
|
||||
|
||||
handlers.register_handler(LoginHandler);
|
||||
handlers.register_handler(RoomListHandler);
|
||||
handlers.register_handler(ExcludedSearchPhrasesHandler);
|
||||
handlers.register_handler(PrivilegedUsersHandler);
|
||||
handlers.register_handler(MessageUser);
|
||||
handlers.register_handler(WishListIntervalHandler);
|
||||
handlers.register_handler(ParentMinSpeedHandler);
|
||||
handlers.register_handler(ParentSpeedRatioHandler);
|
||||
handlers.register_handler(PrivilegedUsersHandler);
|
||||
handlers.register_handler(FileSearchHandler);
|
||||
handlers.register_handler(GetPeerAddressHandler);
|
||||
handlers.register_handler(ConnectToPeerHandler);
|
||||
|
||||
self.dispatcher = Some(MessageDispatcher::new(
|
||||
"server_actor".into(),
|
||||
dispatcher_sender,
|
||||
handlers,
|
||||
));
|
||||
}
|
||||
|
||||
fn process_dispatcher_messages(&mut self) {
|
||||
let messages: Vec<ServerOperation> = self
|
||||
.dispatcher_receiver
|
||||
.as_ref()
|
||||
.map_or_else(Vec::new, |receiver| {
|
||||
let mut msgs = Vec::new();
|
||||
while let Ok(msg) = receiver.try_recv() {
|
||||
msgs.push(msg);
|
||||
}
|
||||
msgs
|
||||
});
|
||||
|
||||
messages
|
||||
.iter()
|
||||
.for_each(|msg| self.handle_message(msg.clone()));
|
||||
}
|
||||
|
||||
pub fn login(
|
||||
&mut self,
|
||||
username: &str,
|
||||
password: &str,
|
||||
) -> Result<bool, SoulseekRs> {
|
||||
self.queue_message(MessageFactory::build_login_message(
|
||||
username, password,
|
||||
));
|
||||
let context = self.context.clone();
|
||||
let mut logged_in;
|
||||
let timeout = Duration::from_secs(5);
|
||||
let start = Instant::now();
|
||||
|
||||
loop {
|
||||
if start.elapsed() > timeout {
|
||||
warn!("Timeout waiting for login response");
|
||||
return Err(SoulseekRs::Timeout);
|
||||
}
|
||||
|
||||
{
|
||||
logged_in = context.read().unwrap().logged_in
|
||||
}
|
||||
|
||||
if logged_in.is_some() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if logged_in.unwrap() {
|
||||
info!("Logged in as {}", username);
|
||||
self.queue_message(MessageFactory::build_shared_folders_message(
|
||||
1, 499,
|
||||
));
|
||||
self.queue_message(MessageFactory::build_no_parent_message());
|
||||
self.queue_message(MessageFactory::build_set_status_message(2));
|
||||
if self.enable_listen {
|
||||
self.queue_message(
|
||||
MessageFactory::build_set_wait_port_message(
|
||||
self.listen_port,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(logged_in.unwrap())
|
||||
}
|
||||
|
||||
pub fn file_search(&mut self, token: u32, query: &str) {
|
||||
self.queue_message(MessageFactory::build_file_search_message(
|
||||
token, query,
|
||||
));
|
||||
}
|
||||
|
||||
fn handle_message(&mut self, msg: ServerOperation) {
|
||||
if !matches!(self.connection_state, ConnectionState::Connected) {
|
||||
match &msg {
|
||||
ServerOperation::ProcessRead => {
|
||||
// Always process read operations
|
||||
}
|
||||
_ => {
|
||||
// Queue all other messages when not connected
|
||||
self.queued_messages.push(msg);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
match msg {
|
||||
ServerOperation::ConnectToPeer(peer) => {
|
||||
if let Some(op) = match peer.connection_type {
|
||||
ConnectionType::P | ConnectionType::F => {
|
||||
Some(ClientOperation::ConnectToPeer(peer.clone()))
|
||||
}
|
||||
ConnectionType::D => None,
|
||||
} {
|
||||
self.client_channel.send(op).unwrap();
|
||||
}
|
||||
}
|
||||
ServerOperation::LoginStatus(message) => {
|
||||
self.context.write().unwrap().logged_in = Some(message);
|
||||
}
|
||||
ServerOperation::PierceFirewall(token) => {
|
||||
self.send_message(
|
||||
MessageFactory::build_pierce_firewall_message(token),
|
||||
);
|
||||
}
|
||||
ServerOperation::SendMessage(message) => {
|
||||
self.send_message(message);
|
||||
}
|
||||
ServerOperation::GetPeerAddress(username) => {
|
||||
self.send_message(MessageFactory::build_get_peer_address(
|
||||
&username,
|
||||
));
|
||||
}
|
||||
ServerOperation::GetPeerAddressResponse {
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port,
|
||||
} => {
|
||||
debug!(
|
||||
"[server_actor] Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})"
|
||||
, username, host, port, obfuscation_type, obfuscated_port
|
||||
);
|
||||
|
||||
if let Err(e) = self.client_channel.send(
|
||||
ClientOperation::GetPeerAddressResponse {
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port,
|
||||
},
|
||||
) {
|
||||
error!("[server_actor] Error forwarding GetPeerAddress response to client: {}", e);
|
||||
}
|
||||
}
|
||||
ServerOperation::ProcessRead => {
|
||||
self.process_read();
|
||||
}
|
||||
ServerOperation::Login {
|
||||
username,
|
||||
password,
|
||||
response,
|
||||
} => {
|
||||
self.queue_message(MessageFactory::build_login_message(
|
||||
&username, &password,
|
||||
));
|
||||
|
||||
let start = std::time::Instant::now();
|
||||
let timeout = Duration::from_secs(5);
|
||||
|
||||
let context = self.context.clone();
|
||||
std::thread::spawn(move || loop {
|
||||
if start.elapsed() >= timeout {
|
||||
let _ = response.send(Err(SoulseekRs::Timeout));
|
||||
break;
|
||||
}
|
||||
|
||||
if let Some(logged_in) = context.read().unwrap().logged_in {
|
||||
let result = if logged_in {
|
||||
Ok(true)
|
||||
} else {
|
||||
Err(SoulseekRs::AuthenticationFailed)
|
||||
};
|
||||
let _ = response.send(result);
|
||||
break;
|
||||
}
|
||||
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
});
|
||||
}
|
||||
ServerOperation::FileSearch { token, query } => {
|
||||
self.file_search(token, &query);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn process_read(&mut self) {
|
||||
if self.reader.buffer_len() > 0 {
|
||||
self.extract_and_process_messages();
|
||||
}
|
||||
|
||||
{
|
||||
let stream = match self.stream.as_mut() {
|
||||
Some(s) => s,
|
||||
None => return,
|
||||
};
|
||||
|
||||
match self.reader.read_from_socket(stream) {
|
||||
Ok(()) => {}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
|
||||
debug!("[server_actor] Read operation timed out",);
|
||||
}
|
||||
Err(e) => {
|
||||
error!(
|
||||
"[server_actor] Error reading from server: {} (kind: {:?}). Disconnecting.",
|
||||
e, e.kind()
|
||||
);
|
||||
self.disconnect_with_error(e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
self.extract_and_process_messages();
|
||||
}
|
||||
|
||||
fn extract_and_process_messages(&mut self) {
|
||||
let mut extracted_count = 0;
|
||||
loop {
|
||||
match self.reader.extract_message() {
|
||||
Ok(Some(mut message)) => {
|
||||
extracted_count += 1;
|
||||
trace!(
|
||||
"[server_actor] ← Message #{}: {:?}",
|
||||
extracted_count,
|
||||
message
|
||||
.get_message_name(
|
||||
MessageType::Server,
|
||||
message.get_message_code() as u32
|
||||
)
|
||||
.map_err(|e| e.to_string())
|
||||
);
|
||||
if let Some(ref dispatcher) = self.dispatcher {
|
||||
dispatcher.dispatch(&mut message);
|
||||
} else {
|
||||
warn!("[server_actor] No dispatcher available!",);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!( "[server_actor] Error extracting message: {}. Disconnecting.", e);
|
||||
self.disconnect_with_error(e);
|
||||
return;
|
||||
}
|
||||
Ok(None) => {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
self.process_dispatcher_messages();
|
||||
}
|
||||
|
||||
fn queue_message(&mut self, message: Message) {
|
||||
if let Some(sender) = &self.dispatcher_sender {
|
||||
match sender.send(ServerOperation::SendMessage(message)) {
|
||||
Ok(_) => {}
|
||||
Err(e) => error!("Failed to send: {}", e),
|
||||
}
|
||||
} else {
|
||||
self.queued_messages
|
||||
.push(ServerOperation::SendMessage(message));
|
||||
}
|
||||
}
|
||||
|
||||
fn send_message(&mut self, message: Message) {
|
||||
let stream = match self.stream.as_mut() {
|
||||
Some(s) => s,
|
||||
None => {
|
||||
error!("[server_actor] Cannot send message: stream is None");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
trace!(
|
||||
"[server_actor] ➡ {:?}",
|
||||
message
|
||||
.get_message_name(
|
||||
MessageType::Server,
|
||||
u32::from_le_bytes(
|
||||
message.get_slice(0, 4).try_into().unwrap()
|
||||
)
|
||||
)
|
||||
.map_err(|e| e.to_string())
|
||||
);
|
||||
|
||||
if let Err(e) = stream.write_all(&message.get_buffer()) {
|
||||
error!(
|
||||
"[server_actor] Error writing message: {}. Disconnecting.",
|
||||
e
|
||||
);
|
||||
self.disconnect_with_error(e);
|
||||
return;
|
||||
}
|
||||
|
||||
if let Err(e) = stream.flush() {
|
||||
error!(
|
||||
"[server_actor] Error flushing stream: {}. Disconnecting.",
|
||||
e
|
||||
);
|
||||
self.disconnect_with_error(e);
|
||||
}
|
||||
}
|
||||
|
||||
fn disconnect_with_error(&mut self, _error: Error) {
|
||||
debug!("[server_actor] disconnect");
|
||||
|
||||
self.stream.take();
|
||||
}
|
||||
|
||||
fn disconnect(&mut self) {
|
||||
debug!("[server_actor] disconnected");
|
||||
|
||||
self.stream.take();
|
||||
}
|
||||
|
||||
fn check_connection_status(&mut self) {
|
||||
let ConnectionState::Connecting { since } = self.connection_state
|
||||
else {
|
||||
return;
|
||||
};
|
||||
|
||||
if since.elapsed() > Duration::from_secs(20) {
|
||||
error!("[server_actor] Connection timeout after 20 seconds");
|
||||
self.disconnect_with_error(io::Error::new(
|
||||
io::ErrorKind::TimedOut,
|
||||
"Connection timeout",
|
||||
));
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(ref stream) = self.stream else {
|
||||
return;
|
||||
};
|
||||
|
||||
match stream.peer_addr() {
|
||||
Ok(_) => {
|
||||
self.connection_state = ConnectionState::Connected;
|
||||
self.on_connection_established();
|
||||
}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::NotConnected => {}
|
||||
Err(e) => {
|
||||
error!("[server_actor] Connection failed: {}", e);
|
||||
self.disconnect_with_error(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn on_connection_established(&mut self) {
|
||||
let Some(_) = self.stream else {
|
||||
panic!("Stream should be available here")
|
||||
};
|
||||
|
||||
self.initialize_dispatcher();
|
||||
|
||||
let queued = std::mem::take(&mut self.queued_messages);
|
||||
for msg in queued {
|
||||
self.handle_message(msg);
|
||||
}
|
||||
|
||||
if let Some(ref handle) = self.self_handle {
|
||||
handle.send(ServerOperation::ProcessRead).ok();
|
||||
}
|
||||
|
||||
self.process_read();
|
||||
}
|
||||
}
|
||||
|
||||
impl Actor for ServerActor {
|
||||
type Message = ServerOperation;
|
||||
|
||||
fn handle(&mut self, msg: Self::Message) {
|
||||
self.handle_message(msg);
|
||||
}
|
||||
|
||||
fn on_start(&mut self) {
|
||||
if self.stream.is_none() {
|
||||
self.initiate_connection();
|
||||
} else {
|
||||
self.connection_state = ConnectionState::Connected;
|
||||
self.on_connection_established();
|
||||
}
|
||||
}
|
||||
|
||||
fn on_stop(&mut self) {
|
||||
trace!("[server_actor] actor stopping");
|
||||
self.disconnect();
|
||||
}
|
||||
|
||||
fn tick(&mut self) {
|
||||
match self.connection_state {
|
||||
ConnectionState::Connecting { .. } => {
|
||||
self.check_connection_status();
|
||||
}
|
||||
ConnectionState::Connected => {
|
||||
if self.stream.is_some() {
|
||||
self.process_read();
|
||||
}
|
||||
}
|
||||
ConnectionState::Disconnected => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
+50
-65
@@ -1,9 +1,10 @@
|
||||
use crate::actor::server_actor::{PeerAddress, ServerActor, ServerOperation};
|
||||
use crate::actor::ActorHandle;
|
||||
use crate::types::{DownloadResult, DownloadStatus};
|
||||
use crate::{
|
||||
actor::{peer_registry::PeerRegistry, ActorSystem},
|
||||
error::{Result, SoulseekRs},
|
||||
peer::{listen::Listen, ConnectionType, DownloadPeer, NewPeer, Peer},
|
||||
server::{PeerAddress, Server, ServerOperation},
|
||||
peer::{ConnectionType, DownloadPeer, NewPeer, Peer},
|
||||
types::{Download, FileSearchResult},
|
||||
utils::{md5, thread_pool::ThreadPool},
|
||||
Transfer,
|
||||
@@ -43,11 +44,12 @@ pub enum ClientOperation {
|
||||
obfuscated_port: u16,
|
||||
},
|
||||
UploadFailed(String, String),
|
||||
SetServerSender(Sender<ServerOperation>),
|
||||
}
|
||||
pub struct ClientContext {
|
||||
pub peer_registry: Option<PeerRegistry>,
|
||||
sender: Option<Sender<ClientOperation>>,
|
||||
server_sender: Option<Sender<crate::server::ServerOperation>>,
|
||||
server_sender: Option<Sender<ServerOperation>>,
|
||||
search_results: Vec<FileSearchResult>,
|
||||
pub download_tokens: HashMap<u32, Download>,
|
||||
actor_system: Arc<ActorSystem>,
|
||||
@@ -84,7 +86,7 @@ pub struct Client {
|
||||
address: PeerAddress,
|
||||
username: String,
|
||||
password: String,
|
||||
server: Option<Server>,
|
||||
server_handle: Option<ActorHandle<ServerOperation>>,
|
||||
context: Arc<RwLock<ClientContext>>,
|
||||
}
|
||||
|
||||
@@ -97,15 +99,14 @@ impl Client {
|
||||
listen_port: Option<u32>,
|
||||
) -> Self {
|
||||
crate::utils::logger::init();
|
||||
let context = Arc::new(RwLock::new(ClientContext::new()));
|
||||
Self {
|
||||
enable_listen,
|
||||
listen_port: listen_port.unwrap_or(DEFALT_LISTEN_PORT),
|
||||
address,
|
||||
username,
|
||||
password,
|
||||
server: None,
|
||||
context,
|
||||
context: Arc::new(RwLock::new(ClientContext::new())),
|
||||
server_handle: None,
|
||||
}
|
||||
}
|
||||
pub fn with_defaults(
|
||||
@@ -122,72 +123,50 @@ impl Client {
|
||||
Receiver<ClientOperation>,
|
||||
) = mpsc::channel();
|
||||
|
||||
let listen_port = self.listen_port;
|
||||
let context = self.context.clone();
|
||||
let own_username = self.username.clone();
|
||||
let mut ctx = self.context.write().unwrap();
|
||||
ctx.sender = Some(sender.clone());
|
||||
let peer_registry =
|
||||
PeerRegistry::new(ctx.actor_system.clone(), sender.clone());
|
||||
ctx.peer_registry = Some(peer_registry);
|
||||
|
||||
// Initialize PeerRegistry with the actor system
|
||||
{
|
||||
let mut ctx = self.context.write().unwrap();
|
||||
ctx.sender = Some(sender.clone());
|
||||
let peer_registry =
|
||||
PeerRegistry::new(ctx.actor_system.clone(), sender.clone());
|
||||
ctx.peer_registry = Some(peer_registry);
|
||||
}
|
||||
|
||||
let client_sender = sender.clone();
|
||||
|
||||
self.server = match Server::new(
|
||||
let server_actor = ServerActor::new(
|
||||
self.address.clone(),
|
||||
sender,
|
||||
self.listen_port,
|
||||
self.enable_listen,
|
||||
) {
|
||||
Ok(server) => {
|
||||
info!(
|
||||
"Connected to server at {}:{}",
|
||||
server.get_address().get_host(),
|
||||
server.get_address().get_port()
|
||||
);
|
||||
);
|
||||
|
||||
if self.enable_listen {
|
||||
thread::spawn(move || {
|
||||
Listen::start(
|
||||
listen_port,
|
||||
client_sender.clone(),
|
||||
context.clone(),
|
||||
own_username,
|
||||
);
|
||||
});
|
||||
}
|
||||
let mut unlocked_context = self.context.write().unwrap();
|
||||
unlocked_context.server_sender =
|
||||
Some(server.get_sender().clone());
|
||||
self.server_handle = Some(ctx.actor_system.spawn_with_handle(
|
||||
server_actor,
|
||||
|actor, handle| {
|
||||
actor.set_self_handle(handle);
|
||||
},
|
||||
));
|
||||
|
||||
Self::listen_to_client_operations(
|
||||
message_reader,
|
||||
self.context.clone(),
|
||||
self.username.clone(),
|
||||
);
|
||||
Some(server)
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Error connecting to server: {}", e);
|
||||
None
|
||||
}
|
||||
};
|
||||
Self::listen_to_client_operations(
|
||||
message_reader,
|
||||
self.context.clone(),
|
||||
self.username.clone(),
|
||||
);
|
||||
}
|
||||
|
||||
pub fn login(&self) -> Result<bool> {
|
||||
info!("Logging in as {}", self.username);
|
||||
if let Some(server) = &self.server {
|
||||
return if (server.login(&self.username, &self.password))? {
|
||||
Ok(true)
|
||||
} else {
|
||||
Err(SoulseekRs::AuthenticationFailed)
|
||||
};
|
||||
if let Some(handle) = &self.server_handle {
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let _ = handle.send(ServerOperation::Login {
|
||||
username: self.username.clone(),
|
||||
password: self.password.clone(),
|
||||
response: tx,
|
||||
});
|
||||
|
||||
match rx.recv() {
|
||||
Ok(result) => result,
|
||||
Err(_) => Err(SoulseekRs::Timeout),
|
||||
}
|
||||
} else {
|
||||
Err(SoulseekRs::NotConnected)
|
||||
}
|
||||
Err(SoulseekRs::NotConnected)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
@@ -206,10 +185,13 @@ impl Client {
|
||||
timeout: Duration,
|
||||
) -> Result<Vec<FileSearchResult>> {
|
||||
info!("Searching for {}", query);
|
||||
if let Some(server) = &self.server {
|
||||
if let Some(handle) = &self.server_handle {
|
||||
let hash = md5::md5(query);
|
||||
let token = u32::from_str_radix(&hash[0..5], 16)?;
|
||||
server.file_search(token, query);
|
||||
let _ = handle.send(ServerOperation::FileSearch {
|
||||
token,
|
||||
query: query.to_string(),
|
||||
});
|
||||
} else {
|
||||
return Err(SoulseekRs::NotConnected);
|
||||
}
|
||||
@@ -325,7 +307,6 @@ impl Client {
|
||||
let client_context_clone = client_context.clone();
|
||||
let own_username_clone = own_username.clone();
|
||||
|
||||
// Spawn in background to avoid blocking the receiver thread
|
||||
thread::spawn(move || {
|
||||
Self::connect_to_peer(
|
||||
peer,
|
||||
@@ -536,7 +517,6 @@ impl Client {
|
||||
client_context.clone();
|
||||
let own_username_clone = own_username.clone();
|
||||
|
||||
// Spawn in background to avoid blocking the receiver thread
|
||||
thread::spawn(move || {
|
||||
Self::connect_to_peer(
|
||||
peer,
|
||||
@@ -594,6 +574,11 @@ impl Client {
|
||||
Some(&filename),
|
||||
);
|
||||
}
|
||||
ClientOperation::SetServerSender(sender) => {
|
||||
client_context.write().unwrap().server_sender =
|
||||
Some(sender);
|
||||
debug!("[client] Server sender initialized");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
|
||||
+3
-3
@@ -5,23 +5,23 @@ pub mod dispatcher;
|
||||
pub mod error;
|
||||
pub mod message;
|
||||
pub mod peer;
|
||||
pub mod server;
|
||||
pub mod types;
|
||||
#[macro_use]
|
||||
pub mod utils;
|
||||
|
||||
// Prelude module for commonly used items
|
||||
pub mod prelude {
|
||||
pub use crate::actor::server_actor::PeerAddress;
|
||||
pub use crate::types::{
|
||||
DownloadResult, DownloadStatus, File, FileSearchResult, Transfer,
|
||||
};
|
||||
pub use crate::{debug, error, info, trace, warn};
|
||||
pub use crate::{Client, PeerAddress};
|
||||
}
|
||||
|
||||
// Re-export commonly used types
|
||||
pub use actor::server_actor::PeerAddress;
|
||||
pub use client::Client;
|
||||
pub use error::{Result, SoulseekRs};
|
||||
pub use server::PeerAddress;
|
||||
pub use types::{
|
||||
DownloadResult, DownloadStatus, File, FileSearchResult, Transfer,
|
||||
};
|
||||
|
||||
+1
-1
@@ -46,7 +46,7 @@ fn main() {
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("Failed to login: {e}");
|
||||
eprintln!("Failed to login: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use crate::message::{Message, MessageHandler};
|
||||
use crate::peer::Peer;
|
||||
use crate::server::ServerOperation;
|
||||
use crate::actor::server_actor::ServerOperation;
|
||||
use std::sync::mpsc::Sender;
|
||||
pub struct ConnectToPeerHandler;
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use crate::debug;
|
||||
use crate::{
|
||||
actor::server_actor::ServerOperation,
|
||||
message::{Message, MessageHandler},
|
||||
server::ServerOperation,
|
||||
};
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{
|
||||
message::handlers::MessageHandler, message::Message,
|
||||
server::ServerOperation,
|
||||
actor::server_actor::ServerOperation,
|
||||
};
|
||||
|
||||
pub struct FileSearchHandler;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use crate::message::{Message, MessageHandler};
|
||||
use crate::server::ServerOperation;
|
||||
use crate::actor::server_actor::ServerOperation;
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
pub struct GetPeerAddressHandler;
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
use crate::{debug, info};
|
||||
use crate::{
|
||||
actor::server_actor::ServerOperation, debug, info, message::Message,
|
||||
};
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{message::Message, server::ServerOperation};
|
||||
|
||||
use crate::message::MessageHandler;
|
||||
|
||||
pub struct LoginHandler;
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
use crate::actor::server_actor::{ServerOperation, UserMessage};
|
||||
use crate::info;
|
||||
use crate::{
|
||||
message::{Message, MessageHandler},
|
||||
server::{ServerOperation, UserMessage},
|
||||
};
|
||||
use crate::message::{Message, MessageHandler};
|
||||
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{
|
||||
message::{Message, MessageHandler},
|
||||
server::ServerOperation,
|
||||
actor::server_actor::ServerOperation,
|
||||
};
|
||||
|
||||
pub struct ParentMinSpeedHandler;
|
||||
|
||||
@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{
|
||||
message::{Message, MessageHandler},
|
||||
server::ServerOperation,
|
||||
actor::server_actor::ServerOperation,
|
||||
};
|
||||
|
||||
pub struct ParentSpeedRatioHandler;
|
||||
|
||||
@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{
|
||||
message::{Message, MessageHandler},
|
||||
server::ServerOperation,
|
||||
actor::server_actor::ServerOperation,
|
||||
};
|
||||
|
||||
pub struct PrivilegedUsersHandler;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use crate::{
|
||||
actor::server_actor::ServerOperation,
|
||||
message::{Message, MessageHandler},
|
||||
server::ServerOperation,
|
||||
};
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ use std::sync::mpsc::Sender;
|
||||
|
||||
use crate::{
|
||||
message::{Message, MessageHandler},
|
||||
server::ServerOperation,
|
||||
actor::server_actor::ServerOperation,
|
||||
};
|
||||
|
||||
pub struct WishListIntervalHandler;
|
||||
|
||||
-500
@@ -1,500 +0,0 @@
|
||||
use crate::client::ClientOperation;
|
||||
use crate::dispatcher::MessageDispatcher;
|
||||
use crate::error::{Result, SoulseekRs};
|
||||
use crate::message::server::ConnectToPeerHandler;
|
||||
use crate::message::server::ExcludedSearchPhrasesHandler;
|
||||
use crate::message::server::FileSearchHandler;
|
||||
use crate::message::server::GetPeerAddressHandler;
|
||||
use crate::message::server::LoginHandler;
|
||||
use crate::message::server::MessageFactory;
|
||||
use crate::message::server::MessageUser;
|
||||
use crate::message::server::ParentMinSpeedHandler;
|
||||
use crate::message::server::ParentSpeedRatioHandler;
|
||||
use crate::message::server::PrivilegedUsersHandler;
|
||||
use crate::message::server::RoomListHandler;
|
||||
use crate::message::server::WishListIntervalHandler;
|
||||
use crate::message::Message;
|
||||
use crate::message::MessageReader;
|
||||
use crate::message::{Handlers, MessageType};
|
||||
use crate::peer::ConnectionType;
|
||||
use crate::peer::Peer;
|
||||
|
||||
use std::io::{self, Write};
|
||||
use std::net::{TcpStream, ToSocketAddrs};
|
||||
use std::sync::mpsc;
|
||||
use std::sync::mpsc::{Receiver, Sender};
|
||||
use std::sync::{Arc, Barrier, RwLock};
|
||||
use std::thread::{self};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::{debug, error, info, trace, warn};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct PeerAddress {
|
||||
host: String,
|
||||
port: u16,
|
||||
}
|
||||
|
||||
impl PeerAddress {
|
||||
pub fn new(host: String, port: u16) -> Self {
|
||||
Self { host, port }
|
||||
}
|
||||
|
||||
pub fn get_host(&self) -> &str {
|
||||
&self.host
|
||||
}
|
||||
|
||||
pub fn get_port(&self) -> u16 {
|
||||
self.port
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Context {
|
||||
pub logged_in: Option<bool>,
|
||||
#[allow(dead_code)]
|
||||
rooms: Rooms,
|
||||
}
|
||||
|
||||
impl Default for Context {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Context {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
#[allow(dead_code)]
|
||||
rooms: Rooms::new(),
|
||||
logged_in: Option::None,
|
||||
}
|
||||
}
|
||||
}
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct UserMessage {
|
||||
id: u32,
|
||||
timestamp: u32,
|
||||
username: String,
|
||||
message: String,
|
||||
new_message: bool,
|
||||
}
|
||||
impl UserMessage {
|
||||
pub fn new(
|
||||
id: u32,
|
||||
timestamp: u32,
|
||||
username: String,
|
||||
message: String,
|
||||
new_message: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
id,
|
||||
timestamp,
|
||||
username,
|
||||
message,
|
||||
new_message,
|
||||
}
|
||||
}
|
||||
pub fn print(&self) {
|
||||
debug!(
|
||||
"Timestamp: {}. User: {}, Id: #{}, New message: {} Message: {}",
|
||||
self.timestamp,
|
||||
self.username,
|
||||
self.id,
|
||||
self.new_message,
|
||||
self.message
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug)]
|
||||
pub struct Rooms {
|
||||
pub public_rooms: Vec<Room>,
|
||||
pub owned_private_rooms: Vec<Room>,
|
||||
pub private_rooms: Vec<Room>,
|
||||
pub operated_private_rooms: Vec<Room>,
|
||||
}
|
||||
impl Rooms {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
public_rooms: Vec::new(),
|
||||
owned_private_rooms: Vec::new(),
|
||||
private_rooms: Vec::new(),
|
||||
operated_private_rooms: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn print(&self) {
|
||||
info!("Public rooms ({}):", self.public_rooms.len());
|
||||
for room in &self.public_rooms {
|
||||
room.print();
|
||||
}
|
||||
info!("Owned private rooms ({}):", self.owned_private_rooms.len());
|
||||
for room in &self.owned_private_rooms {
|
||||
room.print();
|
||||
}
|
||||
info!("Private rooms ({}):", self.private_rooms.len());
|
||||
for room in &self.private_rooms {
|
||||
room.print();
|
||||
}
|
||||
info!(
|
||||
"Operated private rooms ({}):",
|
||||
self.operated_private_rooms.len()
|
||||
);
|
||||
for room in &self.operated_private_rooms {
|
||||
room.print();
|
||||
}
|
||||
}
|
||||
}
|
||||
#[derive(Debug)]
|
||||
pub struct Room {
|
||||
name: String,
|
||||
number_of_users: i32,
|
||||
}
|
||||
impl Room {
|
||||
#[allow(dead_code)]
|
||||
pub fn new(name: String, number_of_users: i32) -> Self {
|
||||
Self {
|
||||
name,
|
||||
number_of_users,
|
||||
}
|
||||
}
|
||||
#[allow(dead_code)]
|
||||
pub fn set_number_of_users(&mut self, number_of_users: i32) {
|
||||
self.number_of_users = number_of_users;
|
||||
}
|
||||
pub fn print(&self) {
|
||||
debug!(
|
||||
"Room: {}, Number of users: {}",
|
||||
self.name, self.number_of_users
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
pub enum ServerOperation {
|
||||
LoginStatus(bool),
|
||||
SendMessage(Message),
|
||||
#[allow(dead_code)]
|
||||
ConnectToPeer(Peer),
|
||||
PierceFirewall(u32),
|
||||
GetPeerAddress(String),
|
||||
GetPeerAddressResponse {
|
||||
username: String,
|
||||
host: String,
|
||||
port: u32,
|
||||
obfuscation_type: u32,
|
||||
obfuscated_port: u16,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Server {
|
||||
address: PeerAddress,
|
||||
sender: Sender<ServerOperation>,
|
||||
context: Arc<RwLock<Context>>,
|
||||
listen_port: u32,
|
||||
enable_listen: bool,
|
||||
}
|
||||
impl Server {
|
||||
pub fn new(
|
||||
address: PeerAddress,
|
||||
client_channel: Sender<ClientOperation>,
|
||||
listen_port: u32,
|
||||
enable_listen: bool,
|
||||
) -> std::result::Result<Self, io::Error> {
|
||||
let (sender, server_channel): (
|
||||
Sender<ServerOperation>,
|
||||
Receiver<ServerOperation>,
|
||||
) = mpsc::channel();
|
||||
|
||||
let context = Arc::new(RwLock::new(Context::new()));
|
||||
|
||||
let mut server = Self {
|
||||
address,
|
||||
context,
|
||||
sender,
|
||||
listen_port,
|
||||
enable_listen,
|
||||
};
|
||||
|
||||
server.start_read_write_loops(server_channel, client_channel)?;
|
||||
Ok(server)
|
||||
}
|
||||
|
||||
pub fn get_address(&self) -> &PeerAddress {
|
||||
&self.address
|
||||
}
|
||||
|
||||
pub fn get_sender(&self) -> &Sender<ServerOperation> {
|
||||
&self.sender
|
||||
}
|
||||
|
||||
/// Start reading and writing loops in separate threads
|
||||
fn start_read_write_loops(
|
||||
&mut self,
|
||||
server_channel: Receiver<ServerOperation>,
|
||||
client_channel: Sender<ClientOperation>,
|
||||
) -> std::result::Result<(), io::Error> {
|
||||
let socket_address =
|
||||
format!("{}:{}", self.address.host, self.address.port)
|
||||
.to_socket_addrs()?
|
||||
.next()
|
||||
.ok_or_else(|| {
|
||||
io::Error::new(
|
||||
io::ErrorKind::InvalidInput,
|
||||
"Invalid address",
|
||||
)
|
||||
})?;
|
||||
|
||||
info!(
|
||||
"Connecting to server at {}:{}",
|
||||
self.address.host, self.address.port
|
||||
);
|
||||
|
||||
let stream = TcpStream::connect_timeout(
|
||||
&socket_address,
|
||||
Duration::from_secs(10),
|
||||
)?;
|
||||
|
||||
stream.set_read_timeout(Some(Duration::from_secs(5)))?;
|
||||
stream.set_write_timeout(Some(Duration::from_secs(5)))?;
|
||||
stream.set_nodelay(true)?;
|
||||
|
||||
let mut read_stream = stream.try_clone()?;
|
||||
let mut write_stream = stream.try_clone()?;
|
||||
|
||||
let barrier = Arc::new(Barrier::new(3));
|
||||
let read_barrier = barrier.clone();
|
||||
let write_barrier = barrier.clone();
|
||||
let done_barrier = barrier.clone();
|
||||
let sender = self.sender.clone();
|
||||
|
||||
thread::spawn(move || {
|
||||
read_barrier.wait();
|
||||
|
||||
let mut handlers = Handlers::new();
|
||||
|
||||
handlers.register_handler(LoginHandler);
|
||||
handlers.register_handler(RoomListHandler);
|
||||
handlers.register_handler(ExcludedSearchPhrasesHandler);
|
||||
handlers.register_handler(PrivilegedUsersHandler);
|
||||
handlers.register_handler(MessageUser);
|
||||
handlers.register_handler(WishListIntervalHandler);
|
||||
handlers.register_handler(ParentMinSpeedHandler);
|
||||
handlers.register_handler(ParentSpeedRatioHandler);
|
||||
handlers.register_handler(PrivilegedUsersHandler);
|
||||
handlers.register_handler(FileSearchHandler);
|
||||
handlers.register_handler(GetPeerAddressHandler);
|
||||
handlers.register_handler(ConnectToPeerHandler);
|
||||
|
||||
let dispatcher =
|
||||
MessageDispatcher::new("server".to_string(), sender, handlers);
|
||||
|
||||
let mut buffered_reader = MessageReader::new();
|
||||
loop {
|
||||
match buffered_reader.read_from_socket(&mut read_stream) {
|
||||
Ok(_) => {}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
|
||||
continue
|
||||
}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
|
||||
debug!("Read operation timed out");
|
||||
continue;
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Error reading from server: {}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
loop {
|
||||
match buffered_reader.extract_message() {
|
||||
Ok(Some(mut message)) => {
|
||||
trace!(
|
||||
"[server] ← {:?} - {}",
|
||||
message
|
||||
.get_message_name(
|
||||
MessageType::Server,
|
||||
message.get_message_code_u32()
|
||||
)
|
||||
.map_err(|e| e.to_string()),
|
||||
message.get_message_code_u32()
|
||||
);
|
||||
dispatcher.dispatch(&mut message)
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Error extracting message: {}", e);
|
||||
break;
|
||||
}
|
||||
Ok(None) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let context = self.context.clone();
|
||||
thread::spawn(move || {
|
||||
write_barrier.wait();
|
||||
loop {
|
||||
if let Ok(operation) = server_channel.recv() {
|
||||
match operation {
|
||||
ServerOperation::ConnectToPeer(peer) => {
|
||||
if let Some(op) = match peer.connection_type {
|
||||
ConnectionType::P | ConnectionType::F => {
|
||||
Some(ClientOperation::ConnectToPeer(
|
||||
peer.clone(),
|
||||
))
|
||||
}
|
||||
ConnectionType::D => None,
|
||||
} {
|
||||
client_channel.send(op).unwrap();
|
||||
}
|
||||
}
|
||||
ServerOperation::LoginStatus(message) => {
|
||||
context.write().unwrap().logged_in = Some(message);
|
||||
}
|
||||
ServerOperation::PierceFirewall(token) => {
|
||||
if let Err(e) = write_stream.write_all(
|
||||
&MessageFactory::build_pierce_firewall_message(
|
||||
token,
|
||||
)
|
||||
.get_buffer(),
|
||||
) {
|
||||
error!(
|
||||
"Error writing PierceFirewall message: {}",
|
||||
e
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
ServerOperation::SendMessage(message) => {
|
||||
match write_stream.write_all(&message.get_buffer())
|
||||
{
|
||||
Ok(_) => trace!(
|
||||
"[server] → {:?} - {:}",
|
||||
message
|
||||
.get_message_name(
|
||||
MessageType::Server,
|
||||
message.get_message_code_send()
|
||||
as u32
|
||||
)
|
||||
.map_err(|e| e.to_string()),
|
||||
message.get_message_code_send() as u32
|
||||
),
|
||||
Err(e) => error!(
|
||||
"Error writing message to server: {}",
|
||||
e
|
||||
),
|
||||
}
|
||||
}
|
||||
ServerOperation::GetPeerAddress(username) => {
|
||||
if let Err(e) = write_stream.write_all(
|
||||
&MessageFactory::build_get_peer_address(
|
||||
&username,
|
||||
)
|
||||
.get_buffer(),
|
||||
) {
|
||||
error!("Error writing get_peeer_address message: {}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
ServerOperation::GetPeerAddressResponse {
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port,
|
||||
} => {
|
||||
debug!(
|
||||
"Received GetPeerAddress response for {}: {}:{} (obf_type: {}, obf_port: {})",
|
||||
username, host, port, obfuscation_type, obfuscated_port
|
||||
);
|
||||
|
||||
// Forward the response to the client channel
|
||||
if let Err(e) = client_channel.send(
|
||||
ClientOperation::GetPeerAddressResponse {
|
||||
username,
|
||||
host,
|
||||
port,
|
||||
obfuscation_type,
|
||||
obfuscated_port,
|
||||
},
|
||||
) {
|
||||
error!("Error forwarding GetPeerAddress response to client: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
done_barrier.wait();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn queue_message(&self, message: Message) {
|
||||
match self.sender.send(ServerOperation::SendMessage(message)) {
|
||||
Ok(_) => {}
|
||||
Err(e) => error!("Failed to send: {}", e),
|
||||
}
|
||||
}
|
||||
pub fn login(&self, username: &str, password: &str) -> Result<bool> {
|
||||
// Send the login message
|
||||
self.queue_message(MessageFactory::build_login_message(
|
||||
username, password,
|
||||
));
|
||||
let context = self.context.clone();
|
||||
let mut logged_in;
|
||||
let timeout = Duration::from_secs(5);
|
||||
let start = Instant::now();
|
||||
|
||||
// wait till server says your logged in or not
|
||||
loop {
|
||||
if start.elapsed() > timeout {
|
||||
warn!("Timeout waiting for login response");
|
||||
return Err(SoulseekRs::Timeout);
|
||||
}
|
||||
|
||||
{
|
||||
logged_in = context.read().unwrap().logged_in
|
||||
}
|
||||
|
||||
if logged_in.is_some() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if logged_in.unwrap() {
|
||||
info!("Logged in as {}", username);
|
||||
self.queue_message(MessageFactory::build_shared_folders_message(
|
||||
1, 499,
|
||||
));
|
||||
self.queue_message(MessageFactory::build_no_parent_message());
|
||||
self.queue_message(MessageFactory::build_set_status_message(2));
|
||||
if self.enable_listen {
|
||||
self.queue_message(
|
||||
MessageFactory::build_set_wait_port_message(
|
||||
self.listen_port,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(logged_in.unwrap())
|
||||
}
|
||||
|
||||
pub fn file_search(&self, token: u32, query: &str) {
|
||||
self.queue_message(MessageFactory::build_file_search_message(
|
||||
token, query,
|
||||
));
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn pierce_firewall(&self, token: u32) {
|
||||
match self.sender.send(ServerOperation::PierceFirewall(token)) {
|
||||
Ok(_) => {}
|
||||
Err(e) => error!("Failed to send PierceFirewall: {}", e),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user