Class BaseMessageQueue<TMessage, TItem>

Namespace

ATAS.DataFeedsCore

Assembly

ATAS.DataFeedsCore.dll

A queue processed by its own thread.
The queue runs in sessions: Start(IDataFeedConnector, Action<TMessage>) begins one (the first connector of a shared queue), Stop(IDataFeedConnector) ends it (the last connector). An item belongs to the session it was queued in: the items still queued when the session ends are dropped - also when the next session has already begun (a connect issued right after a disconnect) and the thread of the stopped session goes on to serve it. A message or an action queued while no session runs waits for the next session (unless a thread of the stopped queue is still alive and drops it); a continuation (EnqueueContinuation(IDataFeedConnector, Action)) queued after a session ended is dropped, it never runs in a later session.

public abstract class BaseMessageQueue<TMessage, TItem> : BaseLoggerSource, ILoggerSource

Type Parameters

TMessage

TItem

Inheritance

object

BaseLoggerSource

BaseMessageQueue<TMessage, TItem>

Implements

ILoggerSource

Derived

SimpleMessageQueue<TMessage>

Extension Methods

Extensions.ConvertToMyTrade<T>(T, Portfolio, Security, Action<MyTrade, T>)

LoggerHelper.DisposeLogger(object)

LoggerHelper.LogDebug(object, string, params object?[])

LoggerHelper.LogError(object, string, Exception, bool)

LoggerHelper.LogError(object, string, params object?[])

LoggerHelper.LogFatal(object, string, Exception, bool)

LoggerHelper.LogInfo(object, string, params object?[])

LoggerHelper.LogWarn(object, string, params object?[])

Constructors

BaseMessageQueue()

Initializes the queue with the logger name "Queue" and the logging level inherited from its parent.

protected BaseMessageQueue()

Properties

HeartbeatTimeout

Interval of the Heartbeat event; System.TimeSpan.Zero disables it. It also bounds how long the idle thread waits for an item, so the heartbeat is raised when no items arrive.

public TimeSpan HeartbeatTimeout { get; set; }

Property Value

TimeSpan

Methods

CreateItem(IDataFeedConnector, TMessage?, Action?)

Packs a message or an action of a connector into a queue item; exactly one of message and action is set.

Added in ATAS 8.0.15.

protected abstract TItem CreateItem(IDataFeedConnector connector, TMessage? message, Action? action)

Parameters

connector IDataFeedConnector

The connector the item belongs to.

message TMessage

The message, or null for an action.

action Action

The action, or null for a message.

Returns

TItem

The queue item.

DescribeItem(TItem)

Short representation of a queued item for debug logging. Full message dumps are logged by connectors on receive; dumping them again here would double the cost.

Added in ATAS 8.0.15.

protected virtual object? DescribeItem(TItem item)

Parameters

item TItem

Returns

object

Enqueue(IDataFeedConnector, Action)

Queues an action to run on the connector's thread.

public void Enqueue(IDataFeedConnector connector, Action action)

Parameters

connector IDataFeedConnector

The connector the action belongs to.

action Action

The action.

Exceptions

ArgumentNullException

connector or action is null.

Enqueue(IDataFeedConnector, TMessage)

Queues a message of the connector for its handler.

public void Enqueue(IDataFeedConnector connector, TMessage message)

Parameters

connector IDataFeedConnector

The connector the message belongs to.

message TMessage

The message.

Exceptions

ArgumentNullException

connector or message is null.

EnqueueContinuation(IDataFeedConnector, Action)

Not in ATAS 8.0.15 yet: available from the next version.

public void EnqueueContinuation(IDataFeedConnector connector, Action continuation)

Parameters

connector IDataFeedConnector

continuation Action

Exceptions

ArgumentNullException

connector or continuation is null.

OnProcess(TItem)

Processes an item of the running session on the process thread: passes the message to the handler of its connector or runs the action. An exception is logged and processing goes on.

protected abstract void OnProcess(TItem item)

Parameters

item TItem

The item.

OnStart(IDataFeedConnector, Action<TMessage>)

Registers the handler of a connector; called by Start(IDataFeedConnector, Action<TMessage>) under the queue's lifetime lock.

protected abstract void OnStart(IDataFeedConnector connector, Action<TMessage> handler)

Parameters

connector IDataFeedConnector

The connector.

handler Action<TMessage>

The connector's message handler.

OnStop(IDataFeedConnector)

Unregisters a connector; called by Stop(IDataFeedConnector) under the queue's lifetime lock.

protected abstract bool OnStop(IDataFeedConnector connector)

Parameters

connector IDataFeedConnector

The connector.

Returns

bool

true when no connector is left and the session may end.

Start(IDataFeedConnector, Action<TMessage>)

Registers the connector's handler (OnStart(IDataFeedConnector, Action<TMessage>)) and begins a session if none runs; starts the process thread unless the thread of the stopped session is still alive, which then serves the new session.

public void Start(IDataFeedConnector connector, Action<TMessage> handler)

Parameters

connector IDataFeedConnector

The connector.

handler Action<TMessage>

Processes a message of the connector on the process thread.

Exceptions

ArgumentNullException

connector or handler is null.

Stop(IDataFeedConnector)

Unregisters the connector (OnStop(IDataFeedConnector)) and, when that was the last one, ends the session: the items still queued are dropped and the process thread exits once the queue is empty.

public void Stop(IDataFeedConnector connector)

Parameters

connector IDataFeedConnector

The connector.

Exceptions

ArgumentNullException

connector is null.

Events

Heartbeat

Raised on the process thread every HeartbeatTimeout while a session runs; the first interval of the thread only starts the count and raises nothing.

Added in ATAS 8.0.15.

public event Action? Heartbeat

Event Type

Action