easyq

A simple and plain text tcp message broker

Instalation

pip install easyq

Or

pip install git+https://github.com/pylover/easyq

Development

clone git@github.com:pylover/easyq.git
cd easyq
pip install -r requirements-dev.txt
pip install -e .

CLI auto completion

easyq autocompletion install [-s]

Uninstall

easyq autocompletion uninstall [-s]

If you are inside a virtualenv it will add the bash completion in $VIRTUAL_ENV/bin/postactivate. if not, the $HOME/.bashrc will be used.

The -s flag tells the installer to add the # PYTHON_ARGCOMPLETE_OK after the shebang line of $(which easyq).

For more info see: argcomplete

Starting the server

easyq server start [-h] [-b [HOST:]PORT] [-C DIRECTORY] [-c FILE] [-o key1.key2=value]

example

easyq server start
easyq server start -o dispatchers=3
easyq server start -o logging.level=warning

More:

usage: easyq server start [-h] [-b {HOST:}PORT] [-C DIRECTORY] [-c FILE]
                          [-o key1.key2=value]

optional arguments:
  -h, --help            show this help message and exit
  -b {HOST:}PORT, --bind {HOST:}PORT
                        Bind Address. if ommited the value from the config
                        file will be usedThe default config value is:
                        localhost:1085
  -C DIRECTORY, --directory DIRECTORY
                        Change to this path before starting the server
  -c FILE, --config-file FILE
  -o key1.key2=value, --option key1.key2=value
                        Configuration value to override. this option could be
                        passed multiple times.

Server Configuration

The builtin configuration is same as the bellow:

bind: localhost:1085 

authentication:
  method: trust

logging:
  level: debug

queues:
  default:
    maxsize: 100

dispatchers: 1
dispatcher:
  messages_per_queue: 5
  intervals: .3

But can oveeride it by -o/--option key.subkey=value or specifying a yaml config file with the -c/--config-file myfile.yaml.

Client

Console

Feel free to use any regular tcp text client such as telnet and or nc.

Start a server in another terminal to handle our connections.

easyq server start

A new shell to listen on queue: q:

echo "LOGIN Listener;PULL FROM q;" | nc localhost 1085

Watch here for messages you send in the next step.

Netcat

$ nc localhost 1085
LOGIN username;
HI username;
PULL FROM q;
PUSH Hello INTO q;
MESSAGE Hello FROM q;
IGNORE q;
^C

Telnet

$ telnet localhost 1085
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
LOGIN username;
HI username;
PULL FROM q;
PUSH Hello INTO q;
MESSAGE Hello FROM q;
IGNORE q;
^]
telnet> close
Connection closed.

The server's log should be something like this:

2018-05-22 23:50:27 DEBUG PROTO Data received: b'LOGIN username;'
2018-05-22 23:50:27 INFO PROTO Authenticating: ('127.0.0.1', 33826)
2018-05-22 23:50:27 INFO PROTO Login success: username from ('127.0.0.1', 33826)
2018-05-22 23:50:41 DEBUG PROTO Data received: b'PULL FROM q;'
2018-05-22 23:50:41 INFO QM Queue q just created.
2018-05-22 23:50:41 INFO QM Queue q was subscribed by username
2018-05-22 23:50:57 DEBUG WORKER 0 Cycle: 600
2018-05-22 23:51:00 DEBUG PROTO Data received: b'PUSH Hello INTO q;'
2018-05-22 23:51:00 DEBUG WORKER 0 Dispatching b'Hello'
2018-05-22 23:51:00 DEBUG QM Dispatching message b'Hello' from queue q to username
2018-05-22 23:51:09 DEBUG PROTO Data received: b'IGNORE q;'
2018-05-22 23:51:09 INFO QM Queue q was ignored by username
2018-05-22 23:51:17 DEBUG PROTO EOF Received: ('127.0.0.1', 33826)
2018-05-22 23:51:17 INFO PROTO Connection lost: ('127.0.0.1', 33826)

Python AsyncIO API

import asyncio

from easyq.client import connect


async def message_received(queue, message):
    print(f'Messsage received from {queue.decode()}: {message.decode()}')


async def main():
    client = await connect('Username', 'localhost', 1085)
    await client.pull(b'q', message_received)
    await client.push(b'q', b'Hello')
    await asyncio.sleep(2)
    await client.ignore(b'q', message_received)


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

Systemd

/etc/systemd/system/easyq.service

[Unit]
Description="EasyQ daemon"
After=network.target

[Service]
User=eq
ExecStart=/usr/local/bin/easyq server start --config-file /etc/easyq/easyq.yml
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

adduser --system eq
systemctl daemon-reload
systemctl enable easyq.service
systemctl start easysq