API: Server-Side Event Stream

[ETSells]

Closes: #23

Adds a method to broadcast received messages to all connected clients rather than requiring polling.

Preliminary performance testing does indicate slightly worse performance than the polling implementation. We should determine the cause of this. This performance hit does not affect the API as long as no SSE clients are connected though.

Library Side: SunDevilRocketry/SDECv2#76

Merge is dependent on the state-estims-telemetry integration branch.

[coderabbitai]

📝 Walkthrough

Walkthrough

Adds a threaded SSE broadcaster that receives telemetry messages, formats and buffers them, and exposes them through a new Flask GET /stream endpoint.

Changes

SSE telemetry streaming

Layer / File(s)	Summary
Stream buffering and broadcast stream.py	Defines SSE metadata and implements synchronized message storage, sequence tracking, event notification, rate limiting, and generator-based broadcasting.
Flask streaming integration app.py	Initializes Stream and Telemetry, forwards telemetry callbacks into the stream, and serves the SSE response from /stream.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Telemetry
  participant Stream
  participant SSEClient
  Telemetry->>Stream: Forward telemetry message
  Stream->>Stream: Format, buffer, and signal update
  Stream-->>SSEClient: Yield SSE event

Loading

Poem

A rabbit hops where data streams,
Through quiet wires and waking dreams.
Each message finds its waiting ear,
Then bounds along, both bright and clear.
SSE carrots for all to cheer!

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	The changes add a publish/subscribe stream, support message types, and push received data to connected clients as required by #23.
Out of Scope Changes check	✅ Passed	The modified files and additions are all directly tied to implementing the SSE broadcast pipeline.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Title check	✅ Passed	The title clearly summarizes the main change: adding a server-side event stream to the API.
Description check	✅ Passed	The description is directly related and correctly describes broadcasting received messages to connected clients.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[ETSells]

Simple test app (vibecoded):

import time
import requests
from requests.exceptions import ChunkedEncodingError, ConnectionError as ReqConnectionError
 
 
def subscribe_raw_sse(url, max_retries=None, base_delay=1, max_delay=30):
    """
    Connect to an SSE endpoint and print events.
    Automatically reconnects with exponential backoff if the connection drops.
 
    max_retries: None = retry forever, or an int cap.
    """
    attempt = 0
 
    while max_retries is None or attempt <= max_retries:
        try:
            headers = {'Accept': 'text/event-stream'}
            response = requests.get(url, stream=True, headers=headers, timeout=(5, None))
            response.raise_for_status()
 
            print(f"Connected to {url}")
            attempt = 0  # reset backoff after a successful connect
 
            for line in response.iter_lines(decode_unicode=True):
                if line:
                    if line.startswith("data:"):
                        data_payload = line[5:].strip()
                        print(f"Data: {data_payload}")
                    elif line.startswith("event:"):
                        event_type = line[6:].strip()
                        print(f"Event: {event_type}")
 
            # If the loop ends without an exception, server closed the stream cleanly.
            print("Stream ended by server (clean close). Reconnecting...")
 
        except (ChunkedEncodingError, ReqConnectionError) as e:
            print(f"Connection dropped ({e.__class__.__name__}). Reconnecting...")
 
        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
 
        except KeyboardInterrupt:
            print("\nStopped by user.")
            return
 
        attempt += 1
        delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
        print(f"Retrying in {delay:.1f}s (attempt {attempt})...")
        time.sleep(delay)
 
    print("Max retries reached. Giving up.")
 
 
if __name__ == "__main__":
    STREAM_URL = "http://localhost:5000/stream"
    subscribe_raw_sse(STREAM_URL)

Results snapshot:
Data: {'origin': 'SDEC-API', 'version': '1.0', 'timestamp': 1062.047, 'messageType': 'VEHICLE_ID', 'payload': {'target': 'Flight Computer (A0002 Rev 2.0)', 'firmware': 'APPA', 'latency': 1062047, 'sig_strength': 0, 'status': 'OK'}}
Data: {'origin': 'SDEC-API', 'version': '1.0', 'timestamp': 1062.157, 'messageType': 'DASHBOARD_DATA', 'payload': {'quat_w': 0.528710663318634, 'quat_x': -0.537405252456665, 'quat_y': -0.5664240717887878, 'quat_z': -0.3329029977321625, 'alt': 384.0848693847656, 'long': 0.0, 'lat': 0.0, 'acc_z': -0.06699321419000626, 'roll_rate': -4.554085731506348}}
Data: {'origin': 'SDEC-API', 'version': '1.0', 'timestamp': 1062.266, 'messageType': 'VEHICLE_ID', 'payload': {'target': 'Flight Computer (A0002 Rev 2.0)', 'firmware': 'APPA', 'latency': 1062266, 'sig_strength': 0, 'status': 'OK'}}
Data: {'origin': 'SDEC-API', 'version': '1.0', 'timestamp': 1062.374, 'messageType': 'DASHBOARD_DATA', 'payload': {'quat_w': 0.5212531685829163, 'quat_x': -0.5389803647994995, 'quat_y': -0.5755689740180969, 'quat_z': -0.32636758685112, 'alt': 384.313720703125, 'long': 0.0, 'lat': 0.0, 'acc_z': 0.0478522926568985, 'roll_rate': -7.66692590713501}}

