Crate lightstreamer_rs

Expand description

A Rust client for TLCP, the Text Lightstreamer Client Protocol.

Lightstreamer is a server that pushes real-time data to clients over an ordinary web connection. TLCP is the text protocol it speaks. This crate implements TLCP 2.5.0 and nothing else: it opens a session, keeps it alive through the network’s bad behaviour, delivers updates as a typed stream, and gets out of the way. There is no market-data model here, no cache, no reconnect-and-hope layer — just the protocol, done properly.

§A complete program

This connects to the public demo server the specification’s own transcripts use, subscribes to two items, and prints what changes.

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

#[tokio::main]
async fn main() -> lightstreamer_rs::Result<()> {
    let config = ClientConfig::builder(ServerAddress::try_new("https://push.lightstreamer.com")?)
        .with_adapter_set(AdapterSet::try_new("DEMO")?)
        .build()?;

    // This example is only interested in the data. Opting out of the
    // session events is a `drop`, not an underscore: naming a receiver
    // `_session_events` would *keep* it, and a stream that is held but
    // never read stalls the client once it fills. See `SessionEvents`.
    let (client, session_events) = Client::connect(config).await?;
    drop(session_events);

    let mut updates = client
        .subscribe(
            Subscription::new(
                SubscriptionMode::Merge,
                ItemGroup::from_items(["item1", "item2"])?,
                FieldSchema::from_fields(["last_price", "time", "pct_change"])?,
            )
            .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 the subscription: {error}");
                break;
            }
            _ => {}
        }
    }

    client.disconnect().await
}

§The three things worth knowing before you start

Delivery is a Stream, not a callback. Subscribing gives you an Updates stream; the client gives you a SessionEvents stream. Both are ordinary futures_util::Streams carrying closed, matchable enums, so the borrow checker never lands in the middle of your update handler (docs/adr/0003-typed-event-stream-as-delivery-surface.md).

Reconnection is automatic; its consequences are not hidden. When the connection breaks, this crate reconnects. What it will not do is pretend nothing happened: SessionEvent::Connected carries a Continuity that says whether the same session came back — in which case whatever you computed from it is still valid — or whether a new one replaced it, in which case it is not (docs/adr/0005-recovery-is-visible-in-the-event-stream.md). An application holding an order book or a portfolio needs that distinction, and TLCP is one of the few protocols that actually provides it.

Streams are bounded and lossless. Every stream this crate hands you has a fixed capacity and, when it fills, the client blocks rather than discarding anything — so a consumer that stops reading stalls the client instead of silently losing data. The full reasoning, and what to do if you genuinely cannot keep up, is on Updates.

§Two things that catch people out

Null and empty are different values. TLCP distinguishes a field with no value from a field holding the empty string, and so does FieldValue: Null and Text("") are not equal and do not mean the same thing. A quote adapter with no bid to report sends the first; one whose status line is deliberately blank sends the second. Collapsing them is a decision you make explicitly, with FieldValue::text or FieldValue::text_or — this crate will not make it for you. Note also the two levels of absence: ItemUpdate::field returns None when the schema has no such field at all, and Some(FieldValue::Null) when the field exists and is null.

Item and field names come from you, not from the server. TLCP transmits counts, never names [docs/spec/04-notifications.md §3.1]. So ItemUpdate::item_name can only report a name you supplied — which you do by spelling out the list with ItemGroup::from_items and FieldSchema::from_fields. If you subscribe by naming a server-side group with ItemGroup::try_new instead, the server resolves it and this crate never learns what the items are called, so names fall back to the 1-based position rendered in decimal. That stand-in is deliberately not a lookup key: ItemUpdate::field_by_name with "3" returns None rather than the third field, so a placeholder can never be mistaken for a real name. Use ItemUpdate::declared_item_name when you need to know which you got.

§Shutdown

Dropping the Client stops everything: the background tasks, the socket, and every stream still outstanding. Dropping an Updates stream unsubscribes. Nothing needs to be deregistered and nothing needs std::process::exit.

§What is deliberately not here

The transports, the wire parser and the session state machine are private. They are the crate’s job, not yours, and keeping them private is what lets them change without breaking you. What lib.rs re-exports is the whole public surface and the whole semantic-versioning promise.

