Lynkio · Real-Time Event Engine

from lynkio import Lynk

# Single legacy port
app = Lynk(host="0.0.0.0", port=8080, protocol="AUTO", rate_limit=30, debug=True)

# Multi-listener (preferred)
from lynkio import Lynk, Listener

app = Lynk(
    listeners=[
        Listener("public",   port=8080, protocol="TCP", max_connections=1000),
        Listener("admin",    port=8081, protocol="TCP", max_connections=10),
        Listener("internal", port=9091, protocol="UDP", rate_limit=100),
    ],
    debug=True,
    serve_client=True,
)

app.run()

from lynkio import Lynk, Listener

listeners = [
    Listener("public",   port=8080, protocol="TCP", max_connections=1000),
    Listener("admin",    port=8081, protocol="TCP", max_connections=10),
    Listener("internal", port=9091, protocol="UDP", rate_limit=100),
]

app = Lynk(listeners=listeners, debug=True, serve_client=True)

# Scope routes / middleware to one listener:
@app.get("/", listener="public")
async def index(req):
    return {"listener": req.listener, "port": req.port}

@app.get("/admin/stats", listener="admin")
async def stats(req):
    return app.get_metrics()

# Global route: no listener= means it exists on every listener
@app.get("/health")
async def health(req):
    return {"status": "ok"}

await app.emit("chat", {"msg": "hi"})                 # All clients everywhere
await app.emit_to_listener("public", "news", {...})   # Only "public" listener
await app.emit_to_port(8080, "news", {...})           # Only port 8080
await app.emit_to_room("lobby", "chat", {...})        # Room broadcast

app.connection_count()                 # total
app.connection_count(listener="public")# per-listener
app.get_metrics()                      # per-listener dict

@app.get("/users/<user_id>/posts/<post_id>")
async def get_post(req, user_id, post_id):
    return {"user": user_id, "post": post_id}

@app.post("/submit", listener="admin")
async def submit(req):
    return json_response({"ok": True}, 201)

@app.route("/mixed", methods=["GET", "POST"])
async def mixed(req):
    return {"method": req.method}

from lynkio import send_file, render_template, json_response

@app.get("/download")
async def download(req):
    return send_file("report.pdf", as_attachment=True, cache_control="max-age=3600")

@app.get("/page")
async def page(req):
    return render_template("index.html", {"title": "Lynkio"})

app.static("/assets", "public", listener="public")

api = app.group("/api/v1", listener="admin")
@api.get("/health")
async def health(req):
    return {"status": "ok"}

ADMIN_KEY = "admin123"

@app.http_middleware(listener=None)      # runs on every listener
async def log_request(req):
    print(f"[{req.listener}] {req.method} {req.path}")
    return None

@app.http_middleware(listener="admin")
async def admin_auth(req):
    if req.headers.get("x-api-key") != ADMIN_KEY:
        return json_response({"error": "unauthorized"}, 401)
    return None

app.enable_cors(
    allowed_origins=["https://example.com"],
    allow_credentials=True,
)

@app.on("chat")                             # JSON events: {"event": "...", "data": ...}
async def chat(client, data):
    await app.emit_to_room(data["room"], "chat", data["text"])

@app.on_binary                            # Any binary frame
async def any_binary(client, payload: bytes):
    print(len(payload))

@app.on_binary_event("image")             # Named binary event
async def image_stream(client, payload: bytes):
    print(f"chunk {len(payload)} bytes")

from lynkio import StopProcessing

@app.middleware                     # global
async def log_mw(client, event, data):
    print(client.listener, event)
    return data

@app.middleware(listener="admin")   # listener-scoped NEW
async def admin_only(client, event, data):
    if not client.session.get("auth"):
        raise StopProcessing
    return data

@app.on_internal("connect")
async def on_connect(client, _):
    print("connected", client.id)

# Available: "connect", "disconnect", "subscribe", "unsubscribe", "message"

@app.on("join")
async def join(client, data):
    room = data["room"]
    app.join_room(client.id, room)
    await app.emit_to_room(room, "system", f"{client.id} joined", exclude=client.id)

@app.on("leave")
async def leave(client, data):
    app.leave_room(client.id, data["room"])

app.get_room_clients("lobby")          # -> Set[str]
app.is_client_in_room(client_id, "lobby")  # -> bool

