FixAlchemy.Client (FIXAlchemy v0.3.0)

View Source

GenServer-based FIX protocol session client — the initiator side.

Dials a venue, sends the Logon, and reconnects with backoff when the connection drops. Everything the session protocol does once bytes are flowing — framing, sequence numbers, heartbeats, resends, delivery to subscribers — belongs to FixAlchemy.Engine, which this client owns and drives. FixAlchemy.Server.Session drives the same engine from the acceptor side.

The engine handles the universal session protocol itself — heartbeats, TestRequest, reject, resend/sequence-reset — and reports the Logon, the Logout, and any fatal condition back here. Every other inbound message is routed to subscribers through a keyed pub/sub bus (FixAlchemy.Dispatch): a process subscribes to a message type — for all messages of that type, for those where a tag equals a value, or those where a tag satisfies a predicate — and receives {:fix, type, raw, meta} for each match. Field decoding happens in the subscriber, at the point of use, off the socket loop; meta.fields carries the one framing-level field split the bus already performed for routing.

Start sessions through FixAlchemy.SessionSupervisor, which supervises this client alongside the subscriber processes an adapter supplies.

Examples

{:ok, _sup} = FixAlchemy.SessionSupervisor.start_link(
  connection_id: "demo",
  host: "fix.example.com",
  port: 9043,
  username: "demo_user",
  sender_comp_id: "CLIENT1",
  target_comp_id: "SERVER"
)

[{client, _}] = Registry.lookup(FixAlchemy.Registry, {"demo", :trading})
FixAlchemy.Client.subscribe(client, self(), "W", "EUR/USD")

Summary

Types

t()

Functions

cancel_logon_send(conn, key)

Stop re-sending the logon message registered under key.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_capabilities(conn)
milestone_reached?(conn, milestone)

Whether the session has reached milestone.

reach(conn, milestone)

Mark that the session has reached milestone, flushing anything waiting on it.

ready(conn)

Mark the session ready for business messages; equivalent to reach(conn, :ready).

register_logon_send(conn, key, arg)

Register a message, keyed by key, that the engine sends each time the session becomes ready.

send_custom_message(conn, arg)

Send a FIX message built from {msg_type, fields}.

send_when(conn, milestone, arg)

Send message once the session reaches milestone (immediately if already reached).

session_key(conn)
start_link(opts)

Start a FIX client session.

stop(connection_id)

Stop a FIX session, including its subscriber processes.

subscribe(conn, subscriber, type, value \\ :all, opts \\ [])

Subscribe subscriber to a message type on this session.

unsubscribe(conn, subscriber)

Remove every subscription held by subscriber on this session.

Types

t()

@type t() :: %FixAlchemy.Client{
  connect_timeout: term(),
  connection_id: term(),
  default_appl_ver_id: term(),
  defer_ready: term(),
  engine: term(),
  host: term(),
  password: term(),
  port: term(),
  reconnect_attempts: term(),
  reset_seq_number: term(),
  session_name: term(),
  tls: term(),
  tls_opts: term(),
  user_on_login: term(),
  username: term()
}

Functions

cancel_logon_send(conn, key)

@spec cancel_logon_send(GenServer.server(), term()) :: :ok

Stop re-sending the logon message registered under key.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_capabilities(conn)

@spec get_capabilities(GenServer.server()) :: FixAlchemy.Capabilities.t()

milestone_reached?(conn, milestone)

@spec milestone_reached?(GenServer.server(), atom()) :: boolean()

Whether the session has reached milestone.

reach(conn, milestone)

@spec reach(GenServer.server(), atom()) :: :ok

Mark that the session has reached milestone, flushing anything waiting on it.

Milestones name points in a broker's login flow — :ready on FIX logon by default, plus adapter-defined ones such as :account. Reaching :ready flushes register_logon_send/3 messages; reaching any milestone flushes the send_when/3 messages queued for it.

ready(conn)

@spec ready(GenServer.server()) :: :ok

Mark the session ready for business messages; equivalent to reach(conn, :ready).

register_logon_send(conn, key, arg)

@spec register_logon_send(GenServer.server(), term(), {binary(), list()}) :: :ok

Register a message, keyed by key, that the engine sends each time the session becomes ready.

Never sent before the session is ready (see ready/1), and re-sent after a reconnect. Re-registering the same key replaces the message; remove it with cancel_logon_send/2.

send_custom_message(conn, arg)

@spec send_custom_message(
  GenServer.server(),
  {binary(), list()}
) :: :ok

Send a FIX message built from {msg_type, fields}.

Examples

FixAlchemy.Client.send_custom_message(conn, {"D", [{55, "EUR/USD"}, {54, "1"}]})

send_when(conn, milestone, arg)

@spec send_when(GenServer.server(), atom(), {binary(), list()}) :: :ok

Send message once the session reaches milestone (immediately if already reached).

session_key(conn)

@spec session_key(GenServer.server()) :: {binary(), FixAlchemy.SessionConfig.name()}

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Start a FIX client session.

Options

  • :connection_id - Unique connection identifier (required)
  • :host - Server hostname (required)
  • :port - Server port (required)
  • :username - FIX Username (553), optional; sent only when set
  • :sender_comp_id - Sender component ID (required)
  • :target_comp_id - Target component ID (required)
  • :password - FIX password (optional)
  • :sender_sub_id - SenderSubID (50), sent when set
  • :target_sub_id - TargetSubID (57), sent when set
  • :fix_version - BeginString; defaults to the one the dictionary declares
  • :heartbeat_interval - Heartbeat interval in seconds (default: 30)
  • :reset_seq_number - Reset sequence on login (default: "Y")
  • :user_on_login - Send username/password in logon message (default: false)
  • :validate_checksum - Drop inbound messages with bad checksums (default: false)
  • :validate_inbound - Check every inbound message against the dictionary and Reject a bad one (default: false)
  • :sequence_recovery - Track inbound MsgSeqNum, request resends on gaps, answer inbound ResendRequests, and honour SequenceReset (default: false)
  • :message_store - FixAlchemy.MessageStore implementation used to replay sent messages (default: FixAlchemy.MessageStore.Memory)
  • :message_store_opts - Options passed to the store's new/1
  • :routing_keys - Map of message type to routing tag, merged over the dispatch defaults (default: %{})
  • :tls - Connect over TLS (default: false)
  • :tls_opts - Extra :ssl.tls_client_option() entries merged over the defaults (verify_peer against the OS trust store, SNI from :host)
  • :process_name - Name used as Logger source metadata (default: "")

stop(connection_id)

@spec stop(GenServer.server() | binary()) :: :ok

Stop a FIX session, including its subscriber processes.

Sends a Logout (35=5) before closing when the session is logged in.

subscribe(conn, subscriber, type, value \\ :all, opts \\ [])

@spec subscribe(GenServer.server(), pid(), binary(), binary() | :all, keyword()) ::
  :ok

Subscribe subscriber to a message type on this session.

  • subscribe(conn, pid, type) — every message of that type
  • subscribe(conn, pid, type, value) — where the routing tag equals value
  • subscribe(conn, pid, type, :all, tag: t, where: fun) — where fun.(value)

The subscriber receives {:fix, type, raw, meta} for each match, where meta is %{connection_id, session_name, client, fields}. Subscriptions are pruned automatically when the subscriber exits.

unsubscribe(conn, subscriber)

@spec unsubscribe(GenServer.server(), pid()) :: :ok

Remove every subscription held by subscriber on this session.