Reading Messages

The reader is the low-level, fundamental data retrieval mechanism. It reads messages in order from a single message stream or category.

A reader reads raw message data. It doesn't convert message data into typed messages.

Note: The reader is rarely directly used in-practice except for certain circumstances where it might be leveraged in a utility script. In most cases, a reader is used indirectly by retrieving an entity from a store, or when a consumer retrieves and dispatches messages.

Example

account_id = '123'

deposit = Deposit.new()
deposit.deposit_id = '456'
deposit.account_id = account_id
deposit.amount = 11
deposit.time = '2000-01-01T11:11:11.000Z'

command_stream_name = "account:command-#{account_id}"

Messaging::Postgres::Write.(deposit, command_stream_name)

MessageStore::Postgres::Read.(command_stream_name) do |message_data|
  pp message_data.data
end

# => {:account_id=>"123", :deposit_id=>"456", :amount=>11, :time=>"2000-01-01T11:11:11.000Z"}

Reader Facts

• Each message data is passed to the reader's block
• Readers don't return anything
• Readers retrieve messages in batches
• A reader reads events in order, and continues to read until the end of the stream
• A reader's batch size is configurable
• A reader's starting position is configurable
• A reader can be configured with a condition that filters the messages retrieved based on a SQL condition
• A reader can be configured with an existing session, or it can create a new session

MessageStore::Postgres::Read Class

The Read class is a concrete class from the MessageStore::Postgres library and namespace.

The Read class provides:

• The principle instance actuator .() (or the call instance method) for starting a reader
• The class actuator .() (or the class call method) that provides a convenient invocation shortcut that does not require instantiating the reader class first

Reading a Stream

A reader can be actuated either via its class interface, as a matter of convenience, or via its instance interface, which allows for greater control of the configuration of the reader.

Readers are implemented as callable objects. Actuating them is simply a matter of invoking their call method.

Class Actuator

self.call(stream_name, position: 0, batch_size: 1000, condition: nil, session: nil, &action)

Parameters

Name	Description	Type
stream_name	Name of the stream that the reader will read	String
position	Position of the message to start reading from	Integer
batch_size	Number of messages to retrieve with each query to the message store	Integer
condition	SQL condition to filter the batch by	String
session	An existing session object to use, rather than allowing the reader to create a new session	MessageStore::Postgres::Session
action	Block to be evaluated for each message read	Callable

Instance Actuator

call(&action)

Parameters

Name	Description	Type
action	Block to be evaluated for each message read	Callable

Read Loop

One Message per Loop

The action passed to the reader's actuator is processed once per message read, irrespective of batch size.

MessageStore::Postgres::Read.(stream_name) do |message_data|
  # Block is evaluated once per message data read
end

Reading in Batches

The reader retrieves messages in batches. The number of messages retrieved in each batch are specified using the reader actuator's batch_size parameter. The default batch size is 1000 messages.

MessageStore::Postgres::Read.(stream_name, batch_size: 10) do |message_data|
  # ...
end

The action passed to the reader's actuator is evaluated once per message read. However, the reader doesn't query messages from the message store every time the block is evaluated.

A reader will retrieve a batch of messages, and then process each of those messages in-sequence by passing each message to the action block.

When the reader has completed the processing of a batch of messages, the reader retrieves another batch of messages. When there are no more messages to be processed, ie: when a retrieval of a batch yields no new results, the reader terminates.

Terminating a Reader

A reader will terminate under two conditions:

1. There are no more messages to be retrieved from the message store
2. A break statement is issued from within the action block passed to the reader

MessageStore::Postgres::Read.(stream_name) do |message_data|
  break # The reading will stop here
end

Filtering Messages with a SQL Condition

The reader can be given a SQL condition which further filters the messages read beyond selecting only the messages of the stream being read.

For example, the reader can read messages from someStream-123 whose position is 0.

Read.('someStream-123', condition: 'position = 0')

The above example isn't a realistic use of this feature. It's a contrived example merely intended to demonstrate the mechanics of use the SQL condition.

Constructing a Reader

Readers can be constructed in one of two ways:

• Via the constructor
• Via the initializer

Via the Constructor

The constructor not only instantiates the reader, but also invokes the reader's configure instance method, which constructs the reader's operational dependencies.

self.build(stream_name, position: 0, batch_size: 1000, condition: nil, session: nil)

Returns

Instance of the MessageStore::Postgres::Read class.

Parameters

Name	Description	Type
stream_name	Name of stream that the reader will read	String
position	Position of the message to start reading from	Integer
batch_size	Number of messages to retrieve with each query to the message store	Integer
condition	SQL condition to filter the batch by	String
session	An existing session object to use, rather than allowing the reader to create a new session	MessageStore::Postgres::Session

Via the Initializer

self.initialize(stream_name, position, batch_size)

Returns

Instance of the MessageStore::Postgres::Read class.

Parameters

Name	Description	Type
stream_name	Name of stream that the reader will read	String
position	Position of the message to start reading from	Integer
batch_size	Number of messages to retrieve with each query to the message store	Integer
session	An existing session object to use, rather than allowing the reader to create a new session	MessageStore::Postgres::Session

By constructing a reader using the initializer, the reader's dependencies are not set to operational dependencies. They remain inert substitutes.

TIP

See the useful objects user guide for background on inert substitutes.

Assigning a Reader as a Dependency

self.configure(receiver, stream_name, attr_name: :read, position: 0, batch_size: 1000, condition: nil, session: nil)

Constructs an instance of the reader and assigns it to the receiver's read attribute. By default, the receiving attribute's name is expected to be read, but it can be altered with the use of the attr_name parameter.

something = Something.new
Messaging::Postgres::Read.configure(something)

something.write
# => #<Messaging::Postgres::Read:0x...>

Parameters

Name	Description	Type
receiver	The object that will receive the constructed reader	Object
attr_name	The receiver's attribute that will be assigned the constructed reader	Symbol
stream_name	Name of stream that the reader will read	String
position	Position of the message to start reading from	Integer
batch_size	Number of messages to retrieve with each query to the message store	Integer
condition	SQL condition to filter the batch by	String
session	An existing session object to use, rather than allowing the reader to create a new session	MessageStore::Postgres::Session

TIP

See the useful objects user guide for background on configuring dependencies.

Log Tags

The following tags are applied to log messages recorded by a reader:

Tag	Description
read	Applied to all log messages recorded by a reader
message_store	Applied to all log messages recorded inside the MessageStore namespace

The following tags may be applied to log messages recorded by a reader:

Tag	Description
message_data	Applied to log messages that record the data content of read message data
data	Applied to log messages that record the data content of read message data

See the logging user guide for more on log tags.