liner_broker 1.3.3

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
53
54
55
56
57
58
59
60
61
62
63
64
65

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

import argparse
import asyncio
import os
import shutil
import sys
import time
from pathlib import Path

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

from python import liner  # noqa: E402


def release_lib() -> Path:
    target_base = Path(os.environ["CARGO_TARGET_DIR"]) if os.environ.get("CARGO_TARGET_DIR") else ROOT / "target"
    lib_path = target_base / "release" / "libliner_broker.so"
    deps_lib = target_base / "release" / "deps" / "libliner_broker.so"
    if not lib_path.exists() and deps_lib.exists():
        shutil.copy2(deps_lib, lib_path)
    return lib_path


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("--postgres-url", required=True, help="Shared PostgreSQL URL (LINER_TEST_POSTGRES_URL)")
    parser.add_argument("--subscr-topic", default="")
    parser.add_argument("--unsubscr-topic", default="")

    args = parser.parse_args()

    liner.loadLib(str(release_lib()))

    h = liner.Client.new_postgres(
        args.client_name,
        args.client_topic,
        args.client_addr,
        args.postgres_url,
    )
    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 not h.run(receive_cback1):
        raise SystemExit("liner run() failed")

    if args.unsubscr_topic:
        # Brief pause so peer discovery / TCP can come up before the event is emitted.
        time.sleep(0.3)
        h.unsubscribe(args.unsubscr_topic)

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