microdot.sse --- Server-Sent Events

Server-Sent Events (SSE) เป็นโปรโตคอล push ทางเดียวแบบง่าย: client เปิด HTTP request ปกติไปยัง Content-Type: text/event-stream และเซิร์ฟเวอร์เปิดการเชื่อมต่อไว้และเขียน event ที่จัดรูปแบบเมื่อเกิดขึ้น เบราว์เซอร์เปิดเผย stream เป็น JavaScript EventSource object เมื่อเทียบกับ WebSocket SSE ง่ายกว่าเมื่อเซิร์ฟเวอร์เท่านั้นที่ push; client ร้องขอเสมอ

การใช้งานทั่วไปสำหรับกล้องคือการเผยแพร่การตรวจจับ / การอ่านค่าจาก sensor / การอัปเดตสถานะไปยังโทรศัพท์หรือ dashboard ตามอัตราที่เกิดขึ้น โดยไม่ต้องให้ dashboard ทำ poll

class SSE

class microdot.sse.SSE

handle ที่ route ได้รับ ส่งไปยัง handler เป็นอาร์กิวเมนต์ที่สองเมื่อ with_sse() ถูกใช้งาน

async send(data, event: str | None = None, event_id: str | None = None, retry: int | None = None, comment: bool = False)

Push event เดียวไปยัง client

data

Event payload String และ bytes ส่งตรง Dict และ list ถูก serialize เป็น JSON สิ่งอื่นถูกแปลงด้วย str()

event

ชื่อ event ที่ไม่บังคับ -- EventSource ของเบราว์เซอร์ dispatch ไปยัง JavaScript listener ที่มีชื่อเดียวกันเมื่อตั้งค่านี้

event_id

Event id ที่ไม่บังคับ -- เบราว์เซอร์ส่งกลับเป็น Last-Event-ID หากการเชื่อมต่อหลุดและเชื่อมต่อใหม่ ซึ่งช่วยให้เซิร์ฟเวอร์ต่อจากที่ค้างไว้ได้

retry

หน่วงเวลาเชื่อมต่อใหม่เป็นวินาที เบราว์เซอร์ใช้สิ่งนี้หากการเชื่อมต่อหลุด

comment

หากเป็น True payload จะถูกส่งเป็น SSE comment line (ถูกเบราว์เซอร์ละเว้น) มีประโยชน์เป็น keep-alive heartbeat เพื่อหยุด NAT timeout ไม่ให้ปิด stream ที่ไม่มีการใช้งาน

Module-level decorator

microdot.sse.with_sse(f)

Decorator ที่แปลง route เป็น SSE endpoint handler ได้รับ request และ object SSE

from microdot import Microdot
from microdot.sse import with_sse

app = Microdot()

@app.get('/events')
@with_sse
async def events(request, sse):
    import asyncio
    while True:
        await sse.send({'temp': read_sensor()}, event='reading')
        await asyncio.sleep(1)

อายุของ handler เท่ากับอายุของ stream -- ทำงานตราบเท่าที่ client ยังเชื่อมต่ออยู่ การยกเลิก (client ตัดการเชื่อมต่อ, เซิร์ฟเวอร์ปิด) เผยแพร่เป็น asyncio.CancelledError ซึ่งเฟรมเวิร์กกลืนทิ้ง

microdot.sse.sse_response(request, event_function, *args, **kwargs)

entry point ระดับต่ำ: คืน response tuple ที่ with_sse() คืนภายใน มีประโยชน์เมื่อ SSE endpoint ต้องการ header กำหนดเองหรือ status code ที่ decorator ไม่อนุญาต:

from microdot.sse import sse_response

@app.get('/events')
async def events(request):
    if not request.g.current_user:
        return 401
    async def emit(req, sse):
        await sse.send('hello')
    return sse_response(request, emit)

event_function รับ (request, sse, *args, **kwargs) และใช้ await sse.send(...) เพื่อ push event

SSE response ตั้ง Content-Type: text/event-stream และเปิดการเชื่อมต่อไว้จนกว่า event_function จะคืนค่าหรือ client ตัดการเชื่อมต่อ ใช้ while True loop กับ await asyncio.sleep ระหว่างการส่งเพื่อหลีกเลี่ยง hot-loop