liner_broker 1.3.0

Redis based message serverless broker.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import argparse
import asyncio
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(ROOT))

from python import liner  # noqa: E402


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--client-name", required=True)
    parser.add_argument("--client-topic", required=True)
    parser.add_argument("--client-addr", required=True)
    parser.add_argument("--sqlite-db", required=True, help="Shared SQLite store path")
    parser.add_argument("--subscr-topic", default="")
    parser.add_argument("--unsubscr-topic", default="")

    args = parser.parse_args()

    lib = ROOT / "target" / "release" / "libliner_broker.so"
    liner.loadLib(str(lib))

    h = liner.Client.new_sqlite(
        args.client_name,
        args.client_topic,
        args.client_addr,
        args.sqlite_db,
        "",
    )
    h.clear_addresses_of_topic()
    h.clear_stored_messages()

    def receive_cback1(to: str, from_: str, data_: bytes):
        print(f"{args.client_name} receive_from {from_}, data: {data_}")
        h.send_to(from_, bytearray(data_), True)

    if args.subscr_topic:
        h.subscribe(args.subscr_topic)
    if args.unsubscr_topic:
        h.unsubscribe(args.unsubscr_topic)

    if not h.run(receive_cback1):
        raise SystemExit("liner run() failed")

    loop = asyncio.new_event_loop()
    loop.run_forever()