@app.udp("/sensor", listener="internal")
async def sensor(req):
    payload = await req.json()
    return {"status": "ok", "echo": payload}

# Client side:
# echo '{"path":"/sensor","data":{"temp":22.5},"client_id":"sensor1"}' | nc -u localhost 9091

@app.task
async def monitor():
    while app._running:
        await asyncio.sleep(30)
        print(f"Active connections: {app.connection_count()}")

@app.schedule(3600)
async def hourly_cleanup():
    await app.query_database("logs", "DELETE FROM old_entries")

app = Lynk(enable_database=True)
db = app.create_database("myapp", create_log_table=True, auto_sync_log=True)

# Manual log entry
await app.add_log("runtime", level="INFO", message="Custom event", source="auth")

# Distributed query (runs on any registered soketDB in a thread executor)
rows = await app.query_database(
    "myapp",
    "SELECT * FROM http_logs WHERE status_code = $1",
    (200,),
)

<script src="/lynkio/client.js"></script>

<script>
  const lynk = new LynkClient(
    "ws://localhost:8080",          // WebSocket host
    { public: 8080, admin: 8081 },  // Optional: listener → port map
    { maxReconnectAttempts: 10 }
  );
</script>

const lynk = new LynkClient("ws://localhost:8080", { public: 8080 });

lynk.on("message", (data) => console.log("message", data));
lynk.onBinary((chunk) => console.log("binary", chunk.length));
lynk.onBinaryEvent("audio", (payload) => { /* Uint8Array */ });

await lynk.connect();
lynk.joinRoom("lobby");
lynk.emit("chat", { room: "lobby", text: "hi" });
lynk.sendBinaryEvent("image", new Uint8Array([1, 2, 3]));

const tid = lynk.createTask(() => lynk.emit("ping", {}), 5000);
lynk.clearTask(tid);

lynk.close();

const lynk = new LynkClient("ws://localhost:8080", { public: 8080 });
const srv  = lynk.server();

srv.use(async (req, res, next) => {
  console.log("[SW]", req.method, req.pathname);
  await next();
});

srv.intercept(async (req, res) => {
  res.setHeader("X-Powered-By", "Lynkio-SW");
});

srv.get("/api/hello", null, async (req, res) => {
  return { message: "hello from the service worker" };
});

srv.get("/users/<id>", null, async (req, res) => {
  return { id: req.params.id, query: req.query };
});

srv.post("/echo", null, async (req, res) => {
  return { body: req.body, type: req.bodyType };
});

await srv.ready();

const db = lynk.createDataset("mydb");

await db.create_table("users");
await db.insert("users", { name: "Alex", role: "admin" }, "user_1");

const user = await db.select_from("users", "user_1");
console.log(user);                        // { name: "Alex", role: "admin" }

await db.delete_from("users", "user_1");
await db.delete_all("users");
await db.delete_dataset();

from lynkio import LynkClient
import asyncio

async def main():
    client = LynkClient("localhost", 8765)
    await client.ws.connect()
    client.on("greet", print)
    await client.ws.emit("ping", "hello")
    await client.ws.send_binary_event("data", b'\x01\x02')
    # HTTP
    status, headers, body = await client.http.get("/api/status")
    # UDP
    resp = await client.udp.send(b'{"path":"/ping"}')
    await client.ws.close()

asyncio.run(main())

def metrics_plugin(app):
    @app.get("/metrics")
    async def metrics(req):
        return app.get_metrics()

app.use(metrics_plugin)

resp = await app.fetch("https://api.example.com/data")
data = await resp.json()
print(resp.status_code, resp.headers, resp.content)

@app.both("/room")
async def room(*args):
    if len(args) == 1:          # HTTP request
        return {"rooms": len(app._rooms)}
    client, data = args         # WebSocket event
    await client.send(f"hi {client.id}")

import asyncio
from lynkio import Lynk, render_template

app = Lynk(protocol="AUTO", serve_client=True, enable_database=True)
app.create_database("chat_logs", create_log_table=True, auto_sync_log=True)

@app.get("/")
async def index(req):
    return render_template("chat.html", {"title": "Lynkio Chat"})

@app.on("join")
async def join(client, data):
    name = data.get("name", "anon")
    client.session["name"] = name
    app.join_room(client.id, "main")
    await app.emit_to_room("main", "system", f"{name} joined")

