rust-pipe 0.1.0

Lightweight typed task dispatch from Rust to polyglot workers (TypeScript, Python, Go, Java, C#, Ruby, Elixir, Swift, PHP)
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
66
67
68
69
70
71
72
73
74
75
76

"""
Minimal Python worker for e2e testing.
Connects to rust-pipe dispatcher via WebSocket.
"""
import asyncio
import json
import os
import time

import websockets

DISPATCHER_URL = os.environ.get("DISPATCHER_URL", "ws://127.0.0.1:19901")
WORKER_ID = f"py-test-{int(time.time())}"


async def main():
    async with websockets.connect(DISPATCHER_URL) as ws:
        # Register
        await ws.send(json.dumps({
            "type": "WorkerRegister",
            "registration": {
                "workerId": WORKER_ID,
                "supportedTasks": ["echo", "multiply", "reverse-string"],
                "maxConcurrency": 5,
                "language": "Python",
            },
        }))
        print(f"[py-worker] Registered as {WORKER_ID}", flush=True)

        async for raw in ws:
            msg = json.loads(raw)

            if msg["type"] == "TaskDispatch":
                task = msg["task"]
                print(f"[py-worker] Received task: {task['taskType']} ({task['id']})", flush=True)

                start = time.time()
                try:
                    if task["taskType"] == "echo":
                        result = {"echoed": task["payload"]}
                    elif task["taskType"] == "multiply":
                        result = {"product": task["payload"]["a"] * task["payload"]["b"]}
                    elif task["taskType"] == "reverse-string":
                        result = {"reversed": task["payload"]["text"][::-1]}
                    else:
                        raise ValueError(f"Unknown task type: {task['taskType']}")

                    await ws.send(json.dumps({
                        "type": "TaskResult",
                        "result": {
                            "taskId": task["id"],
                            "status": "Completed",
                            "payload": result,
                            "durationMs": int((time.time() - start) * 1000),
                            "workerId": WORKER_ID,
                        },
                    }))
                except Exception as e:
                    await ws.send(json.dumps({
                        "type": "TaskResult",
                        "result": {
                            "taskId": task["id"],
                            "status": "Failed",
                            "error": {"code": "HANDLER_ERROR", "message": str(e), "retryable": False},
                            "durationMs": int((time.time() - start) * 1000),
                            "workerId": WORKER_ID,
                        },
                    }))

            elif msg["type"] == "Shutdown":
                print("[py-worker] Shutdown received", flush=True)
                break


if __name__ == "__main__":
    asyncio.run(main())