FixAlchemy.Engine (FIXAlchemy v0.2.1)

The FIX session protocol for one connection, independent of which side dialed.

An engine owns everything a session does once bytes are flowing: framing the stream, sequence numbers in both directions, Heartbeats and TestRequests, ResendRequests and replay, inbound validation, and delivery to subscribed processes. It does not open, accept, or re-open sockets — a session process does that and hands the engine a connected socket with attach/3.

FixAlchemy.Client (initiator) and FixAlchemy.Server.Session (acceptor) both drive an engine. The engine does not act on a Logon or a Logout; it reports each as an action for the session process to handle.

Driving an engine

A session process calls receive_data/2 for every packet, check_liveness/1 on every :liveness_tick, and send_message/3 to send. Each returns the new engine; receive_data/2 and check_liveness/1 also return actions:

{:logon, raw} - a Logon arrived
{:logout, raw} - a Logout arrived
{:disconnect, reason} - the session must end; the engine has already sent whatever the protocol requires first

The engine sets timers in and delivers messages from the calling process, so every function must be called from the process that owns the socket. It sends itself :liveness_tick, which the session process must route to check_liveness/1.

Sequence recovery

With sequence_recovery: false (the default) inbound sequence numbers are not checked and sent messages are not stored, which costs nothing per message. With it on, the engine holds messages arriving ahead of a gap, requests a resend, and answers inbound ResendRequests from its FixAlchemy.MessageStore.

Validation

With validate_inbound: true every inbound message is checked by FixAlchemy.Engine.Validation and a failing one is answered with a Reject rather than processed. An acceptor should turn this on; an initiator talking to a venue it trusts need not.

Summary

Types

action()

t()

Functions

activate_once(engine)

Ask the socket for its next packet.

attach(engine, transport, socket)

Give the engine a connected socket to run the session on.

cancel_liveness(engine)

cancel_logon_send(engine, key)

check_liveness(engine)

Send a Heartbeat or TestRequest if the interval has elapsed, and report a connection that has stopped answering.

close(engine)

Close the socket if one is attached.

detach(engine)

Forget the socket, leaving sequence numbers and subscriptions intact.

expect_inbound(engine, seq)

Set the sequence number the next inbound message must carry.

flush_logon_sends(engine)

Send every message registered with register_logon_send/3.

logged_in?(engine)

mark_logged_in(engine)

mark_logged_out(engine)

milestone_reached?(engine, milestone)

new(opts)

Build an engine with no socket attached.

next_expected_inbound(engine)

peek_outbound(engine)

reach_milestone(engine, milestone)

Record that a point in the session's login flow has been reached, sending anything queued for it with send_when/3.

receive_data(engine, data)

Feed bytes off the socket through the session protocol.

register_logon_send(engine, key, arg)

rejected_message_type?(engine, msg_type)

Whether the counterparty rejected this message type as unsupported.

release(engine)

Flush and release the session store, if one is open.

reset_sequence(engine)

Reset both sequence directions to 1 and discard stored messages.

resume(engine)

Open the session store and resume the numbering it holds.

schedule_liveness(engine)

Arm the next :liveness_tick, replacing any timer already set.

send_message(engine, msg_type, fields)

Send a message, assigning it the next outbound sequence number.

send_when(engine, milestone, arg)

Hold a message until milestone is reached.

subscribe(engine, pid, type, value \\ :all, opts \\ [])

Subscribe pid to a message type; see FixAlchemy.Dispatch.subscribe/5.

subscriber_down(engine, ref, pid)

Drop every subscription held by a subscriber that has exited.

unsubscribe(engine, pid)

Types

action()

@type action() :: {:logon, binary()} | {:logout, binary()} | {:disconnect, term()}

t()

@type t() :: %FixAlchemy.Engine{
  buffer: term(),
  connection_id: term(),
  dispatch: term(),
  fix_version: term(),
  heartbeat_interval: term(),
  last_received: term(),
  last_sent: term(),
  liveness_timer: term(),
  logged_in: term(),
  logon_sends: term(),
  max_latency: term(),
  message_store: term(),
  milestones: term(),
  monitors: term(),
  pending_sends: term(),
  process_name: term(),
  rejected_message_types: term(),
  sender_comp_id: term(),
  sender_sub_id: term(),
  sequence: term(),
  sequence_recovery: term(),
  session_name: term(),
  session_store: term(),
  session_store_handle: term(),
  session_store_opts: term(),
  socket: term(),
  spec_modules: term(),
  spec_name: term(),
  store: term(),
  target_comp_id: term(),
  target_sub_id: term(),
  test_request_pending: term(),
  transport: term(),
  validate_checksum: term(),
  validate_inbound: term()
}