@app.on("msg")
async def message(client, data):
    await app.emit_to_room("main", "chat", {
        "name": client.session.get("name"),
        "text": data["text"],
    })

@app.task
async def stats():
    while app._running:
        await asyncio.sleep(60)
        print(f"Active connections: {app.connection_count()}")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

#!/usr/bin/env python3
"""
Lynkio Multi-Listener Server
  • public    : HTTP + WebSocket  → 8080
  • admin     : HTTP (X-API-Key)  → 8081
  • internal  : UDP telemetry     → 9091
Serves the built-in JS client at /lynkio/client.js
"""

import asyncio, time, json
from lynkio import (
    Lynk, Listener, json_response, redirect, render_template
)

listeners = [
    Listener("public",   port=8080, protocol="TCP", max_connections=1000),
    Listener("admin",    port=8081, protocol="TCP", max_connections=10),
    Listener("internal", port=9091, protocol="UDP", rate_limit=100),
]

app = Lynk(listeners=listeners, debug=True, serve_client=True)

# ------------------------------------------------------------------
# Middleware
# ------------------------------------------------------------------
@app.http_middleware(listener=None)
async def log_request(req):
    print(f"[{req.listener}] {req.method} {req.path}")
    return None

ADMIN_KEY = "admin123"

@app.http_middleware(listener="admin")
async def admin_auth(req):
    if req.headers.get("x-api-key") != ADMIN_KEY:
        return json_response({"error": "unauthorized"}, 401)
    return None

# ------------------------------------------------------------------
# Public routes
# ------------------------------------------------------------------
@app.get("/", listener="public")
async def index(req):
    return render_template("home.html")

@app.get("/health", listener="public")
async def health(req):
    return {"status": "ok", "timestamp": time.time(),
            "listener": req.listener}

@app.post("/echo", listener="public")
async def echo(req):
    return json_response(await req.json())

# ------------------------------------------------------------------
# Admin routes
# ------------------------------------------------------------------
@app.get("/admin/stats", listener="admin")
async def admin_stats(req):
    return {
        "connections": app.connection_count(),
        "rooms": len(app._rooms),
        "metrics": app.get_metrics(),
    }

# ------------------------------------------------------------------
# WebSocket events (global across listeners)
# ------------------------------------------------------------------
@app.on("join")
async def ws_join(client, data):
    room = data.get("room", "lobby")
    app.join_room(client.id, room)
    await app.emit_to_room(room, "system", "Client joined",
                           exclude=client.id)
    await client.send(json.dumps({
        "event": "connected",
        "data": {"room": room, "listener": client.listener},
    }))

@app.on("chat")
async def ws_chat(client, data):
    room, text = data.get("room"), data.get("text")
    if room and text and app.is_client_in_room(client.id, room):
        await app.emit_to_room(room, "chat", {
            "from": client.id, "text": text, "timestamp": time.time(),
        }, exclude=client.id)

@app.on_binary_event("binary_ping")
async def ws_binary_ping(client, payload):
    print(f"binary ping from {client.id}: {len(payload)} bytes")
    await client.send_binary_event("binary_echo", payload[:100] + b"...")

# ------------------------------------------------------------------
# UDP
# ------------------------------------------------------------------
@app.udp("/telemetry", listener="internal")
async def udp_telemetry(req):
    payload = await req.json()
    print("telemetry:", payload)
    return {"status": "ok", "received": payload}

# ------------------------------------------------------------------
# Background task
# ------------------------------------------------------------------
@app.task
async def stats_printer():
    while app._running:
        await asyncio.sleep(30)
        print(f"Stats: {app.connection_count()} connections, "
              f"{len(app._rooms)} rooms")

if __name__ == "__main__":
    print("=" * 52)
    print("🚀 Lynkio multi-listener server")
    print("=" * 52)
    print("🌐 public   http/ws : http://localhost:8080")
    print("🔒 admin    http    : http://localhost:8081 (X-API-Key: admin123)")
    print("📡 internal udp     : localhost:9091 (path: /telemetry)")
    print("📦 JS client        : /lynkio/client.js")
    print("=" * 52)
    print("WebSocket clients on any listener share the same "
          "global rooms & broadcast registry.")
    app.run()