| Name | mrsal JSON |
| Version |
0.7.2
JSON |
| download |
| home_page | |
| Summary | Mrsal is a simple to use message broker abstraction on top of RabbitMQ and Pika. |
| upload_time | 2023-12-22 11:11:02 |
| maintainer | Raafat |
| docs_url | None |
| author | Raafat |
| requires_python | >=3.10,<4.0 |
| license | |
| keywords |
message broker
rabbitmq
pika
mrsal
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# MRSAL <img align="right" width="125" alt="20200907_104224" src="https://github.com/NeoMedSys/mrsal/assets/9555305/ffb55c86-1e9d-4806-b1ca-c906d6985e3a">
[](https://pypi.org/project/mrsal/) [](https://www.python.org/downloads/release/python-3103/) [](https://github.com/NeoMedSys/mrsal/blob/main/FullGuide.md)
<!-- [](https://github.com/NeoMedSys/fatty-model-deploy/actions/runs/5576890234) -->
[](./reports/junit/junit.xml) [](./reports/coverage/htmlcov/index.html) [](./reports/flake8/flake8.txt)
## Intro
Mrsal is a simple to use message broker abstraction on top of [RabbitMQ](https://www.rabbitmq.com/) and [Pika](https://pika.readthedocs.io/en/stable/index.html). The goal is to make Mrsal trivial to re-use in all services of a distributed system and to make the use of advanced message queing protocols easy and safe. No more big chunks of repetive code across your services or bespoke solutions to handle dead letters.
###### Mrsal is Arabic for a small arrow and is used to describe something that performs a task with lightness and speed.
## Quick Start guide
### 0. Install
First things first:
```bash
pip install mrsal
```
We need to install RabbitMQ to use Mrsal. Head over to [install](https://www.rabbitmq.com/download.html) RabbitMQ. Make sure to stick to the configuration that you give the installation throughout this guide. You can also use the [Dockerfile](https://github.com/NeoMedSys/mrsal/blob/main/Dockerfile) and the [docker-compose](https://github.com/NeoMedSys/mrsal/blob/main/docker-compose.yml) that we are using in the full guide.
Next set the default username, password and servername for your RabbitMQ setup. It's advisable to use a `.env` script or the rc file for persistence.
```bash
[RabbitEnvVars]
RABBITMQ_DEFAULT_USER=******
RABBITMQ_DEFAULT_PASS=******
RABBITMQ_DEFAULT_VHOST=******
RABBITMQ_DOMAIN=******
RABBITMQ_DOMAIN_TLS=******
RABBITMQ_GUI_PORT=******
RABBITMQ_PORT=******
RABBITMQ_PORT_TLS=******
# FOR TLS
RABBITMQ_CAFILE=/path/to/file
RABBITMQ_CERT=/path/to/file
RABBITMQ_KEY=/path/to/file
```
Please read the [full guide](https://github.com/NeoMedSys/mrsal/blob/main/FullGuide.md) to understand what Mrsal currently can and can't do.
###### Mrsal was first developed by NeoMedSys and the research group [CRAI](https://crai.no/) at the univeristy hospital of Oslo.
### 1. Setup and connect
The first thing we need to do is to setup our rabbit server before we can subscribe and publish to it. Lets set up a server on our localhost with the port and credentials we used when spinning up the docker-compose
```python
import json
import pika
from mrsal.mrsal import Mrsal
# If you want to use SSL for external listening then set it to True
SSL = False
# Note RabbitMQ container is listening on:
# 1. When SSL is False the default port 5672 which is exposed to RABBITMQ_PORT in docker-compose
# 2. When SSL is True the default port 5671 which is exposed to RABBITMQ_PORT_TLS in docker-compose
port = RABBITMQ_PORT_TLS if SSL else RABBITMQ_PORT
host = RABBITMQ_DOMAIN_TLS if SSL else RABBITMQ_DOMAIN
# It should match with the env specifications (RABBITMQ_DEFAULT_USER, RABBITMQ_DEFAULT_PASS)
credentials=(RABBITMQ_DEFAULT_USER, RABBITMQ_DEFAULT_PASS)
# It should match with the env specifications (RABBITMQ_DEFAULT_VHOST)
v_host = RABBITMQ_DEFAULT_VHOST
mrsal = Mrsal(
host=host,
port=port,
credentials=credentials,
virtual_host=v_host,
ssl=SSL
)
mrsal.connect_to_server()
```
### 2 Publish
Now lets publish our message of friendship on the friendship exchange.
Note: When `fast_setup=True` that means Mrsal will create the specified `exchange` and `queue`, then bind them together using `routing_key`.
```python
# BasicProperties is used to set the message properties
prop = pika.BasicProperties(
app_id='friendship_app',
message_id='friendship_msg',
content_type='text/plain',
content_encoding='utf-8',
delivery_mode=pika.DeliveryMode.Persistent,
headers=None)
message_body = 'Hello'
# Publish the message to the exchange to be routed to queue
mrsal.publish_message(exchange='friendship',
exchange_type='direct',
queue='friendship_queue',
routing_key='friendship_key',
message=json.dumps(message_body),
prop=prop,
fast_setup=True)
```
### 3 Consume
Now lets setup a consumer that will listen to our very important messages. If you are using scripts rather than notebooks then it's advisable to run consume and publish separately. We are going to need a callback function which is triggered upon receiving the message from the queue we subscribe to. You can use the callback function to activate something in your system.
Note:
- If you start a consumer with `callback_with_delivery_info=True` then your callback function should have at least these params `(method_frame: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, message_param: str)`.
- If not, then it should have at least `(message_param: str)`
```python
import json
def consumer_callback_with_delivery_info(host_param: str, queue_param: str, method_frame: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, message_param: str):
str_message = json.loads(message_param).replace('"', '')
if 'Hello' in str_message:
app_id = properties.app_id
msg_id = properties.message_id
print(f'app_id={app_id}, msg_id={msg_id}')
print('Hola habibi')
return True # Consumed message processed correctly
return False
def consumer_callback(host_param: str, queue_param: str, message_param: str):
str_message = json.loads(message_param).replace('"', '')
if 'Hello' in str_message:
print('Hola habibi')
return True # Consumed message processed correctly
return False
mrsal.start_consumer(
queue='friendship_queue',
callback=consumer_callback_with_delivery_info,
callback_args=(test_config.HOST, 'friendship_queue'),
inactivity_timeout=1,
requeue=False,
fast_setup=True,
callback_with_delivery_info=True
)
```
Done! Your first message of friendship has been sent to the friendship queue on the exchange of friendship.
### 3 Concurrent Consumers
Sometimes we need to start multiple consumers to listen to the **same** **queue** and process received messages **concurrently**.
You can do that by calling `start_concurrence_consumer` which takes `total_threads` param in addition to the same parameters used in `start_consumer`.
This method will create a **thread pool** and _spawn new_ `Mrsal` object and start **new consumer** for every thread.
```python
import json
import time
import pika
from pika.exchange_type import ExchangeType
import mrsal.config.config as config
import tests.config as test_config
from mrsal.mrsal import Mrsal
mrsal = Mrsal(host=test_config.HOST,
port=config.RABBITMQ_PORT,
credentials=config.RABBITMQ_CREDENTIALS,
virtual_host=config.V_HOST)
mrsal.connect_to_server()
APP_ID="TEST_CONCURRENT_CONSUMERS"
EXCHANGE="GoodFriends"
EXCHANGE_TYPE='direct'
QUEUE_EMERGENCY="alleSindInkludiert" # place the excluded (but no fundamentalist danke) in an emergency queue
NUM_THREADS=3
NUM_MESSAGES=3
INACTIVITY_TIMEOUT=30 # time out after 30 seconds
ROUTING_KEY="bleib-cool"
MESSAGE_ID="Bleib cool und alles wird besser"
def test_concurrent_consumer():
# Start concurrent consumers
mrsal.start_concurrence_consumer(total_threads=NUM_THREADS, queue=QUEUE_EMERGENCY,
callback=consumer_callback_with_delivery_info,
callback_args=(test_config.HOST, QUEUE_EMERGENCY),
exchange=EXCHANGE, exchange_type=EXCHANGE_TYPE,
routing_key=ROUTING_KEY,
inactivity_timeout=INACTIVITY_TIMEOUT,
fast_setup=True,
callback_with_delivery_info=True)
mrsal.close_connection()
def consumer_callback_with_delivery_info(host_param: str, queue_param: str, method_frame: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, message_param: str):
time.sleep(5)
return True
```
That simple! You have now setup a full advanced message queueing protocol that you can use to promote friendship or other necessary communication between your services.
###### Note! Please refer to the >>>`FULL GUIDE`<<< on how to use customize Mrsal to meet specific needs. There are many parameters and settings that you can use to set up a more sophisticated communication protocol.
---
## References
- [RabbitMQ Tutorials](https://www.rabbitmq.com/getstarted.html)
- [RabbitMQ Exchange Types: 6 Categories Explained Easy](https://hevodata.com/learn/rabbitmq-exchange-type/)
- [What is a Delayed Message Exchange?](https://www.cloudamqp.com/blog/what-is-a-delayed-message-exchange-in-rabbitmq.html#:~:text=The%20RabbitMQ%20delayed%20exchange%20plugin,in%20milliseconds%20can%20be%20specified.)
- [RabbitMQ Delayed Message Plugin](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange)
- [When and how to use the RabbitMQ Dead Letter Exchange](https://www.cloudamqp.com/blog/when-and-how-to-use-the-rabbitmq-dead-letter-exchange.html)
- [What is a RabbitMQ vhost?](https://www.cloudamqp.com/blog/what-is-a-rabbitmq-vhost.html)
- [Message Brokers](https://www.ibm.com/cloud/learn/message-brokers)
- [How to Use map() with the ThreadPoolExecutor in Python](https://superfastpython.com/threadpoolexecutor-map/)
- [concurrent.futures](https://docs.python.org/3/library/concurrent.futures.html)
- [mrsal_icon](https://www.pngegg.com/en/png-mftic)
---
Raw data
{
"_id": null,
"home_page": "",
"name": "mrsal",
"maintainer": "Raafat",
"docs_url": null,
"requires_python": ">=3.10,<4.0",
"maintainer_email": "rafatzahran90@gmail.com",
"keywords": "message broker,RabbitMQ,Pika,Mrsal",
"author": "Raafat",
"author_email": "rafatzahran90@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/3b/cc/cb0f373778c77c6cbe8caa347b8ab598f338696d750859dc39c7f8b19357/mrsal-0.7.2.tar.gz",
"platform": null,
"description": "# MRSAL <img align=\"right\" width=\"125\" alt=\"20200907_104224\" src=\"https://github.com/NeoMedSys/mrsal/assets/9555305/ffb55c86-1e9d-4806-b1ca-c906d6985e3a\">\n[](https://pypi.org/project/mrsal/) [](https://www.python.org/downloads/release/python-3103/) [](https://github.com/NeoMedSys/mrsal/blob/main/FullGuide.md)\n<!-- [](https://github.com/NeoMedSys/fatty-model-deploy/actions/runs/5576890234) -->\n[](./reports/junit/junit.xml) [](./reports/coverage/htmlcov/index.html) [](./reports/flake8/flake8.txt)\n\n\n## Intro\nMrsal is a simple to use message broker abstraction on top of [RabbitMQ](https://www.rabbitmq.com/) and [Pika](https://pika.readthedocs.io/en/stable/index.html). The goal is to make Mrsal trivial to re-use in all services of a distributed system and to make the use of advanced message queing protocols easy and safe. No more big chunks of repetive code across your services or bespoke solutions to handle dead letters. \n\n###### Mrsal is Arabic for a small arrow and is used to describe something that performs a task with lightness and speed. \n\n## Quick Start guide\n\n### 0. Install\n\nFirst things first: \n\n```bash\npip install mrsal\n```\n\nWe need to install RabbitMQ to use Mrsal. Head over to [install](https://www.rabbitmq.com/download.html) RabbitMQ. Make sure to stick to the configuration that you give the installation throughout this guide. You can also use the [Dockerfile](https://github.com/NeoMedSys/mrsal/blob/main/Dockerfile) and the [docker-compose](https://github.com/NeoMedSys/mrsal/blob/main/docker-compose.yml) that we are using in the full guide.\n\nNext set the default username, password and servername for your RabbitMQ setup. It's advisable to use a `.env` script or the rc file for persistence.\n\n```bash\n[RabbitEnvVars]\nRABBITMQ_DEFAULT_USER=******\nRABBITMQ_DEFAULT_PASS=******\nRABBITMQ_DEFAULT_VHOST=******\nRABBITMQ_DOMAIN=******\nRABBITMQ_DOMAIN_TLS=******\n\nRABBITMQ_GUI_PORT=******\nRABBITMQ_PORT=******\nRABBITMQ_PORT_TLS=******\n\n# FOR TLS\nRABBITMQ_CAFILE=/path/to/file\nRABBITMQ_CERT=/path/to/file\nRABBITMQ_KEY=/path/to/file\n```\n\nPlease read the [full guide](https://github.com/NeoMedSys/mrsal/blob/main/FullGuide.md) to understand what Mrsal currently can and can't do.\n\n###### Mrsal was first developed by NeoMedSys and the research group [CRAI](https://crai.no/) at the univeristy hospital of Oslo.\n\n### 1. Setup and connect\n\n\nThe first thing we need to do is to setup our rabbit server before we can subscribe and publish to it. Lets set up a server on our localhost with the port and credentials we used when spinning up the docker-compose\n\n```python\nimport json\nimport pika\nfrom mrsal.mrsal import Mrsal\n\n# If you want to use SSL for external listening then set it to True\nSSL = False\n\n# Note RabbitMQ container is listening on:\n# 1. When SSL is False the default port 5672 which is exposed to RABBITMQ_PORT in docker-compose\n# 2. When SSL is True the default port 5671 which is exposed to RABBITMQ_PORT_TLS in docker-compose\nport = RABBITMQ_PORT_TLS if SSL else RABBITMQ_PORT\nhost = RABBITMQ_DOMAIN_TLS if SSL else RABBITMQ_DOMAIN\n\n# It should match with the env specifications (RABBITMQ_DEFAULT_USER, RABBITMQ_DEFAULT_PASS)\ncredentials=(RABBITMQ_DEFAULT_USER, RABBITMQ_DEFAULT_PASS)\n\n# It should match with the env specifications (RABBITMQ_DEFAULT_VHOST)\nv_host = RABBITMQ_DEFAULT_VHOST\n\nmrsal = Mrsal(\n host=host,\n port=port,\n credentials=credentials,\n virtual_host=v_host,\n ssl=SSL\n)\n\nmrsal.connect_to_server()\n```\n\n### 2 Publish\nNow lets publish our message of friendship on the friendship exchange.\nNote: When `fast_setup=True` that means Mrsal will create the specified `exchange` and `queue`, then bind them together using `routing_key`.\n\n```python\n# BasicProperties is used to set the message properties\nprop = pika.BasicProperties(\n app_id='friendship_app',\n message_id='friendship_msg',\n content_type='text/plain',\n content_encoding='utf-8',\n delivery_mode=pika.DeliveryMode.Persistent,\n headers=None)\n\nmessage_body = 'Hello'\n\n# Publish the message to the exchange to be routed to queue\nmrsal.publish_message(exchange='friendship',\n exchange_type='direct',\n queue='friendship_queue',\n routing_key='friendship_key',\n message=json.dumps(message_body), \n prop=prop,\n fast_setup=True)\n```\n\n### 3 Consume\n\nNow lets setup a consumer that will listen to our very important messages. If you are using scripts rather than notebooks then it's advisable to run consume and publish separately. We are going to need a callback function which is triggered upon receiving the message from the queue we subscribe to. You can use the callback function to activate something in your system.\n\nNote: \n- If you start a consumer with `callback_with_delivery_info=True` then your callback function should have at least these params `(method_frame: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, message_param: str)`. \n- If not, then it should have at least `(message_param: str)`\n\n```python\nimport json\n\ndef consumer_callback_with_delivery_info(host_param: str, queue_param: str, method_frame: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, message_param: str):\n str_message = json.loads(message_param).replace('\"', '')\n if 'Hello' in str_message:\n app_id = properties.app_id\n msg_id = properties.message_id\n print(f'app_id={app_id}, msg_id={msg_id}')\n print('Hola habibi')\n return True # Consumed message processed correctly\n return False\n\ndef consumer_callback(host_param: str, queue_param: str, message_param: str):\n str_message = json.loads(message_param).replace('\"', '')\n if 'Hello' in str_message:\n print('Hola habibi')\n return True # Consumed message processed correctly\n return False\n\nmrsal.start_consumer(\n queue='friendship_queue',\n callback=consumer_callback_with_delivery_info,\n callback_args=(test_config.HOST, 'friendship_queue'),\n inactivity_timeout=1,\n requeue=False,\n fast_setup=True,\n callback_with_delivery_info=True\n )\n```\n\nDone! Your first message of friendship has been sent to the friendship queue on the exchange of friendship.\n\n\n### 3 Concurrent Consumers\n\nSometimes we need to start multiple consumers to listen to the **same** **queue** and process received messages **concurrently**. \nYou can do that by calling `start_concurrence_consumer` which takes `total_threads` param in addition to the same parameters used in `start_consumer`.\nThis method will create a **thread pool** and _spawn new_ `Mrsal` object and start **new consumer** for every thread. \n\n```python\nimport json\nimport time\n\nimport pika\nfrom pika.exchange_type import ExchangeType\n\nimport mrsal.config.config as config\nimport tests.config as test_config\nfrom mrsal.mrsal import Mrsal\n\n\nmrsal = Mrsal(host=test_config.HOST,\n port=config.RABBITMQ_PORT,\n credentials=config.RABBITMQ_CREDENTIALS,\n virtual_host=config.V_HOST)\nmrsal.connect_to_server()\n\nAPP_ID=\"TEST_CONCURRENT_CONSUMERS\"\nEXCHANGE=\"GoodFriends\"\nEXCHANGE_TYPE='direct'\nQUEUE_EMERGENCY=\"alleSindInkludiert\" # place the excluded (but no fundamentalist danke) in an emergency queue \nNUM_THREADS=3\nNUM_MESSAGES=3\nINACTIVITY_TIMEOUT=30 # time out after 30 seconds\nROUTING_KEY=\"bleib-cool\"\nMESSAGE_ID=\"Bleib cool und alles wird besser\"\n\ndef test_concurrent_consumer():\n # Start concurrent consumers\n mrsal.start_concurrence_consumer(total_threads=NUM_THREADS, queue=QUEUE_EMERGENCY,\n callback=consumer_callback_with_delivery_info,\n callback_args=(test_config.HOST, QUEUE_EMERGENCY),\n exchange=EXCHANGE, exchange_type=EXCHANGE_TYPE,\n routing_key=ROUTING_KEY,\n inactivity_timeout=INACTIVITY_TIMEOUT,\n fast_setup=True,\n callback_with_delivery_info=True)\n\n mrsal.close_connection()\n\ndef consumer_callback_with_delivery_info(host_param: str, queue_param: str, method_frame: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, message_param: str):\n time.sleep(5)\n return True\n```\n\nThat simple! You have now setup a full advanced message queueing protocol that you can use to promote friendship or other necessary communication between your services.\n\n###### Note! Please refer to the >>>`FULL GUIDE`<<< on how to use customize Mrsal to meet specific needs. There are many parameters and settings that you can use to set up a more sophisticated communication protocol.\n---\n## References\n\n- [RabbitMQ Tutorials](https://www.rabbitmq.com/getstarted.html)\n- [RabbitMQ Exchange Types: 6 Categories Explained Easy](https://hevodata.com/learn/rabbitmq-exchange-type/)\n- [What is a Delayed Message Exchange?](https://www.cloudamqp.com/blog/what-is-a-delayed-message-exchange-in-rabbitmq.html#:~:text=The%20RabbitMQ%20delayed%20exchange%20plugin,in%20milliseconds%20can%20be%20specified.)\n- [RabbitMQ Delayed Message Plugin](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange)\n- [When and how to use the RabbitMQ Dead Letter Exchange](https://www.cloudamqp.com/blog/when-and-how-to-use-the-rabbitmq-dead-letter-exchange.html)\n- [What is a RabbitMQ vhost?](https://www.cloudamqp.com/blog/what-is-a-rabbitmq-vhost.html)\n- [Message Brokers](https://www.ibm.com/cloud/learn/message-brokers)\n- [How to Use map() with the ThreadPoolExecutor in Python](https://superfastpython.com/threadpoolexecutor-map/)\n- [concurrent.futures](https://docs.python.org/3/library/concurrent.futures.html)\n- [mrsal_icon](https://www.pngegg.com/en/png-mftic)\n---\n",
"bugtrack_url": null,
"license": "",
"summary": "Mrsal is a simple to use message broker abstraction on top of RabbitMQ and Pika.",
"version": "0.7.2",
"project_urls": null,
"split_keywords": [
"message broker",
"rabbitmq",
"pika",
"mrsal"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "06a1b9947d2cfee69b23183805bb93ba1ca2e928510ed64ec5db3b60e003da8e",
"md5": "2537e2b45adee2ae28d4a6b1117ed5d4",
"sha256": "7a0a68e60bec7669be3314e9352e25823540077eafb3412b0323d68aab962b20"
},
"downloads": -1,
"filename": "mrsal-0.7.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2537e2b45adee2ae28d4a6b1117ed5d4",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10,<4.0",
"size": 26833,
"upload_time": "2023-12-22T11:11:00",
"upload_time_iso_8601": "2023-12-22T11:11:00.407255Z",
"url": "https://files.pythonhosted.org/packages/06/a1/b9947d2cfee69b23183805bb93ba1ca2e928510ed64ec5db3b60e003da8e/mrsal-0.7.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3bcccb0f373778c77c6cbe8caa347b8ab598f338696d750859dc39c7f8b19357",
"md5": "50c2861dbcf0718df35e1e36e541c1f7",
"sha256": "7092f02cb463db681b084fee095373c90c8086518875adc6d68c168a5137a849"
},
"downloads": -1,
"filename": "mrsal-0.7.2.tar.gz",
"has_sig": false,
"md5_digest": "50c2861dbcf0718df35e1e36e541c1f7",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10,<4.0",
"size": 28568,
"upload_time": "2023-12-22T11:11:02",
"upload_time_iso_8601": "2023-12-22T11:11:02.542671Z",
"url": "https://files.pythonhosted.org/packages/3b/cc/cb0f373778c77c6cbe8caa347b8ab598f338696d750859dc39c7f8b19357/mrsal-0.7.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-22 11:11:02",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "mrsal"
}
