An Elixir client for Lightstreamer servers, speaking TLCP 2.5.0 over WebSocket.
Unofficial and vendor-agnostic: this library implements the published TLCP protocol and nothing else. Adapter sets, item naming, field semantics and credential formats are defined by the server you connect to. "Lightstreamer" is a trademark of Lightstreamer Srl; this project is not affiliated with them.
Quick start
Stream live stock quotes from Lightstreamer's public demo server — paste
this into iex -S mix:
{:ok, session} =
Lightstreamer.connect("https://push.lightstreamer.com", adapter_set: "DEMO")
{:ok, sub} =
Lightstreamer.subscribe(session,
data_adapter: "QUOTE_ADAPTER",
items: ["item1", "item2", "item3"],
fields: ["stock_name", "last_price", "time"]
)
for _ <- 1..10 do
receive do
{:lightstreamer, ^sub, {:update, %Lightstreamer.Update{} = update}} ->
IO.inspect(update.values, label: update.item)
end
end
Lightstreamer.close(session)The first three updates are the snapshot (one full state per item); the rest are live changes as the simulated market moves.
A real consumer is usually a GenServer that owns its subscription and
handles the same messages in GenServer.handle_info/2 — updates are plain
messages, so there is no callback API to integrate with.
Delivery contract
Each subscription has an owner pid (default: the subscribe/2 caller),
which receives:
{:lightstreamer, sub, {:update, %Lightstreamer.Update{}}}— merged current field state plus the set of fields changed by this event{:lightstreamer, sub, {:end_of_snapshot, item}}{:lightstreamer, sub, {:overflow, item, lost_count}}— the server droppedlost_countupdates for the item (buffer limits){:lightstreamer, sub, {:unsubscribed, reason}}— the subscription is gone;reasonis:requestedafterunsubscribe/2,{:closed, reason}when the session ends,{:unsupported_diff, tag}if the server sent a diff format this library refuses (see below), or{:resubscribe_failed, %Lightstreamer.Error{}}when re-issuing after a reconnect was rejected
Session-level events go to the connect/2 caller (or the :events_to pid):
{:lightstreamer, session, {:disconnected, reason}}{:lightstreamer, session, {:reconnected, session_id}}{:lightstreamer, session, {:closed, reason}}
Field values are strings (or nil for wire-null), exactly as delivered —
typing and parsing are the consumer's job.
Connection resilience
By default (reconnect: :automatic) a session heals itself: server-driven
rebinds are transparent, and on connection loss it rebinds — or re-creates
the session and re-issues every subscription — with capped exponential
backoff. After a re-create, each item delivers a fresh snapshot
(snapshot?: true) before live updates resume, and a gap in updates is
possible across the reconnect. Watch the session-level events above if you
need to react to that. Bad credentials are never retried.
Architecture
A pure protocol codec (Lightstreamer.Protocol, Lightstreamer.Protocol.Frame,
Lightstreamer.Protocol.Update) with no processes or sockets, driven by a
Lightstreamer.Session state machine that owns one WebSocket via the
Lightstreamer.Transport behaviour (Lightstreamer.Transport.MintWs in
production). See the Internals module group.
The library starts no processes of its own and reads no application config —
every session is created by your code with per-call options, and can sit in
your supervision tree via child_spec/1:
children = [
{Lightstreamer, url: "https://push.lightstreamer.com", adapter_set: "DEMO"}
]
Summary
Functions
Child spec so {Lightstreamer, url: ..., adapter_set: ...} can sit directly
under a consumer's supervisor.
Closes the session and its transport. Subscription owners receive
{:lightstreamer, sub, {:unsubscribed, {:closed, :normal}}}.
Connects to a Lightstreamer server and creates a session.
Starts a session for a supervision tree, without blocking for CONOK.
Subscribes to items on the session. Blocks until the server confirms
(SUBOK) or rejects (REQERR) the subscription.
Unsubscribes. Blocks until the server confirms (UNSUB); the owner also
receives {:lightstreamer, sub, {:unsubscribed, :requested}}.
Functions
@spec child_spec(keyword()) :: Supervisor.child_spec()
Child spec so {Lightstreamer, url: ..., adapter_set: ...} can sit directly
under a consumer's supervisor.
@spec close(:gen_statem.server_ref()) :: :ok
Closes the session and its transport. Subscription owners receive
{:lightstreamer, sub, {:unsubscribed, {:closed, :normal}}}.
@spec connect(String.t(), opts) :: {:ok, pid()} | {:error, Lightstreamer.Error.t() | term()} when opts: [ adapter_set: String.t(), user: String.t(), password: String.t(), keepalive_millis: pos_integer(), keepalive_grace_millis: pos_integer(), inactivity_millis: pos_integer(), name: atom(), events_to: pid(), reconnect: :automatic | :never, connect_timeout: pos_integer(), transport: module(), transport_opts: keyword() ]
Connects to a Lightstreamer server and creates a session.
Blocks until the session is established (CONOK) or refused. The calling
process receives the session-level events documented in the moduledoc
unless :events_to directs them elsewhere.
Options
:adapter_set- adapter set to open the session on (server default:"DEFAULT"):user- user credential; semantics defined by the server's Metadata Adapter:password- password credential:keepalive_millis- requested keepalive interval; the server'sCONOKreply is authoritative:keepalive_grace_millis- extra silence tolerated beyond the keepalive interval before the connection is considered dead (default2000):inactivity_millis- advertises a maximum client-side silence to the server, committing the session to outbound heartbeats at that interval:name- locally registered name for the session process:events_to- pid receiving session-level events (default: the caller):reconnect-:automaticrecovers from connection loss (rebind first, then re-create with exponential backoff and resubscribe);:neverturns any failure into events and a stop. Default:automatic. Bad credentials (CONERRcode 1) are never retried.:connect_timeout- milliseconds to wait for session establishment (default15000):transport-Lightstreamer.Transportimplementation (defaultLightstreamer.Transport.MintWs):transport_opts- options passed through to the transport'sconnect/2
Examples
{:ok, session} =
Lightstreamer.connect("https://push.lightstreamer.com", adapter_set: "DEMO")
Starts a session for a supervision tree, without blocking for CONOK.
Accepts the same options as connect/2 plus :url. Prefer connect/2
when calling directly.
@spec subscribe(:gen_statem.server_ref(), opts) :: {:ok, reference()} | {:error, Lightstreamer.Error.t()} when opts: [ items: [String.t()], fields: [String.t()], mode: :merge, data_adapter: String.t(), snapshot: boolean(), max_frequency: :unlimited | :unfiltered | float(), owner: pid() ]
Subscribes to items on the session. Blocks until the server confirms
(SUBOK) or rejects (REQERR) the subscription.
Returns a subscription handle; the owner process then receives the subscription messages documented in the moduledoc, tagged with that handle.
Options
:items- required; item names (theLS_group):fields- required; field names (theLS_schema):mode- only:mergein v0.1 (default:merge):data_adapter- data adapter within the adapter set (server default:"DEFAULT"):snapshot- request initial full state per item (defaulttrue):max_frequency-:unlimited,:unfiltered, or max updates per second as a float (server default::unlimited):owner- pid receiving this subscription's updates (default: the caller). The session monitors it and unsubscribes if it dies.
Examples
{:ok, sub} =
Lightstreamer.subscribe(session,
data_adapter: "QUOTE_ADAPTER",
items: ["item1", "item2"],
fields: ["stock_name", "last_price"]
)
@spec unsubscribe(:gen_statem.server_ref(), reference()) :: :ok | {:error, Lightstreamer.Error.t()}
Unsubscribes. Blocks until the server confirms (UNSUB); the owner also
receives {:lightstreamer, sub, {:unsubscribed, :requested}}.