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

⬆️ (Cargo.toml): bump version from 0.1.6 to 0.1.7 for new release

Browse Source 
♻️ (various): refactor code to use underscore prefix for unused variables
✨ (ls_client.rs): add constants for WebSocket connection setup
✨ (various): add unimplemented!() stubs for methods to be implemented
🔥 (main.rs): remove unused imports and commented code
🔥 (ls_client.rs): remove unused HashMap import and commented code
🐛 (subscription.rs): fix get_command_value to correctly format key before lookup
🔧 (Cargo.toml): remove hyper dependency as it's no longer used
This commit is contained in:
Daniel López Azaña 2024-04-07 21:14:31 +02:00
parent 2a68c66704
commit 7d7f380e30
9 changed files with 65 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "lightstreamer-client"
version = "0.1.6"
version = "0.1.7"
edition = "2021"
authors = ["Daniel López Azaña <daniloaz@gmail.com>"]
description = "A Rust client for Lightstreamer, designed to facilitate real-time communication with Lightstreamer servers."
@ -14,7 +14,6 @@ documentation = "https://github.com/daniloaz/lightstreamer-client#readme"
cookie = { version = "0", features = ["percent-encode"]}
futures = "0"
futures-util = "0"
hyper = { version = "1", features = ["full"] }
reqwest = { version = "0", features = ["json", "stream"] }
serde = { version = "1", features = ["derive"] }
serde_json = { version = "1" }

9
src/client_listener.rs
View File

