Lightstreamer (Lightstreamer v0.1.0)

Copy Markdown View Source

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 dropped lost_count updates for the item (buffer limits)
  • {:lightstreamer, sub, {:unsubscribed, reason}} — the subscription is gone; reason is :requested after unsubscribe/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

child_spec(opts)

@spec child_spec(keyword()) :: Supervisor.child_spec()

Child spec so {Lightstreamer, url: ..., adapter_set: ...} can sit directly under a consumer's supervisor.

close(session)

@spec close(:gen_statem.server_ref()) :: :ok

Closes the session and its transport. Subscription owners receive {:lightstreamer, sub, {:unsubscribed, {:closed, :normal}}}.

connect(url, opts \\ [])

@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's CONOK reply is authoritative
  • :keepalive_grace_millis - extra silence tolerated beyond the keepalive interval before the connection is considered dead (default 2000)
  • :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 - :automatic recovers from connection loss (rebind first, then re-create with exponential backoff and resubscribe); :never turns any failure into events and a stop. Default :automatic. Bad credentials (CONERR code 1) are never retried.
  • :connect_timeout - milliseconds to wait for session establishment (default 15000)
  • :transport - Lightstreamer.Transport implementation (default Lightstreamer.Transport.MintWs)
  • :transport_opts - options passed through to the transport's connect/2

Examples

{:ok, session} =
  Lightstreamer.connect("https://push.lightstreamer.com", adapter_set: "DEMO")

start_link(opts)

@spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}

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.

subscribe(session, opts)

@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 (the LS_group)
  • :fields - required; field names (the LS_schema)
  • :mode - only :merge in v0.1 (default :merge)
  • :data_adapter - data adapter within the adapter set (server default: "DEFAULT")
  • :snapshot - request initial full state per item (default true)
  • :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"]
  )

unsubscribe(session, sub)

@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}}.