§Testing your own code against this crate

An ItemUpdate cannot be constructed from outside, because forging one that a session never produced is not something an application should be able to do by accident. Your own parsing layer still deserves unit tests, so the off-by-default test-util feature adds one thing and nothing else: test_util::ItemUpdateBuilder, which assembles an update with no session behind it. Enable it under [dev-dependencies] and it never reaches your release build.

§Errors

Everything fallible returns Result, and Error has one variant per condition worth reacting to. Codes the server supplied are preserved as numbers in ServerError, never flattened into a message — and which of the protocol’s two code catalogs a number belongs to is decided by the variant carrying it, because the two overlap. See error.

§Provenance

Version 1.0.0 onwards is an independently authored implementation licensed under MIT. Versions up to and including 0.3.3 were GPL-3.0-only and share no code with this one; they remain published, under that licence, as historical versions.

Every wire behaviour in this crate is derived from the official specification and cites the chapter of docs/spec/ it comes from. See docs/adr/0001-mit-relicensing-clean-room.md.

Re-exports§

pub use config::AdapterSet;
pub use config::ClientConfig;
pub use config::ClientConfigBuilder;
pub use config::ConfigError;
pub use config::ConnectionOptions;
pub use config::Credentials;
pub use config::MAX_TIMING;
pub use config::RetryPolicy;
pub use config::ServerAddress;
pub use config::Transport;
pub use error::Error;
pub use error::Result;
pub use error::ServerError;
pub use error::TransportError;

Modules§

config: Everything a client needs to be told before it can open a session.
error: The public error taxonomy.

Structs§

ChangedFields: Iterator over the fields an ItemUpdate changed; see ItemUpdate::changed_fields.
Client: A live Lightstreamer session.
CommandFields: Where the key and the command sit in a COMMAND-mode field schema [docs/spec/04-notifications.md §3.2].
Connected: The session is bound to a server and streaming.
Field: One field of an ItemUpdate: where it sits, what it is called, what it holds, and whether this update changed it.
FieldSchema: Which fields a subscription asks for.
Fields: Iterator over every field of an ItemUpdate; see ItemUpdate::fields.
FrequencyLimit: A checked decimal frequency, in updates per second.
ItemGroup: Which items a subscription covers.
ItemUpdate: One real-time update, decoded: the new state of one item of one subscription, plus which of its fields this update actually changed.
Message: A message on its way to the server’s Metadata Adapter.
MessageOutcome: The report of one message’s fate.
Recovery: Where a recovered flow actually resumed, relative to where it was asked to.
Resubscribed: One subscription re-created on a session that replaced a lost one.
SequenceName: The name of an ordered message sequence [docs/spec/03-requests.md §12.1].
SessionEvents: The stream of session-level events.
Subscription: A request to subscribe to a set of items.
SubscriptionId: An opaque handle identifying one subscription for as long as you hold it.
Updates: The stream of one subscription’s events.

Enums§

BufferSize: How much the server may buffer per item before it has to drop or merge [docs/spec/03-requests.md §6.1].
ClosedReason: Why the session ended for good.
Continuity: What a newly bound session means for state you built on the previous one.
DisconnectReason: Why the session stopped streaming.
FieldValue: The value of one field of one item.
Filtering: Whether the server may drop or merge updates for a subscription to keep up [docs/spec/04-notifications.md §3.8].
ItemCommand: What a COMMAND-mode update did to the row named by its key.
MaxFrequency: A ceiling on how often the server may send updates for an item [docs/spec/03-requests.md §6.1].
MessageResult: What became of a message you sent.
RecoveryOutcome: The three ways a recovery can turn out.
ServerInfo: Something the server said about itself or about the session, which changes nothing but may be worth logging.
SessionEvent: Everything the client tells you about the session.
Snapshot: Whether the server should send the current state before the live flow [docs/spec/03-requests.md §6.1].
StateValidity: What a bind means for state an application derived from an earlier one.
SubscriptionEvent: Everything one subscription can tell you.
SubscriptionMode: How the server treats the items of a subscription [docs/spec/03-requests.md §6.1].
UpdateFrequency: The update rate the server has settled on for a subscription [docs/spec/04-notifications.md §3.8].
UpdateKind: Whether an update carries part of an item’s initial state or a live change.