Struct Client 

pub struct Client { /* private fields */ }

A live Lightstreamer session.

Get one from Client::connect. It is a handle, not the session itself: the work happens on background tasks, and the client is what you use to steer them.

Shutdown is part of the contract

Dropping the client shuts the session down. The background tasks stop, the socket is closed, and every stream you are still holding ends. You never need std::process::exit to stop a lightstreamer-rs client, and you never need to remember a deregistration step.

Dropping does not tell the server the session is over: it just goes quiet, and the server keeps the session buffering until its own timeout expires [docs/spec/02-session-lifecycle.md §6.3]. Call Client::disconnect when you want the server to know at once — for instance because the same user is limited to a fixed number of sessions.

Cloning

The client is not Clone, deliberately: its identity is what decides when the session ends, and a clone would make “dropping the client shuts the session down” untrue in a way that is hard to see at the call site. To use it from several tasks, put it in an std::sync::Arc — the session then ends when the last reference goes, which is the same rule stated once.

Implementations

impl Client

pub async fn connect(config: ClientConfig) -> Result<(Self, SessionEvents)>

Opens a session and waits until it is streaming.

Returns the client and the session’s event stream. The stream is handed over here rather than fetched later so that ignoring it is a decision you write down — drop(events) — instead of an omission that quietly costs you the recovery notifications (docs/adr/0005-recovery-is-visible-in-the-event-stream.md).

This resolves when the server has confirmed the session, so a Client you hold is one that was streaming at least once. It does not mean the connection is still up now: reconnection happens underneath, and the session event stream is where you learn about it.

Errors

• Error::Transport if the connection could not be established at all, or the address could not be turned into an endpoint.
• Error::Session if the server refused the session with a code that admits no retry — bad credentials (Appendix A code 1) and an unknown Adapter Set (code 2) are the usual ones.
• Error::ReconnectExhausted if every attempt allowed by RetryPolicy failed. Codes the specification marks temporary are retried before this is reported, and the last failure is carried in the error rather than reduced to a count.
• Error::Internal if this crate could not build a request it should always be able to build.

Examples

use lightstreamer_rs::{AdapterSet, Client, ClientConfig, ServerAddress};

let config = ClientConfig::builder(ServerAddress::try_new("https://push.lightstreamer.com")?)
    .with_adapter_set(AdapterSet::try_new("DEMO")?)
    .build()?;

let (client, events) = Client::connect(config).await?;

pub async fn subscribe(&self, subscription: Subscription) -> Result<Updates>

Subscribes to a set of items and returns the stream of what happens to them.

The subscription is requested immediately and re-created automatically on any session that replaces a lost one, so the stream you get back survives a reconnection without any action from you. Whether the server accepted it is reported on the stream, as SubscriptionEvent::Activated or SubscriptionEvent::Rejected — this call does not wait for that round trip, because a caller usually wants to subscribe to several things and then start reading.

Errors

• Error::Config if the subscription’s options and its mode cannot go together — see Subscription::validate, which this calls before anything is sent. Checking here rather than at encoding time is what makes an impossible subscription a mistake you see at the call site instead of a stream that quietly never activates.
• Error::Disconnected if the session has already stopped.
• Error::Internal if this client has run out of the identifiers it names subscriptions by — which takes 2⁶⁴ of them, and is reported rather than wrapped around because a reused identifier would silently conflate two subscriptions.

Everything the server has to say about the subscription arrives on the returned stream instead, where it can be interleaved correctly with the data.

Examples

use futures_util::StreamExt;
use lightstreamer_rs::{
    Client, FieldSchema, ItemGroup, Snapshot, Subscription, SubscriptionEvent,
    SubscriptionMode,
};

let mut updates = client
    .subscribe(
        Subscription::new(
            SubscriptionMode::Merge,
            ItemGroup::from_items(["item1", "item2"])?,
            FieldSchema::from_fields(["last_price", "time"])?,
        )
        .with_snapshot(Snapshot::On),
    )
    .await?;

