soaos/lightstreamer-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added option to log messages to tracing over stdout

Browse Source
This commit is contained in:
Alina Hagan 2024-08-09 23:33:25 +01:00
parent 65282048ae
commit 4a1625a311
3 changed files with 86 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.toml
View File

@ -23,4 +23,6 @@ serde_urlencoded = "0"
signal-hook = "0" signal-hook = "0"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] } tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tokio-tungstenite = { version = "0", features = ["native-tls"] } tokio-tungstenite = { version = "0", features = ["native-tls"] }
tracing = "0.1.40"
tracing-subscriber = "0.3.18"
url = "2" url = "2"

3
src/error.rs
View File

@ -1,11 +1,13 @@
use std::error::Error; use std::error::Error;
use std::fmt; use std::fmt;
use tracing::error;
#[derive(Debug)] #[derive(Debug)]
pub struct IllegalArgumentException(String); pub struct IllegalArgumentException(String);
impl IllegalArgumentException { impl IllegalArgumentException {
pub fn new(msg: &str) -> IllegalArgumentException { pub fn new(msg: &str) -> IllegalArgumentException {
error!(msg);
IllegalArgumentException(msg.to_string()) IllegalArgumentException(msg.to_string())
} }
} }
@ -29,6 +31,7 @@ pub struct IllegalStateException {
impl IllegalStateException { impl IllegalStateException {
pub fn new(msg: &str) -> IllegalStateException { pub fn new(msg: &str) -> IllegalStateException {
error!(msg);
IllegalStateException { IllegalStateException {
details: msg.to_string(), details: msg.to_string(),
} }

97
src/ls_client.rs
View File

@ -21,6 +21,7 @@ use tokio_tungstenite::{
Message, Message,
}, },
}; };
use tracing::{debug, error, info, instrument, trace, warn, Level};
use url::Url; use url::Url;
/// Represents the current status of the `LightstreamerClient`. /// Represents the current status of the `LightstreamerClient`.
@ -44,6 +45,11 @@ pub enum DisconnectionType {
TryingRecovery, TryingRecovery,
} }
pub enum LogType {
TracingLogs,
StdLogs,
}
/// Facade class for the management of the communication to Lightstreamer Server. Used to provide /// Facade class for the management of the communication to Lightstreamer Server. Used to provide
/// configuration settings, event handlers, operations for the control of the connection lifecycle, /// configuration settings, event handlers, operations for the control of the connection lifecycle,
/// Subscription handling and to send messages. /// Subscription handling and to send messages.
@ -102,6 +108,8 @@ pub struct LightstreamerClient {
subscriptions: Vec<Subscription>, subscriptions: Vec<Subscription>,
/// The current status of the client. /// The current status of the client.
status: ClientStatus, status: ClientStatus,
/// Logging Type to be used
logging: LogType,
} }
impl Debug for LightstreamerClient { impl Debug for LightstreamerClient {
@ -203,6 +211,7 @@ impl LightstreamerClient {
/// See also `ClientListener.onStatusChange()` /// See also `ClientListener.onStatusChange()`
/// ///
/// See also `ConnectionDetails.setServerAddress()` /// See also `ConnectionDetails.setServerAddress()`
#[instrument]
pub async fn connect(&mut self, shutdown_signal: Arc<Notify>) -> Result<(), Box<dyn Error>> { pub async fn connect(&mut self, shutdown_signal: Arc<Notify>) -> Result<(), Box<dyn Error>> {
// Check if the server address is configured. // Check if the server address is configured.
if self.server_address.is_none() { if self.server_address.is_none() {
@ -281,12 +290,15 @@ impl LightstreamerClient {
let ws_stream = match connect_async(request).await { let ws_stream = match connect_async(request).await {
Ok((ws_stream, response)) => { Ok((ws_stream, response)) => {
if let Some(server_header) = response.headers().get("server") { if let Some(server_header) = response.headers().get("server") {
println!( self.make_log(
Level::INFO,
&format!(
"Connected to Lightstreamer server: {}", "Connected to Lightstreamer server: {}",
server_header.to_str().unwrap_or("") server_header.to_str().unwrap_or("")
),
); );
} else { } else {
println!("Connected to Lightstreamer server"); self.make_log(Level::INFO, "Connected to Lightstreamer server");
} }
ws_stream ws_stream
} }
@ -335,7 +347,7 @@ impl LightstreamerClient {
// Errors from server. // Errors from server.
// //
"conerr" | "reqerr" => { "conerr" | "reqerr" => {
println!("Received connection error from server: {}", clean_text); self.make_log( Level::ERROR, &format!("Received connection error from Lightstreamer server: {}", clean_text) );
break; break;
}, },
// //
@ -343,8 +355,8 @@ impl LightstreamerClient {
// //
"conok" => { "conok" => {
if let Some(session_id) = submessage_fields.get(1).as_deref() { if let Some(session_id) = submessage_fields.get(1).as_deref() {
println!("Session creation confirmed by server: '{}'", clean_text); self.make_log( Level::INFO, &format!("Session creation confirmed by server: {}", clean_text) );
println!("Session created with ID: {:?}", session_id); self.make_log( Level::INFO, &format!("Session created with ID: {:?}", session_id) );
// //
// Subscribe to the desired items. // Subscribe to the desired items.
// //
@ -414,7 +426,7 @@ impl LightstreamerClient {
write_stream write_stream
.send(Message::Text(format!("control\r\n{}", encoded_params))) .send(Message::Text(format!("control\r\n{}", encoded_params)))
.await?; .await?;
println!("Sent subscription request: '{}'", encoded_params); info!("Sent subscription request: '{}'", encoded_params);
} }
} else { } else {
return Err(Box::new(std::io::Error::new( return Err(Box::new(std::io::Error::new(
@ -427,26 +439,25 @@ impl LightstreamerClient {
// Notifications from server. // Notifications from server.
// //
"conf" | "cons" | "clientip" | "servname" | "prog" | "sync" => { "conf" | "cons" | "clientip" | "servname" | "prog" | "sync" => {
println!("Received notification from server: {}", clean_text); self.make_log( Level::INFO, &format!("Received notification from server: {}", clean_text) );
// Don't do anything with these notifications for now. // Don't do anything with these notifications for now.
}, },
"probe" => { "probe" => {
println!("Received probe message from server: '{}'", clean_text); self.make_log( Level::INFO, &format!("Received probe message from server: {}", clean_text ) );
}, },
"reqok" => { "reqok" => {
println!("Received reqok message from server: '{}'", clean_text); self.make_log( Level::INFO, &format!("Received reqok message from server: '{}'", clean_text ) );
}, },
// //
// Subscription confirmation from server. // Subscription confirmation from server.
// //
"subok" => { "subok" => {
println!("Subscription confirmed by server: '{}'", clean_text); self.make_log( Level::INFO, &format!("Subscription confirmed by server: '{}'", clean_text) );
}, },
// //
// Data updates from server. // Data updates from server.
// //
"u" => { "u" => {
//println!("Received data update from server: '{}'", clean_text);
// Parse arguments from the received message. // Parse arguments from the received message.
let arguments = clean_text.split(",").collect::<Vec<&str>>(); let arguments = clean_text.split(",").collect::<Vec<&str>>();
// //
@ -456,8 +467,9 @@ impl LightstreamerClient {
let subscription = match self.get_subscriptions().get(subscription_index-1) { let subscription = match self.get_subscriptions().get(subscription_index-1) {
Some(subscription) => subscription, Some(subscription) => subscription,
None => { None => {
println!("Subscription not found for index: {}", subscription_index); self.make_log( Level::WARN, &format!("Subscription not found for index: {}", subscription_index) );
continue; continue;
} }
}; };
// //
@ -467,7 +479,7 @@ impl LightstreamerClient {
let item = match subscription.get_items() { let item = match subscription.get_items() {
Some(items) => items.get(item_index-1), Some(items) => items.get(item_index-1),
None => { None => {
println!("No items found in subscription: {:?}", subscription); self.make_log( Level::WARN, &format!("No items found in subscription: {:?}", subscription) );
continue; continue;
} }
}; };
@ -673,7 +685,7 @@ impl LightstreamerClient {
// Connection confirmation from server. // Connection confirmation from server.
// //
"wsok" => { "wsok" => {
println!("Connection confirmed by server: '{}'", clean_text); self.make_log( Level::INFO, &format!("Connection confirmed by server: '{}'", clean_text) );
// //
// Request session creation. // Request session creation.
// //
@ -702,7 +714,7 @@ impl LightstreamerClient {
write_stream write_stream
.send(Message::Text(format!("create_session\r\n{}\n", encoded_params))) .send(Message::Text(format!("create_session\r\n{}\n", encoded_params)))
.await?; .await?;
println!("Sent create session request: '{}'", encoded_params); self.make_log( Level::INFO, &format!("Sent create session request: '{}'", encoded_params) );
}, },
unexpected_message => { unexpected_message => {
return Err(Box::new(std::io::Error::new( return Err(Box::new(std::io::Error::new(
@ -732,13 +744,13 @@ impl LightstreamerClient {
))); )));
}, },
None => { None => {
println!("No more messages from server"); self.make_log( Level::INFO, "No more messages from server" );
break; break;
}, },
} }
}, },
_ = shutdown_signal.notified() => { _ = shutdown_signal.notified() => {
println!("Received shutdown signal"); self.make_log( Level::INFO, &format!("Received shutdown signal") );
break; break;
}, },
} }
@ -763,9 +775,10 @@ impl LightstreamerClient {
/// "DISCONNECTED", then nothing will be done. /// "DISCONNECTED", then nothing will be done.
/// ///
/// See also `connect()` /// See also `connect()`
#[instrument]
pub async fn disconnect(&mut self) { pub async fn disconnect(&mut self) {
// Implementation for disconnect // Implementation for disconnect
println!("Disconnecting from Lightstreamer server"); self.make_log( Level::INFO, "Disconnecting from Lightstreamer server" );
} }
/// Static inquiry method that can be used to share cookies between connections to the Server /// Static inquiry method that can be used to share cookies between connections to the Server
@ -895,6 +908,7 @@ impl LightstreamerClient {
listeners: Vec::new(), listeners: Vec::new(),
subscriptions: Vec::new(), subscriptions: Vec::new(),
status: ClientStatus::Disconnected(DisconnectionType::WillRetry), status: ClientStatus::Disconnected(DisconnectionType::WillRetry),
logging: LogType::StdLogs,
}) })
} }
@ -1141,6 +1155,53 @@ impl LightstreamerClient {
} }
} }
*/ */
/// Method setting enum for the logging of this instance.
///
/// Default logging type is StdLogs, corresponding to `stdout`
///
/// `LightstreamerClient` has methods for logging that are compatible with the `Tracing` crate.
/// Enabling logging for the `Tracing` crate requires implementation of a tracing subscriber
/// and its configuration and formatting.
///
/// # Parameters
///
/// * `logging`: An enum declaring the logging type of this `LightstreamerClient` instance.
pub fn set_logging_type(&mut self, logging: LogType) {
self.logging = logging;
}
/// Method for logging messages
///
/// Match case wraps log types. `loglevel` param ignored in StdLogs case, all output to stdout.
///
/// # Parameters
///
/// * `loglevel` Enum determining use of stdout or Tracing subscriber.
pub fn make_log(&mut self, loglevel: Level, log: &str) {
match self.logging {
LogType::StdLogs => {
println!("{}", log);
}
LogType::TracingLogs => match loglevel {
Level::INFO => {
info!(log);
}
Level::WARN => {
warn!(log);
}
Level::ERROR => {
error!(log);
}
Level::TRACE => {
trace!(log);
}
Level::DEBUG => {
debug!(log);
}
},
}
}
} }
/// The transport type to be used by the client. /// The transport type to be used by the client.