Functions

activate_once(engine)

@spec activate_once(t()) :: t()

Ask the socket for its next packet.

attach(engine, transport, socket)

@spec attach(t(), FixAlchemy.Transport.t(), FixAlchemy.Transport.socket()) :: t()

Give the engine a connected socket to run the session on.

Clears the logged-in flag and every milestone. Sequence numbers, subscriptions and stored messages are kept.

cancel_liveness(engine)

@spec cancel_liveness(t()) :: t()

cancel_logon_send(engine, key)

@spec cancel_logon_send(t(), term()) :: t()

check_liveness(engine)

@spec check_liveness(t()) :: {t(), [action()]}

Send a Heartbeat or TestRequest if the interval has elapsed, and report a connection that has stopped answering.

close(engine)

@spec close(t()) :: t()

Close the socket if one is attached.

detach(engine)

@spec detach(t()) :: t()

Forget the socket, leaving sequence numbers and subscriptions intact.

expect_inbound(engine, seq)

@spec expect_inbound(t(), pos_integer()) :: t()

Set the sequence number the next inbound message must carry.

flush_logon_sends(engine)

@spec flush_logon_sends(t()) :: t()

Send every message registered with register_logon_send/3.

logged_in?(engine)

@spec logged_in?(t()) :: boolean()

mark_logged_in(engine)

@spec mark_logged_in(t()) :: t()

mark_logged_out(engine)

@spec mark_logged_out(t()) :: t()

milestone_reached?(engine, milestone)

@spec milestone_reached?(t(), term()) :: boolean()

new(opts)

@spec new(keyword()) :: t()

Build an engine with no socket attached.

Options

:sender_comp_id, :target_comp_id - this session's identity (required)
:sender_sub_id, :target_sub_id - SubIDs (50, 57), sent when set
:spec_modules - the generated dictionary, as FixAlchemy.Parser.init/1 returns it (required)
:connection_id, :session_name - naming for status, logs and delivery
:fix_version - BeginString, default "FIX.4.4"
:heartbeat_interval - seconds, default 30
:sequence_recovery - hold gaps and answer resends, default false
:validate_inbound - check every inbound message, default false
:validate_checksum - drop messages whose checksum is wrong, default false
:max_latency - SendingTime tolerance in seconds, default 120
:message_store, :message_store_opts - resend storage
:routing_keys - per-message-type routing tag overrides for delivery

next_expected_inbound(engine)

@spec next_expected_inbound(t()) :: pos_integer()

peek_outbound(engine)

@spec peek_outbound(t()) :: pos_integer()

reach_milestone(engine, milestone)

@spec reach_milestone(t(), term()) :: t()

Record that a point in the session's login flow has been reached, sending anything queued for it with send_when/3.

Reaching :ready also sends every message held by register_logon_send/3, on each logon rather than only the first.

receive_data(engine, data)

@spec receive_data(t(), binary()) :: {t(), [action()]}

Feed bytes off the socket through the session protocol.

Returns the new engine and the actions its session process must decide on.

register_logon_send(engine, key, arg)

@spec register_logon_send(t(), term(), {binary(), list()}) :: t()

Sent at once when the session is already ready, and on reaching :ready after every later reconnect.

rejected_message_type?(engine, msg_type)

@spec rejected_message_type?(t(), binary()) :: boolean()

Whether the counterparty rejected this message type as unsupported.

release(engine)

@spec release(t()) :: t()

Flush and release the session store, if one is open.

reset_sequence(engine)

@spec reset_sequence(t()) :: t()

Reset both sequence directions to 1 and discard stored messages.

resume(engine)

@spec resume(t()) :: t()

Open the session store and resume the numbering it holds.

Called once the session's identity is known. Without a :session_store the engine keeps the numbers it already has.

schedule_liveness(engine)

@spec schedule_liveness(t()) :: t()

Arm the next :liveness_tick, replacing any timer already set.

send_message(engine, msg_type, fields)

@spec send_message(t(), binary(), [{pos_integer() | atom(), term()}]) :: t()

Send a message, assigning it the next outbound sequence number.

fields are {tag, value} pairs; a tag may be the integer tag or the field's atom name, which is resolved through the dictionary. Fields with a nil or empty value are dropped. The standard header is built here.

send_when(engine, milestone, arg)

@spec send_when(t(), term(), {binary(), list()}) :: t()

Hold a message until milestone is reached.

subscriber_down(engine, ref, pid)

@spec subscriber_down(t(), reference(), pid()) :: t()

Drop every subscription held by a subscriber that has exited.

unsubscribe(engine, pid)

@spec unsubscribe(t(), pid()) :: t()