[coderabbitai]

Actionable comments posted: 3

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@app.py`:
- Around line 176-179: Update the SSE Response returned around
stream_obj.sse_broadcast_callback() to include Cache-Control and
X-Accel-Buffering headers that disable caching and proxy buffering, while
preserving the existing text/event-stream mimetype and streaming behavior.

In `@stream.py`:
- Line 21: Update the Stream implementation to maintain a bounded per-client
queue, such as collections.deque, instead of storing only the latest event in
self.data. Enqueue each published message for the relevant connected client and
have client consumption drain its own queue, preserving every event until the
configured capacity is reached.
- Around line 15-59: Refactor Stream to use a single threading.Condition instead
of signal_broadcast_event and stream_mutex, publishing updates by incrementing
sequence_number and notifying waiting clients without immediately clearing a
signal. In sse_put_callback, serialize the response with json.dumps so the SSE
payload is valid JSON. In sse_broadcast_callback, copy the pending data and
update last_seq_num while holding the condition, then release it before each
yield; wait in a loop for a newer sequence or termination so slow clients cannot
block publishers or lose notifications. Add the json import.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 266c0873-c1d7-46c2-9cf4-0db9f76361a2

📥 Commits

Reviewing files that changed from the base of the PR and between e0b0860 and 9561e4d.

📒 Files selected for processing (2)

• app.py
• stream.py

[ETSells]

The old polling route still sees similar performance drops, but now it happens even without SSE clients connected. We'll have to revisit this as this is not acceptable for the production version of the API. Converting to a draft.

[ETSells]

This may be due to the increased rate on the emulator compared to actual hardware use. Every incoming message is published. We should rate limit this to ~30 Hz.

The client-server contract should include a condition that the client must be able to handle requests at this rate.

The rate limiter should consider using delays to hold the publication until the time delta has expired. This way, it will not react to sent events while it is waiting for that delay to expire.

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@stream.py`:
- Around line 47-82: Fix the rate limiter in the stream generator by using
time.monotonic(), correcting the condition so sleeping occurs only when the
elapsed interval is shorter than RATE_LIMITER_PERIOD, and updating last_time
after each broadcast. Ensure lock release and reacquisition around the sleep are
protected with try/finally so generator closure cannot leave stream_cond in an
invalid state; preserve termination handling and message delivery.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 6f54e8b1-1233-4806-bd75-72027b782fd7

📥 Commits

Reviewing files that changed from the base of the PR and between 9561e4d and bec76e7.

📒 Files selected for processing (2)

• app.py
• stream.py

[ETSells]

Okay, works great now with no performance issues!

[MasterUser43]

json.dumps() running 500Hz is prob not the best performance wise, which is why you were having performance issues.

Probably move it from callback into sse_broadcast_callback s.t. json.dump serializes at broadcast rate instead at 500Hz. I assume within the while not self.terminate.

[MasterUser43]

Also, self.data is single variable, which ig is fine for SDEC, but, at least for LQD-DAQ, I would have to use a queue (deque) to prevent last-message overriding, so it's rather a question if you're willing to lose some data.

Optional, just a note.

[ETSells]

it's not 500 hz, it's like 60-ish for the serial path and 25-ish for LoRa. Given that your rate is upwards of 500 hz though this should def be considered. I'm definitely willing to drop packets since my priority is just making sure the data I'm feeding is the most up to date and I have very little latency. The deque sounds great for you if you're trying to preserve every packet, just make sure you're not flooding the client at more than 60 hz. I personally recommend 30 for the clients' sake, since their refresh rate won't go very high. The one concern I have about this rate limiting is on the rolling pressure chart.

[MasterUser43]

Just random thing, but may a try...finally clause in case you want to use finally: as a way to detect a browser tab closing or to decrement client counter if so choose to consider multiple connections.

Inno, just a neat thing instead of waiting time outs for disconnection.

[ETSells]

The API itself doesn't track client state atm. This is handled behind some layers of abstraction by Flask & Werkzeug. The background processing layer has no idea how many clients are in the foreground. We may change this eventually if we need to throttle too many client connections, but I'm planning for ~4 simultaneous clients right now.

[MasterUser43]

SunDevilRocketry/LQD-DAQ#4

JSON structure

[MasterUser43]

Nothing critical that I see, just optimization prob more relevant to LQD-DAQ than SDEC. Will be using the JSON schema.

Nice job :)

[ETSells]

Will tag this to be pulled in alongside the state estims and telemetry epic! Ready to ship with the parent.