while let Some(event) = updates.next().await {
    match event {
        SubscriptionEvent::Update(update) => {
            println!("{}: {:?}", update.item_name(), update.changed_fields());
        }
        SubscriptionEvent::Rejected(error) => {
            eprintln!("the server refused it: {error}");
            break;
        }
        _ => {}
    }
}

pub async fn unsubscribe(&self, id: SubscriptionId) -> Result<()>

Unsubscribes, and waits for the request to be handed to the session.

Usually unnecessary: dropping the Updates stream unsubscribes on its own. Use this when you want the unsubscription to be ordered with respect to your other calls, or when you want to keep reading the stream until SubscriptionEvent::Unsubscribed arrives — after which the server sends nothing more for those items [docs/spec/04-notifications.md §3.4].

Updates may still arrive between this call and that event; that is the protocol’s behaviour, not a race in this crate.

Errors

• Error::ForeignSubscription if the handle was created by a different client. Nothing is sent: every client numbers its subscriptions from one, so the same number names different data on each of them, and acting on it would cancel somebody else’s.
• Error::Disconnected if the session has already stopped, in which case the subscription is gone regardless.

pub async fn send_message(&self, message: Message) -> Result<()>

Sends a message to the server’s Metadata Adapter [docs/spec/03-requests.md §12].

Ok(()) means the message was queued for the session, not that it reached the server. What actually happened to it is reported on the session event stream as SessionEvent::Message — not here, because the round trip runs through the Metadata Adapter and may take as long as the application behind it does.

The guarantee is one report per message: every message that asked for an outcome produces exactly one SessionEvent::Message, whether it was processed, rejected by the adapter, refused by the server, or never sent at all. A Message::fire_and_forget one produces a report only in the last two cases, since it declined the notification the other two would have arrived on.

Note that the server acknowledges acceptance and processing separately: acceptance is internal to this crate, and MessageResult::Delivered is the one that means the Metadata Adapter actually ran [docs/spec/04-notifications.md §4.1].

Messages are never resent

If the session is replaced while a message is in flight, this crate reports MessageResult::NotSent rather than sending it again. The server deduplicates messages by their progressive number within a session, so replaying one across a replacement would restart the numbering it deduplicates against. Whether to retry is a decision only the caller can make correctly.

Errors

• Error::Config if the message’s parts cannot be sent together — see Message::validate, which this calls first.
• Error::Disconnected if the session has already stopped.

Examples

use std::num::NonZeroU32;
use lightstreamer_rs::{Client, Message, SequenceName};

client
    .send_message(
        Message::new("BUY 100", NonZeroU32::MIN)
            .in_sequence(SequenceName::try_new("ORDERS")?),
    )
    .await?;

pub async fn disconnect(self) -> Result<()>

Ends the session, telling the server so, and waits until it is over.

The server closes the session at once instead of holding it open until its own timeout [docs/spec/02-session-lifecycle.md §6.3]. When this returns, the destroy has been written, the socket has been closed, and every background task this client owns has finished — so a caller may shut its runtime down immediately afterwards without cutting the shutdown short. Every stream this client handed out has ended, the last session event being a SessionEvent::Closed.

Dropping the client instead ends the session without telling the server, and without waiting. Prefer this call when the session limit per user matters, or when you want a clean line in the server’s log.

The wait is bounded

A server that never answers, or a stream the caller stopped reading, cannot make this hang: after ConnectionOptions::with_open_timeout — the same budget one server round trip is given anywhere else — the shutdown stops being polite and the tasks are stopped regardless. The socket is closed on that path too.

Errors

Error::Disconnected if the session had already stopped — which is the outcome you asked for, so it is usually safe to ignore. Error::Internal if the session did not finish within the budget above; it has been stopped anyway, and nothing is left running.

Examples

// An already-stopped session is not a failure worth propagating here.
let _ = client.disconnect().await;

Trait Implementations

impl Debug for Client

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for Client

fn drop(&mut self)

Shuts the session down.

Closing the channels the background tasks watch is not enough on its own: a task blocked delivering into a stream the caller stopped reading is not watching anything. The explicit stop is what makes “dropping the client shuts the session down” true even then — it is exempt from backpressure by construction, and whatever was still undelivered goes with the streams it was going to.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.