Bommels05/BankClient
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup & Fixes

Browse Source
This commit is contained in:
Bommels05 committed 2025-02-05 18:36:16 +01:00
1 parent 340b8db1b9
commit 68de25838c
6 files changed
+510 -765
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

banklib/Cargo.toml
+1 -1
View File
@@ -19,6 +19,6 @@ tokio-util.workspace = true
tracing.workspace = true
rand.workspace = true
url.workspace = true
flume.workspace =true
flume.workspace = true
error-stack.workspace = true
thiserror.workspace = true
banklib/src/lib.rs
+53 -327
View File
@@ -23,11 +23,12 @@ pub mod config;
pub struct BankClient {
client: Client<Handler>,
rx: Receiver<ResponseMessage>,
rx_msg: Receiver<ResponseMessage>,
rx_act: Receiver<ClientAction>,
handle: JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>>
}
#[derive(Error, Debug)]
#[derive(Error, Debug, Clone)]
pub enum BankError {
#[error("Request timed out")]
TimedOut,
@@ -41,8 +42,8 @@ pub enum BankError {
DestinationUnknown,
#[error("You don't have enough Steam Coins")]
NotEnoughMoney,
#[error("Unknown unexpected message received")]
UnexpectedMessage
#[error("Unknown unexpected message received {_0:?}")]
UnexpectedMessage(ResponseMessage)
}
pub type BankResult<T, E = BankError> = error_stack::Result<T, E>;
@@ -52,42 +53,38 @@ impl BankClient {
let url = url.into();
let client_config = ClientConfig::new(url);
let random: u8 = random();
let (tx, rx) = flume::unbounded();
let (tx_msg, rx_msg) = flume::unbounded();
let (tx_act, rx_act) = flume::unbounded();
let (client, future) = ezsockets::connect(
move |client| Handler {
client,
tx,
tx_msg,
tx_act,
username: String::new(),
password: String::new(),
id: random as i32,
receivers: VecDeque::new(),
},
client_config,
}, client_config,
).await;
let handle = tokio::spawn(future);
Self {
client,
rx,
handle,
}
Self { client, rx_msg, rx_act, handle }
}
pub async fn send_authenticated(
&self,
message: AuthenticatedMessage,
accept: fn(&ResponseMessage) -> bool,
) -> BankResult<Response> {
pub async fn send_authenticated(&self, message: AuthenticatedMessage, accept: fn(&ResponseMessage) -> bool) -> BankResult<Response> {
self.check_running()?;
let (tx, rx) = oneshot::channel();
self.client.call(ClientCommand::SendAuthenticatedNew {message, accept, tx}).map_err(|_| report!(BankError::InternalError))?;
tokio::time::timeout(Duration::from_secs(2), rx.into_future()).await.map_err(|_| report!(BankError::TimedOut))?.map_err(|_| report!(BankError::InternalError))?.map_err(|err| report!(err))
tokio::time::timeout(Duration::from_secs(5), rx.into_future()).await.map_err(|_| report!(BankError::TimedOut))?.map_err(|_| report!(BankError::InternalError))?.map_err(|err| report!(err))
}
pub fn set_credentials(&self, username: String, password: String) -> BankResult<()> {
self.client.call(ClientCommand::ChangeCredentials {username, password}).map_err(|_| report!(BankError::InternalError))
}
pub fn receiver(&self) -> &Receiver<ResponseMessage> {
&self.rx
pub fn receiver_msg(&self) -> &Receiver<ResponseMessage> {
&self.rx_msg
}
pub fn receiver_act(&self) -> &Receiver<ClientAction> {
&self.rx_act
}
fn check_running(&self) -> BankResult<(), BankError>{
@@ -108,22 +105,11 @@ impl BankClient {
struct Handler {
client: Client<Handler>,
//config: Arc<Mutex<Config>>,
tx: Sender<ResponseMessage>,
tx_msg: Sender<ResponseMessage>,
tx_act: Sender<ClientAction>,
username: String,
password: String,
//last_username: String,
//last_password: String,
id: i32,
//logged_in: bool,
/*message_queue: VecDeque<(
AuthenticatedMessage,
Box<dyn Future<Output = ()> + Send + Unpin>,
)>,
send_time: SystemTime,
retry_count: u8,
inbox_message_count: u8,
pm_message_count: u8,*/
receivers: VecDeque<(fn(&ResponseMessage) -> bool, oneshot::Sender<Result<Response, BankError>>)>
}
@@ -135,34 +121,7 @@ impl Handler {
}
}
/*fn last_credentials(&self) -> Credentials {
Credentials {
username: &self.last_username,
password: &self.last_password,
}
}*/
/*fn send_authenticated(&mut self, message: AuthenticatedMessage) {
self.send_authenticated_callback(message);
}*/
/*fn send_authenticated_callback(
&mut self,
message: AuthenticatedMessage,
callback: Box<dyn Future<Output = ()> + Send + Unpin>,
) {
if self.message_queue.is_empty() {
self.send_authenticated_now(message, callback);
} else {
self.message_queue.push_back((message, callback));
}
}*/
fn send_authenticated(
&mut self,
message: AuthenticatedMessage,
//callback: Box<dyn Future<Output = ()> + Send + Unpin>,
) {
fn send_authenticated(&mut self, message: AuthenticatedMessage) {
#[derive(Serialize)]
struct WithCredentials<'a> {
#[serde(flatten)]
@@ -172,40 +131,20 @@ impl Handler {
id: i32,
}
/*if message == AuthenticatedMessage::Authenticate && self.logged_in {
self.send_authenticated_now(AuthenticatedMessage::Logout, Box::new(Box::pin(async {})));
self.send_authenticated(message);
return;
}*/
//self.send_time = SystemTime::now();
//self.message_queue.push_front((message.clone(), callback));
debug!("Sending: {message:?}");
let credentials: Credentials = self.credentials();
self.client
.text(
serde_json::to_string(&WithCredentials {
credentials,
message,
id: self.id,
})
.unwrap(),
)
.expect("Could not send authenticated message");
self.client.text(serde_json::to_string(&WithCredentials {
credentials,
message,
id: self.id,
}).unwrap(), ).expect("Could not send authenticated message");
}
/*fn backup_auth(&mut self) {
if !self.last_username.is_empty() && !self.last_password.is_empty() {
self.username = self.last_username.clone();
self.password = self.last_password.clone();
self.send_authenticated(AuthenticatedMessage::Authenticate)
fn clear_receivers(&mut self, error: BankError) {
for (_, tx) in self.receivers.drain(..) {
let _ = tx.send(Err(error.clone()));
}
}
fn auth_error(&self) {
warn!("Authentication error while logged in. Did the password change?");
}*/
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -254,7 +193,7 @@ pub enum AuthenticatedMessage {
Motd,
}
#[derive(Deserialize)]
#[derive(Deserialize, Clone, Debug)]
#[serde(tag = "action", rename_all = "camelCase")]
pub enum Response {
DisplayMessage {
@@ -286,7 +225,7 @@ pub enum Response {
},
}
#[derive(Deserialize)]
#[derive(Deserialize, Clone, Debug)]
pub struct ResponseMessage {
#[serde(flatten)]
pub message: Response,
@@ -309,7 +248,6 @@ impl ezsockets::ClientExt for Handler {
async fn on_text(&mut self, text: String) -> Result<(), Error> {
debug!("Received: {text}");
let message: ResponseMessage =
serde_json::from_str(text.as_str()).expect("Error decoding message");
if message.id == self.id || message.id == -1 {
@@ -325,163 +263,21 @@ impl ezsockets::ClientExt for Handler {
_ => {}
}
if let Some((index, (_, _))) = self.receivers.iter().enumerate().find(|(_, (accept, _))| accept(&message)) {
let mut found = false;
while let Some((index, (_, _))) = self.receivers.iter().enumerate().find(|(_, (accept, _))| accept(&message)) {
let (_, tx) = self.receivers.remove(index).unwrap();
if let Err(_) = tx.send(Ok(message.message)) {
warn!("Message receiver dropped");
found = true;
if let Err(message) = tx.send(Ok(message.message.clone())) {
warn!("Message receiver dropped {:?}", message);
continue;
}
return Ok(());
}
let _ = self.tx.send(message);
/*if self.message_queue.is_empty() {
warn!("Received unexpected message from server");
if !found {
let _ = self.tx_msg.send(message);
}
match message.message {
Response::DisplayMessage { message, id, value } => {
if !message.is_empty() {
println!("{message}");
}
let config = self.config.lock().unwrap();
let show_motd_after_login = config.general.show_motd_after_login;
let show_inbox_count_after_login = config.general.show_inbox_count_after_login;
let show_pm_inbox_count_after_login =
config.general.show_pm_inbox_count_after_login;
let show_balance_after_payment = config.general.show_balance_after_payment;
drop(config);
match id.as_str() {
"auth.success" => {
self.id = value as i32;
self.last_username = self.username.clone();
self.last_password = self.password.clone();
self.logged_in = true;
if show_motd_after_login {
self.send_authenticated(AuthenticatedMessage::Motd);
}
if show_inbox_count_after_login {
self.send_authenticated(AuthenticatedMessage::GetInbox {
message_id: -1,
})
}
if show_pm_inbox_count_after_login {
self.send_authenticated(AuthenticatedMessage::GetPmInbox {
message_id: -1,
})
}
println!("(Client Id: {})", self.id);
}
"auth.fail.credentials" => {
self.backup_auth();
}
"register.success" => {
self.send_authenticated(AuthenticatedMessage::Authenticate);
}
"register.fail.usernameTaken" => {
self.backup_auth();
}
"logout.success" => {
self.logged_in = false;
}
"logout.fail.credentials" => {
self.auth_error();
}
"logout.fail.notloggedin" => {
self.logged_in = false;
}
"balance.success" => {}
"balance.fail.credentials" => {
self.auth_error();
}
"pay.fail.negative_amount" => {
panic!("Should not be able to send a negative amount")
}
"pay.fail.orig_user_unknown" => {
self.auth_error();
}
"pay.fail.credentials" => {
self.auth_error();
}
"pay.fail.unknown_dest" => {}
"pay.fail.not_enough_money" => {
if show_balance_after_payment {
self.send_authenticated(AuthenticatedMessage::GetBalance);
}
}
"pay.fail.unknown_error" => {}
"pay.success" => {
if show_balance_after_payment {
self.send_authenticated(AuthenticatedMessage::GetBalance);
}
}
"pay.received" => {
if show_balance_after_payment {
self.send_authenticated(AuthenticatedMessage::GetBalance);
}
}
"message_count" => {
self.inbox_message_count = value as u8;
println!(
"You have {} unread message{}",
self.inbox_message_count,
if self.inbox_message_count == 1 {
""
} else {
"s"
}
)
}
"pm_message_count" => {
self.pm_message_count = value as u8;
println!(
"You have {} unread private message{}",
self.pm_message_count,
if self.pm_message_count == 1 { "" } else { "s" }
)
}
"pm_inbox.send.success" => {}
"pm_inbox.dest.unkown" => {}
"read_inbox.success" => {}
"read_inbox.fail.credentials" => {
self.auth_error();
}
"read_pm.success" => {}
"read_pm.fail.credentials" => {
self.auth_error();
}
_ => {
warn!("Unknown message id: {}", id)
}
}
}
Response::Inbox {
message_id,
message,
} => {
println!("{}. {}", message_id, message);
}
Response::PmInbox {
message_id,
message,
} => {
println!("{}. {}", message_id, message);
}
Response::Motd { motd } => {
println!("--- {motd} ---");
}
}
if !self.message_queue.is_empty() {
self.message_queue.pop_front().unwrap().1.await;
if !self.message_queue.is_empty() {
let queued = self.message_queue.pop_front().unwrap();
self.send_authenticated_now(queued.0, queued.1);
}
}*/
} else {
debug!("ignoring message for other id")
}
@@ -495,11 +291,7 @@ impl ezsockets::ClientExt for Handler {
async fn on_call(&mut self, call: Self::Call) -> Result<(), Error> {
match call {
ClientCommand::SendAuthenticatedNew {
message,
accept,
tx,
} => {
ClientCommand::SendAuthenticatedNew { message, accept, tx, } => {
self.send_authenticated(message);
self.receivers.push_back((accept, tx));
}
@@ -510,116 +302,50 @@ impl ezsockets::ClientExt for Handler {
self.username = username;
self.password = password;
}
/*ClientCommand::TryLastLogin => {
if self.last_username.is_empty() || self.last_password.is_empty() {
println!("You need to enter a username and password");
} else {
self.send_authenticated(AuthenticatedMessage::Authenticate);
}
}
ClientCommand::CheckTimeout => {
if !self.message_queue.is_empty() && self.send_time.elapsed().unwrap().as_secs() > 5
{
if self.retry_count >= 3 {
self.message_queue.pop_front();
self.retry_count = 0;
println!("Request Timeout - All Retries failed")
} else {
self.retry_count += 1;
println!("Request Timeout - Retrying {}/3", self.retry_count);
}
if !self.message_queue.is_empty() {
let queued = self.message_queue.pop_front().unwrap();
self.send_authenticated_now(queued.0, queued.1);
}
}
}
ClientCommand::SendWithCallback(message, callback) => {
self.send_authenticated_callback(message, callback);
}
ClientCommand::PrintInbox => {
for i in 0..self.inbox_message_count {
self.send_authenticated(AuthenticatedMessage::GetInbox {
message_id: (i + 1) as i8,
})
}
}
ClientCommand::PrintPmInbox => {
for i in 0..self.pm_message_count {
self.send_authenticated(AuthenticatedMessage::GetPmInbox {
message_id: (i + 1) as i8,
})
}
}
ClientCommand::TryAccount(success, error) => {
//let last_username = self.last_username.clone();
//let last_password = self.last_password.clone();
self.last_username = String::new();
self.last_password = String::new();
/*self.send_authenticated_callback(AuthenticatedMessage::Authenticate, Box::new(Box::pin(async move {
if !self.logged_in {
self.send_authenticated_callback(AuthenticatedMessage::Register, Box::new(Box::pin(async move {
if !self.logged_in {
error.await;
} else {
success.await;
}
})));
} else {
success.await;
}
})));
*/
}*/
}
Ok(())
}
async fn on_connect(&mut self) -> Result<(), Error> {
info!("Successfully connected to the Server");
if !self.username.is_empty() && !self.password.is_empty() {
self.send_authenticated(AuthenticatedMessage::Authenticate);
}
let _ = self.tx_act.send(ClientAction::Connected);
Ok(())
}
async fn on_connect_fail(&mut self, _error: WSError) -> Result<ClientCloseMode, Error> {
info!("Disconnected, reconnecting...");
info!("Connection failed, reconnecting...");
Ok(ClientCloseMode::Reconnect)
}
async fn on_close(&mut self, _frame: Option<CloseFrame>) -> Result<ClientCloseMode, Error> {
self.clear_receivers(BankError::Disconnected);
let _ = self.tx_act.send(ClientAction::Disconnected);
info!("Disconnected, reconnecting...");
Ok(ClientCloseMode::Reconnect)
}
async fn on_disconnect(&mut self) -> Result<ClientCloseMode, Error> {
self.clear_receivers(BankError::Disconnected);
let _ = self.tx_act.send(ClientAction::Disconnected);
info!("Disconnected, reconnecting...");
Ok(ClientCloseMode::Reconnect)
}
}
pub enum ClientAction {
Connected,
Disconnected
}
pub enum ClientCommand {
SendAuthenticatedNew {
message: AuthenticatedMessage,
accept: fn(&ResponseMessage) -> bool,
tx: tokio::sync::oneshot::Sender<Result<Response, BankError>>,
tx: oneshot::Sender<Result<Response, BankError>>,
},
SendAuthenticated(AuthenticatedMessage),
/*SendWithCallback(
AuthenticatedMessage,
Box<dyn Future<Output = ()> + Send + Unpin>,
),
PrintInbox,
PrintPmInbox,*/
ChangeCredentials {
username: String,
password: String,
},
/*TryLastLogin,
CheckTimeout,
TryAccount(
Box<dyn Future<Output = ()> + Send + Unpin>,
Box<dyn Future<Output = ()> + Send + Unpin>,
),*/
}