<commit_before>OroMessageQueue Component
=========================

The component incorporates message queue in your application via different transports.
It contains several layers.

The lowest layer is called Transport and provides an abstraction of transport protocol.
The Consumption layer provides tools to consume messages, such as cli command, signal handling, logging, extensions.
It works on top of transport layer.

The Client layer provides an ability to start producing\consuming messages with as little configuration as possible .

Table of Contents
-----------------
 - [External Links](#external_links)
 - [What is Message Queue](#what_is_message_queue)
 - [Dictionary](#dictionary)
 - [Message Processors](#message_processors)
 - [Jobs](#jobs)
 - [Flow](#flow)
 - [Usage](#usage)
 
External Links
--------------
 - [What is Message Queue](http://www.ibm.com/support/knowledgecenter/SSFKSJ_9.0.0/com.ibm.mq.pro.doc/q002620_.htm)
 - [Message Queue Benefits](https://www.iron.io/top-10-uses-for-message-queue/) (most of them are applicable for Oro Message Queue Component)
 - [Rabbit MQ Introduction](https://www.rabbitmq.com/tutorials/tutorial-one-php.html)


What is Message Queue
---------------------
Message queues provide an asynchronous communications protocol, meaning that the sender and the receiver of the message 
do not need to interact with the message queue at the same time. Messages placed onto the queue are stored until the 
recipient retrieves them. A message does not have information about previous and next messages. 

###Therefore, Message Queues should be used if:

* A process can be executed asynchronously
* A process does not affect User Experience
* Processes need to be executed in parallel for faster performance
* You need a guaranty of processing
* You need scalability

###Publish/subscribe messaging

OroMessageQueue uses *Publish/subscribe messaging*. It means that the sending application publishes (sends) a 
message with a specific *topic* and a *consumer* finds a subscriber(s) for the topic. Publish/subscribe messaging allows 
to decouple the provider of information, from the consumers of that information. The sending application and the receiving 
application do not need to know anything about each other for the information to be sent and received.

Dictionary
----------

* **Message** - an information message which contains a *message topic* that indicates which *message processor*(s) will process it and a *message body* - array of some parameters needed for the processing, for example an entity id or a channel name. Messages are created and sent by a *message producer* and put to the "tail" of the *message queue*. When the message comes up, it is processed by a *consumer* using a *message processor*. Messages also contain a number of additional settings (see [Message settings](#message-settings)).
* **Message Queue** - a FIFO queue that holds *queue messages* until they are processed. There can be one or more queues. If we use only one queue it is much easier. If there are several queues it is much more difficult but more flexible sometimes. 
* **Consumer** - a component which takes messages from the queue and processes it. It processes one message at a time: once one message has finished processing, the next message follows. For each message, the consumer runs a *message processor* subscribed to the *message topic* (if one exists). If there are several processors subscribed to the same topic, they can be run in parallel (actually messages are sent via broker and if the broker sees that a message has several receivers, it clones the message to create an individual message for each receiver). There can be more than one consumer and they can work on different servers. It can be done to increase the performance. When implementing a message processor, a developer, should remember that *there can be several consumers working on different servers*.  
* **Message processor** processes the queue messages (i.e. contains a code that should run when a consumer processes a message with the specified topic).
* **Message Topic** - some identifier that indicates which message processor should be executed for the message. One processor can subscribe to several topics. Also there can be several processes subscribed to the same topic.
* **Job**. A message processor can process a message directly or create a job. Jobs are created in the db and allow to monitor processes status, start and end time, interrupt processes. Also, if we split a process to a set parallel processes, jobs allow to monitor and control all the set. See details in [Jobs](#jobs) section. 

###Message settings

* **Topic** - refer to the term Message Topic above.
* **Body** - a string or json encoded array with some data.
* **Priority** (can be `MessagePriority::VERY_LOW`, `MessagePriority::LOW`, `MessagePriority::NORMAL`, `MessagePriority::HIGH`, `MessagePriority::VERY_HIGH`). Recognizing priority is simple: there are five queues, one queue per priority. Consumers process messages from the VERY_HIGH queue. If there are no messages in the VERY_HIGH queue, consumers process messages from the HIGH queue, etc. Consequently, if all other queues are empty, consumer processes messages from the VERY_LOW queue.
* **Expire** - a number of seconds after which the message should be removed from the queue without processing.
* **Delay** - a number of seconds the message should be delayed for before it is sent to a queue.


Message Processors
------------------

**Message Processors** are classes that process queue messages. They implement `MessageProcessorInterface`. In addition, they usually
subscribe to the specific topics and implement `TopicSubscriberInterface`.

A `process(MessageInterface $message, SessionInterface $session)` method describes the actions that should be performed when
a message is received. It can perform the actions directly or create a job. It can also produce a new message to run another
processor asynchronously.

###Processing status

The received message can be processed, rejected and re-queued. An exception can also be thrown.

**Message Processor will return `self::ACK` in the following cases:**

* If a message wass processed successfully
* If the created job returned `true`

It means that the message was processed successfully and is removed from the queue.

**Message Processor will return `self::REJECT` in the following cases:**

* If a message is broken
* If the created job returned `false`. 

It means that the message was not processed and is removed from the queue because it is unprocessable and will never become processable (e.g. a required parameter is missing or another permanent error appears).

**There could be two options:**

1. The message became unprocessable as a result of normal work. For example, when the message was sent to an the entity that existed at the moment of sending but somebody deleted it. The entity will not appear again and we can reject the message. It is normal workflow so user intervention is not required.
2. The message became unprocessable due a failure. For example, when an entity id was invalid or missing. This is abnormal behavior, the message should also be rejected but the processor requires user attention (e.g. log a critical error or even throw an exception).

**If a message cannot be processed temporarily**, for example, in case of connection timeout due server overload,  the `process`
method should return `self::REQUEUE`. The message will be returned to the queue again and will be processed later.
**This will also happen if an exception is thrown during processing or job running**.

The difference is that the `self::REQUEUE` returning puts the message to the "tail" of the queue and will be
processed after all other messages in the queue. However, if an exception is thrown, the message is put to the
"head" of the queue and will be processed again soon (this is the default behaviour that can be changed).

###Example

A processor receives a message with the entity id. It finds the entity and changes it status without creating any job.

```php
    /**
     * {@inheritdoc}
     */
    public function process(MessageInterface $message, SessionInterface $session)<commit_msg>BAP-13278: Improve Message Queue documentation (#6372)<commit_after>OroMessageQueue Component
=========================

The component incorporates message queue in your application via different transports.
It contains several layers.

The lowest layer is called Transport and provides an abstraction of transport protocol.
The Consumption layer provides tools to consume messages, such as cli command, signal handling, logging, extensions.
It works on top of transport layer.

The Client layer provides an ability to start producing\consuming messages with as little configuration as possible .

Table of Contents
-----------------
 - [External Links](#external_links)
 - [What is Message Queue](#what_is_message_queue)
 - [Dictionary](#dictionary)
 - [Message Processors](#message_processors)
 - [Jobs](#jobs)
 - [Flow](#flow)
 - [Usage](#usage)
 
External Links
--------------
 - [What is Message Queue](http://www.ibm.com/support/knowledgecenter/SSFKSJ_9.0.0/com.ibm.mq.pro.doc/q002620_.htm)
 - [Message Queue Benefits](https://www.iron.io/top-10-uses-for-message-queue/) (most of them are applicable for Oro Message Queue Component)
 - [Rabbit MQ Introduction](https://www.rabbitmq.com/tutorials/tutorial-one-php.html)


What is Message Queue
---------------------
Message queues provide an asynchronous communications protocol, meaning that the sender and the receiver of the message 
do not need to interact with the message queue at the same time. Messages placed onto the queue are stored until the 
recipient retrieves them. A message does not have information about previous and next messages. 

###Therefore, Message Queues should be used if:

* A process can be executed asynchronously
* A process does not affect User Experience
* Processes need to be executed in parallel for faster performance
* You need a guaranty of processing
* You need scalability

###Publish/subscribe messaging

OroMessageQueue uses *Publish/subscribe messaging*. It means that the sending application publishes (sends) a 
message with a specific *topic* and a *consumer* finds a subscriber(s) for the topic. Publish/subscribe messaging allows 
to decouple the provider of information, from the consumers of that information. The sending application and the receiving 
application do not need to know anything about each other for the information to be sent and received.

Dictionary
----------

* **Message** - an information message which contains a *message topic* that indicates which *message processor*(s) will process it and a *message body* - array of some parameters needed for the processing, for example an entity id or a channel name. Messages are created and sent by a *message producer* and put to the "tail" of the *message queue*. When the message comes up, it is processed by a *consumer* using a *message processor*. Messages also contain a number of additional settings (see [Message settings](#message-settings)).
* **Message Queue** - a FIFO queue that holds *queue messages* until they are processed. There can be one or more queues. If we use only one queue it is much easier. If there are several queues it is much more difficult but more flexible sometimes. 
* **Consumer** - a component which takes messages from the queue and processes it. It processes one message at a time: once one message has finished processing, the next message follows. For each message, the consumer runs a *message processor* subscribed to the *message topic* (if one exists). If there are several processors subscribed to the same topic, they can be run in parallel (actually messages are sent via broker and if the broker sees that a message has several receivers, it clones the message to create an individual message for each receiver). There can be more than one consumer and they can work on different servers. It can be done to increase the performance. When implementing a message processor, a developer, should remember that *there can be several consumers working on different servers*.  
* **Message processor** processes the queue messages (i.e. contains a code that should run when a consumer processes a message with the specified topic).
* **Message Topic** - some identifier that indicates which message processor should be executed for the message. One processor can subscribe to several topics. Also there can be several processes subscribed to the same topic.
* **Job**. A message processor can process a message directly or create a job. Jobs are created in the db and allow to monitor processes status, start and end time, interrupt processes. Also, if we split a process to a set parallel processes, jobs allow to monitor and control all the set. See details in [Jobs](#jobs) section. 

###Message settings

* **Topic** - refer to the term Message Topic above.
* **Body** - a string or json encoded array with some data.
* **Priority** (can be `MessagePriority::VERY_LOW`, `MessagePriority::LOW`, `MessagePriority::NORMAL`, `MessagePriority::HIGH`, `MessagePriority::VERY_HIGH`). Recognizing priority is simple: there are five queues, one queue per priority. Consumers process messages from the VERY_HIGH queue. If there are no messages in the VERY_HIGH queue, consumers process messages from the HIGH queue, etc. Consequently, if all other queues are empty, consumer processes messages from the VERY_LOW queue.
* **Expire** - a number of seconds after which the message should be removed from the queue without processing.
* **Delay** - a number of seconds the message should be delayed for before it is sent to a queue.


Message Processors
------------------

**Message Processors** are classes that process queue messages. They implement `MessageProcessorInterface`. In addition, they usually
subscribe to the specific topics and implement `TopicSubscriberInterface`.

A `process(MessageInterface $message, SessionInterface $session)` method describes the actions that should be performed when
a message is received. It can perform the actions directly or create a job. It can also produce a new message to run another
processor asynchronously.

###Processing status

The received message can be processed, rejected and re-queued. An exception can also be thrown.

**Message Processor will return `self::ACK` in the following cases:**

* If a message wass processed successfully
* If the created job returned `true`

It means that the message was processed successfully and is removed from the queue.

**Message Processor will return `self::REJECT` in the following cases:**

* If a message is broken
* If the created job returned `false`. 

It means that the message was not processed and is removed from the queue because it is unprocessable and will never become processable (e.g. a required parameter is missing or another permanent error appears).

**There could be two options:**

1. The message became unprocessable as a result of normal work. For example, when the message was sent to an the entity that existed at the moment of sending but somebody deleted it. The entity will not appear again and we can reject the message. It is normal workflow so user intervention is not required.
2. The message became unprocessable due a failure. For example, when an entity id was invalid or missing. This is abnormal behavior, the message should also be rejected but the processor requires user attention (e.g. log a critical error or even throw an exception).

**If a message cannot be processed temporarily**, for example, in case of connection timeout due server overload,  the `process`
method should return `self::REQUEUE`. The message will be returned to the queue again and will be processed later.
**This will also happen if an exception is thrown during processing or job running**.

**The workflow of re-queuing messages (processor returns `self::REQUEUE`) is the following:**
 
1. A consumer processes a message (runs the `process` method of the message processor).
2. The `process` method returns `self::REQUEUE`.
3. The consumer puts the message (i.e. a copy of the message) to the end of the queue setting the `redelivery` flag to true. 
4. The consumer continues message processing (the requeued message is at the end of the queue).
5. When the turn of the requeued message comes, the `DelayRedeliveredMessageExtension` works and sets a delay for the requeued message.
6. The time set in the delay passes and the message is processed again.

**The workflow of re-queuing messages when an exception is thrown inside a message processor is slightly different:**

1. A consumer processes a message (runs `process` method of the message processor).
2. An exception is thrown inside the `process` method.
3. The consumer logs the exception and puts the message (i.e. a copy of the message) to the end of the queue setting the `redelivery` flag to true. Then the consumer fails with the exception.
4. The consumer should be re-run at this stage. It can be done manually or automatically with an utility like [supervisord](http://supervisord.org/). Manual re-run is preferred for developing as developers should review the exceptions thrown on the message processing. Automatic re-run is preferred for regression testing or prod.
5. The consumer continues message processing (the failing message is at the end of the queue).
6. When the turn of the failing message comes, the `DelayRedeliveredMessageExtension` works and sets a delay for the failing message.
7. After the delay time passes, the message is processed again and the consumer can fail again.

###Example

A processor receives a message with the entity id. It finds the entity and changes its status without creating any job.

```php
    /**
     * {@inheritdoc}
     */
    public function process(MessageInterface $message, SessionInterface $session)
{'fraction_non_alphanumeric': 0.06978553172574731, 'fraction_numerical': 0.002666962995888432, 'mean_word_length': 4.454242424242424, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5887490', 'n_tokens_mistral': 4236, 'n_tokens_neox': 4089, 'n_words': 2563}