@ -63,8 +63,9 @@ pub trait ClientListener: Debug + Send {
/// See also `LightstreamerClient.connectionDetails`
///
/// See also `LightstreamerClient.connectionOptions`
fn on_property_change(&self, property: &str) {
fn on_property_change(&self, _property: &str) {
// Implementation for on_property_change
unimplemented!("Implement on_property_change method for ClientListener");
}
/// Event handler that is called when the Server notifies a refusal on the client attempt
@ -101,8 +102,9 @@ pub trait ClientListener: Debug + Send {
/// See also `onStatusChange()`
///
/// See also `ConnectionDetails.setAdapterSet()`
fn on_server_error(&self, code: i32, message: &str) {
fn on_server_error(&self, _code: i32, _message: &str) {
// Implementation for on_server_error
unimplemented!("Implement on_server_error method for ClientListener");
}
/// Event handler that receives a notification each time the `LightstreamerClient` status has changed.
@ -179,7 +181,8 @@ pub trait ClientListener: Debug + Send {
/// See also `LightstreamerClient.disconnect()`
///
/// See also `LightstreamerClient.getStatus()`
fn on_status_change(&self, status: &str) {
fn on_status_change(&self, _status: &str) {
// Implementation for on_status_change
unimplemented!("Implement on_status_change method for ClientListener");
}
}

15
src/client_message_listener.rs
View File

@ -15,8 +15,9 @@ pub trait ClientMessageListener {
/// * `sent_on_network`: `true` if the message was sent on the network, `false` otherwise.
/// Even if the flag is `true`, it is not possible to infer whether the message actually
/// reached the Lightstreamer Server or not.
fn on_abort(&self, msg: &str, sent_on_network: bool) {
fn on_abort(&self, _msg: &str, _sent_on_network: bool) {
// Implementation for on_abort
unimplemented!("Implement on_abort method for ClientMessageListener.");
}
/// Event handler that is called by Lightstreamer when the related message has been processed
@ -29,8 +30,9 @@ pub trait ClientMessageListener {
/// - `<= 0`: the Metadata Adapter has refused the message; the code value is dependent
/// on the specific Metadata Adapter implementation.
/// * `error`: the description of the error sent by the Server.
fn on_deny(&self, msg: &str, code: i32, error: &str) {
fn on_deny(&self, _msg: &str, _code: i32, _error: &str) {
// Implementation for on_deny
unimplemented!("Implement on_deny method for ClientMessageListener.");
}
/// Event handler that is called by Lightstreamer to notify that the related message has
@ -40,8 +42,9 @@ pub trait ClientMessageListener {
/// # Parameters
///
/// * `msg`: the message to which this notification is related.
fn on_discarded(&self, msg: &str) {
fn on_discarded(&self, _msg: &str) {
// Implementation for on_discarded
unimplemented!("Implement on_discarded method for ClientMessageListener.");
}
/// Event handler that is called by Lightstreamer when the related message has been processed
@ -51,8 +54,9 @@ pub trait ClientMessageListener {
/// # Parameters
///
/// * `msg`: the message to which this notification is related.
fn on_error(&self, msg: &str) {
fn on_error(&self, _msg: &str) {
// Implementation for on_error
unimplemented!("Implement on_error method for ClientMessageListener.");
}
/// Event handler that is called by Lightstreamer when the related message has been processed
@ -63,7 +67,8 @@ pub trait ClientMessageListener {
/// * `msg`: the message to which this notification is related.
/// * `response`: the response from the Metadata Adapter. If not supplied (i.e. supplied as `None`),
/// an empty message is received here.
fn on_processed(&self, msg: &str, response: Option<&str>) {
fn on_processed(&self, _msg: &str, _response: Option<&str>) {
// Implementation for on_processed
unimplemented!("Implement on_processed method for ClientMessageListener.");
}
}

2
src/connection_details.rs
View File

@ -359,7 +359,7 @@ impl ConnectionDetails {
/// # Parameters
///
/// * `listener`: The listener to be removed.
pub fn remove_listener(&mut self, listener: Box<dyn ClientListener>) {
pub fn remove_listener(&mut self, _listener: Box<dyn ClientListener>) {
unimplemented!("Implement mechanism to remove listener from ConnectionDetails.");
//self.listeners.remove(&listener);
}

2
src/item_update.rs
View File

@ -106,7 +106,7 @@ impl ItemUpdate {
.changed_fields
.iter()
.enumerate()
.map(|(i, (k, v))| (i + 1, v.clone()))
.map(|(i, (_k, v))| (i + 1, v.clone()))
.collect();
changed_fields_by_pos
}

106
src/ls_client.rs
View File

@ -8,7 +8,6 @@ use crate::util::*;
use cookie::Cookie;
use futures_util::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::error::Error;
use std::fmt::{self, Debug, Formatter};
use std::sync::Arc;
@ -123,6 +122,14 @@ impl LightstreamerClient {
/// A constant string representing the version of the library.
pub const LIB_VERSION: &'static str = "0.1.0";
//
// Constants for WebSocket connection.
//
pub const SEC_WEBSOCKET_KEY: &'static str = "PNDUibe9ex7PnsrLbt0N4w==";
pub const SEC_WEBSOCKET_PROTOCOL: &'static str ="TLCP-2.4.0.lightstreamer.com";
pub const SEC_WEBSOCKET_VERSION: &'static str = "13";
pub const SEC_WEBSOCKET_UPGRADE: &'static str = "websocket";
/// A constant string representing the version of the TLCP protocol used by the library.
pub const TLCP_VERSION: &'static str = "TLCP-2.4.0";
@ -142,8 +149,9 @@ impl LightstreamerClient {
/// * `cookies`: an instance of `http.cookies.SimpleCookie`.
///
/// See also `getCookies()`
pub fn add_cookies(uri: &str, cookies: &Cookie) {
pub fn add_cookies(_uri: &str, _cookies: &Cookie) {
// Implementation for add_cookies
unimplemented!("Implement mechanism to add cookies to LightstreamerClient");
}
/// Adds a listener that will receive events from the `LightstreamerClient` instance.
@ -194,12 +202,15 @@ impl LightstreamerClient {
///
/// See also `ConnectionDetails.setServerAddress()`
pub async fn connect(&mut self, shutdown_signal: Arc<Notify>) -> Result<(), Box<dyn Error>> {
// Check if the server address is configured.
if self.server_address.is_none() {
return Err(Box::new(IllegalStateException::new(
"No server address was configured.",
)));
}
//
// Only WebSocket streaming transport is currently supported.
//
let forced_transport = self.connection_options.get_forced_transport();
if forced_transport.is_none()
|| *forced_transport.unwrap() /* unwrap() is safe here */ != Transport::WsStreaming
@ -208,70 +219,6 @@ impl LightstreamerClient {
"Only WebSocket streaming transport is currently supported.",
)));
}
//
// Add optional parameters
//
/*
if let Some(requested_max_bandwidth) = self.connection_options.get_requested_max_bandwidth() {
params.insert("LS_requested_max_bandwidth", &requested_max_bandwidth.to_string());
}
if let Some(content_length) = self.connection_options.get_content_length() {
params.insert("LS_content_length", &content_length.to_string());
}
if let Some(supported_diffs) = &self.connection_options.get_supported_diffs() {
params.insert("LS_supported_diffs", supported_diffs);
}
if self.connection_options.is_polling() {
params.insert("LS_polling", "true");
if let Some(polling_millis) = self.connection_options.get_polling_millis() {
params.insert("LS_polling_millis", &polling_millis.to_string());
}
if let Some(idle_millis) = self.connection_options.get_idle_millis() {
params.insert("LS_idle_millis", &idle_millis.to_string());
}
} else {
if let Some(inactivity_millis) = self.connection_options.get_inactivity_millis() {
params.insert("LS_inactivity_millis", &inactivity_millis.to_string());
}
if let Some(keepalive_millis) = self.connection_options.get_keepalive_millis() {
params.insert("LS_keepalive_millis", &keepalive_millis.to_string());
}
if !self.connection_options.is_send_sync() {
params.insert("LS_send_sync", "false");
}
}
if self.connection_options.is_reduce_head() {
params.insert("LS_reduce_head", "true");
}
if let Some(ttl_millis) = self.connection_options.get_ttl_millis() {
params.insert("LS_ttl_millis", &ttl_millis.to_string());
}
// Build the request body
let request_body = build_request_body(&params);
// Send the create session request
let response = send_create_session_request(
&self.server_address.as_ref().unwrap(),
&request_body,
)?;
// Process the server response
process_create_session_response(&response)?;
*/
//
// Convert the HTTP URL to a WebSocket URL.
//
@ -312,19 +259,19 @@ impl LightstreamerClient {
)
.header(
HeaderName::from_static("sec-websocket-key"),
HeaderValue::from_static("PNDUibe9ex7PnsrLbt0N4w=="),
HeaderValue::from_static(Self::SEC_WEBSOCKET_KEY),
)
.header(
HeaderName::from_static("sec-websocket-protocol"),
HeaderValue::from_static("TLCP-2.4.0.lightstreamer.com"),
HeaderValue::from_static(Self::SEC_WEBSOCKET_PROTOCOL),
)
.header(
HeaderName::from_static("sec-websocket-version"),
HeaderValue::from_static("13"),
HeaderValue::from_static(Self::SEC_WEBSOCKET_VERSION),
)
.header(
HeaderName::from_static("upgrade"),
HeaderValue::from_static("websocket"),
HeaderValue::from_static(Self::SEC_WEBSOCKET_UPGRADE),
)
.body(())?;
@ -364,7 +311,7 @@ impl LightstreamerClient {
// Start reading and processing messages from the server.
//
let mut request_id: usize = 0;
let mut session_id: Option<String> = None;
let mut _session_id: Option<String> = None;
let mut subscription_id: usize = 0;
loop {
tokio::select! {
@ -613,7 +560,7 @@ impl LightstreamerClient {
///
/// A list with the various cookies that can be sent in a HTTP request for the specified URI.
/// If a `None` URI was supplied, all available non-expired cookies will be returned.
pub fn get_cookies(uri: Option<&str>) -> Cookie {
pub fn get_cookies(_uri: Option<&str>) -> Cookie {
// Implementation for get_cookies
unimplemented!()
}
@ -737,7 +684,7 @@ impl LightstreamerClient {
/// * `listener`: The listener to be removed.
///
/// See also `addListener()`
pub fn remove_listener(&mut self, listener: Box<dyn ClientListener>) {
pub fn remove_listener(&mut self, _listener: Box<dyn ClientListener>) {
unimplemented!("Implement mechanism to remove listener from LightstreamerClient");
//self.listeners.remove(&listener);
}
@ -822,19 +769,19 @@ impl LightstreamerClient {
&mut self,
message: &str,
sequence: Option<&str>,
delay_timeout: Option<u64>,
_delay_timeout: Option<u64>,
listener: Option<Box<dyn ClientMessageListener>>,
enqueue_while_disconnected: bool,
) {
let sequence = sequence.unwrap_or_else(|| "UNORDERED_MESSAGES");
let _sequence = sequence.unwrap_or_else(|| "UNORDERED_MESSAGES");
// Handle the message based on the current connection status
match &self.status {
ClientStatus::Connected(connection_type) => {
ClientStatus::Connected(_connection_type) => {
// Send the message to the server in a separate thread
// ...
}
ClientStatus::Disconnected(disconnection_type) => {
ClientStatus::Disconnected(_disconnection_type) => {
if enqueue_while_disconnected {
// Enqueue the message to be sent when a connection is available
// ...
@ -850,6 +797,7 @@ impl LightstreamerClient {
// ...
}
}
unimplemented!("Complete mechanism to send message to LightstreamerClient.");
}
/// Static method that permits to configure the logging system used by the library. The logging
@ -958,7 +906,7 @@ impl LightstreamerClient {
///
/// * `subscription`: An "active" `Subscription` object that was activated by this `LightstreamerClient`
/// instance.
pub fn unsubscribe(&mut self, subscription: Subscription) {
pub fn unsubscribe(&mut self, _subscription: Subscription) {
unimplemented!("Implement mechanism to unsubscribe from LightstreamerClient.");
}
/*

7
src/main.rs
View File

@ -1,20 +1,13 @@
use hyper::client;
use lightstreamer_client::item_update::ItemUpdate;
use lightstreamer_client::ls_client::{LightstreamerClient, Transport};
use lightstreamer_client::subscription::{Snapshot, Subscription, SubscriptionMode};
use lightstreamer_client::subscription_listener::SubscriptionListener;
use futures::stream::StreamExt;
use futures::SinkExt;
use reqwest::Client;
use serde_urlencoded;
use signal_hook::low_level::signal_name;
use signal_hook::{consts::SIGINT, consts::SIGTERM, iterator::Signals};
use std::error::Error;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::sync::{Notify, Mutex};
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
const MAX_CONNECTION_ATTEMPTS: u64 = 1;

5
src/subscription.rs
View File

@ -177,7 +177,7 @@ impl Subscription {
self.listeners.retain(|l| {
let l_ref = l.as_ref() as &dyn SubscriptionListener;
let listener_ref = listener as &dyn SubscriptionListener;
!(l_ref as *const dyn SubscriptionListener == listener_ref as *const dyn SubscriptionListener)
!(std::ptr::addr_of!(*l_ref) == std::ptr::addr_of!(*listener_ref))
});
}
@ -764,8 +764,9 @@ impl Subscription {
/// # Returns
/// The current value for the specified field of the specified key within the specified item (possibly `None`), or `None` if the specified key has not been added yet (note that it might have been added and eventually deleted).
pub fn get_command_value(&self, item_pos: usize, key: &str, field_pos: usize) -> Option<&String> {
let key = format!("{}_{}", item_pos, key);
self.command_values
.get(key)
.get(&key)
.and_then(|fields| fields.get(&field_pos))
}

24
src/subscription_listener.rs
View File

@ -32,8 +32,9 @@ pub trait SubscriptionListener: Send {
/// - `item_name`: name of the involved item. If the Subscription was initialized using an
/// "Item Group" then a `None` value is supplied.
/// - `item_pos`: 1-based position of the item within the "Item List" or "Item Group".
fn on_clear_snapshot(&mut self, item_name: Option<&str>, item_pos: usize) {
fn on_clear_snapshot(&mut self, _item_name: Option<&str>, _item_pos: usize) {
// Default implementation does nothing.
unimplemented!("Implement on_clear_snapshot method for SubscriptionListener.");
}
/// Event handler that is called by Lightstreamer to notify that, due to internal resource
@ -54,8 +55,9 @@ pub trait SubscriptionListener: Send {
/// - `Subscription::set_requested_max_frequency()`
/// - `Subscription::set_command_second_level_fields()`
/// - `Subscription::set_command_second_level_field_schema()`
fn on_command_second_level_item_lost_updates(&mut self, lost_updates: u32, key: &str) {
fn on_command_second_level_item_lost_updates(&mut self, _lost_updates: u32, _key: &str) {
// Default implementation does nothing.
unimplemented!("Implement on_command_second_level_item_lost_updates method for SubscriptionListener.");
}
/// Event handler that is called when the Server notifies an error on a second-level subscription.
@ -87,8 +89,9 @@ pub trait SubscriptionListener: Send {
/// - `ConnectionDetails::set_adapter_set()`
/// - `Subscription::set_command_second_level_fields()`
/// - `Subscription::set_command_second_level_field_schema()`
fn on_command_second_level_subscription_error(&mut self, code: i32, message: Option<&str>, key: &str) {
fn on_command_second_level_subscription_error(&mut self, _code: i32, _message: Option<&str>, _key: &str) {
// Default implementation does nothing.
unimplemented!("Implement on_command_second_level_subscription_error method for SubscriptionListener.");
}
/// Event handler that is called by Lightstreamer to notify that all snapshot events for an item
@ -114,8 +117,9 @@ pub trait SubscriptionListener: Send {
///
/// - `Subscription::set_requested_snapshot()`
/// - `ItemUpdate::is_snapshot()`
fn on_end_of_snapshot(&mut self, item_name: Option<&str>, item_pos: usize) {
fn on_end_of_snapshot(&mut self, _item_name: Option<&str>, _item_pos: usize) {
// Default implementation does nothing.
unimplemented!("Implement on_end_of_snapshot method for SubscriptionListener.");
}
/// Event handler that is called by Lightstreamer to notify that, due to internal resource
@ -139,8 +143,9 @@ pub trait SubscriptionListener: Send {
/// # See also
///
/// - `Subscription::set_requested_max_frequency()`
fn on_item_lost_updates(&mut self, item_name: Option<&str>, item_pos: usize, lost_updates: u32) {
fn on_item_lost_updates(&mut self, _item_name: Option<&str>, _item_pos: usize, _lost_updates: u32) {
// Default implementation does nothing.
unimplemented!("Implement on_item_lost_updates method for SubscriptionListener.");
}
/// Event handler that is called by Lightstreamer each time an update pertaining to an item
@ -151,8 +156,9 @@ pub trait SubscriptionListener: Send {
/// - `update`: a value object containing the updated values for all the fields, together with
/// meta-information about the update itself and some helper methods that can be used to
/// iterate through all or new values.
fn on_item_update(&mut self, update: ItemUpdate) {
fn on_item_update(&mut self, _update: ItemUpdate) {
// Default implementation does nothing.
unimplemented!("Implement on_item_update method for SubscriptionListener.");
}
/// Event handler that receives a notification when the `SubscriptionListener` instance is
@ -189,8 +195,9 @@ pub trait SubscriptionListener: Send {
/// - `frequency`: A decimal number, representing the maximum frequency applied by the Server
/// (expressed in updates per second), or the string "unlimited". A `None` value is possible in
/// rare cases, when the frequency can no longer be determined.
fn on_real_max_frequency(&mut self, frequency: Option<f64>) {
fn on_real_max_frequency(&mut self, _frequency: Option<f64>) {
// Default implementation does nothing.
unimplemented!("Implement on_real_max_frequency method for SubscriptionListener.");
}
/// Event handler that is called by Lightstreamer to notify that a Subscription has been successfully
@ -244,8 +251,9 @@ pub trait SubscriptionListener: Send {
/// # See also
///
/// - `ConnectionDetails::set_adapter_set()`
fn on_subscription_error(&mut self, code: i32, message: Option<&str>) {
fn on_subscription_error(&mut self, _code: i32, _message: Option<&str>) {
// Default implementation does nothing.
unimplemented!("Implement on_subscription_error method for SubscriptionListener.");
}
/// Event handler that is called by Lightstreamer to notify that a Subscription has been successfully