mirror of
https://github.com/azaion/gps-denied-desktop.git
synced 2026-04-23 00:46:37 +00:00
initial structure implemented
docs -> _docs
This commit is contained in:
Oleksandr Bezdieniezhnykh
+61
@@ -0,0 +1,61 @@
|
||||
# Feature: Connection Lifecycle Management
|
||||
|
||||
## Description
|
||||
|
||||
Manages the lifecycle of SSE (Server-Sent Events) connections including creation, tracking, health monitoring, and graceful closure. Supports multiple concurrent client connections per flight with proper resource management.
|
||||
|
||||
## Component APIs Implemented
|
||||
|
||||
### `create_stream(flight_id: str, client_id: str) -> StreamConnection`
|
||||
Establishes a new SSE connection for a client subscribing to a flight's events.
|
||||
|
||||
### `close_stream(flight_id: str, client_id: str) -> bool`
|
||||
Closes an existing SSE connection and cleans up associated resources.
|
||||
|
||||
### `get_active_connections(flight_id: str) -> int`
|
||||
Returns the count of active SSE connections for a given flight.
|
||||
|
||||
## External Tools and Services
|
||||
|
||||
- **FastAPI/Starlette**: SSE response streaming support
|
||||
- **asyncio**: Asynchronous connection handling and concurrent stream management
|
||||
|
||||
## Internal Methods
|
||||
|
||||
| Method | Purpose |
|
||||
|--------|---------|
|
||||
| `_register_connection(flight_id, client_id, stream)` | Adds connection to internal registry |
|
||||
| `_unregister_connection(flight_id, client_id)` | Removes connection from registry |
|
||||
| `_get_connections_for_flight(flight_id)` | Returns all active streams for a flight |
|
||||
| `_generate_stream_id()` | Creates unique identifier for stream |
|
||||
| `_handle_client_disconnect(flight_id, client_id)` | Cleanup on unexpected disconnect |
|
||||
|
||||
## Data Structures
|
||||
|
||||
```python
|
||||
# Connection registry: Dict[flight_id, Dict[client_id, StreamConnection]]
|
||||
# Allows O(1) lookup by flight and client
|
||||
```
|
||||
|
||||
## Unit Tests
|
||||
|
||||
| Test | Description |
|
||||
|------|-------------|
|
||||
| `test_create_stream_returns_valid_connection` | Verify StreamConnection has all required fields |
|
||||
| `test_create_stream_unique_stream_ids` | Multiple streams get unique identifiers |
|
||||
| `test_close_stream_removes_from_registry` | Connection no longer tracked after close |
|
||||
| `test_close_stream_nonexistent_returns_false` | Graceful handling of invalid close |
|
||||
| `test_get_active_connections_empty` | Returns 0 when no connections exist |
|
||||
| `test_get_active_connections_multiple_clients` | Correctly counts concurrent connections |
|
||||
| `test_multiple_flights_isolated` | Connections for different flights don't interfere |
|
||||
| `test_same_client_reconnect` | Client can reconnect with same client_id |
|
||||
|
||||
## Integration Tests
|
||||
|
||||
| Test | Description |
|
||||
|------|-------------|
|
||||
| `test_concurrent_client_connections` | 10 clients connect simultaneously to same flight |
|
||||
| `test_connection_survives_idle_period` | Connection remains open during inactivity |
|
||||
| `test_client_disconnect_detection` | System detects when client drops connection |
|
||||
| `test_connection_cleanup_on_flight_completion` | All connections closed when flight ends |
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
# Feature: Event Broadcasting
|
||||
|
||||
## Description
|
||||
|
||||
Broadcasts various event types to all connected clients for a flight. Handles event formatting per SSE protocol, buffering events for temporarily disconnected clients, and replay of missed events on reconnection. Includes heartbeat mechanism for connection health.
|
||||
|
||||
## Component APIs Implemented
|
||||
|
||||
### `send_frame_result(flight_id: str, frame_result: FrameResult) -> bool`
|
||||
Broadcasts `frame_processed` event with GPS coordinates, confidence, and metadata.
|
||||
|
||||
### `send_search_progress(flight_id: str, search_status: SearchStatus) -> bool`
|
||||
Broadcasts `search_expanded` event during failure recovery search operations.
|
||||
|
||||
### `send_user_input_request(flight_id: str, request: UserInputRequest) -> bool`
|
||||
Broadcasts `user_input_needed` event when processing requires user intervention.
|
||||
|
||||
### `send_refinement(flight_id: str, frame_id: int, updated_result: FrameResult) -> bool`
|
||||
Broadcasts `frame_refined` event when factor graph optimization improves a position.
|
||||
|
||||
### `send_heartbeat(flight_id: str) -> bool`
|
||||
Sends keepalive ping to all clients to maintain connection health.
|
||||
|
||||
## External Tools and Services
|
||||
|
||||
- **FastAPI/Starlette**: SSE response streaming
|
||||
- **asyncio**: Async event delivery to multiple clients
|
||||
- **JSON**: Event data serialization
|
||||
|
||||
## Internal Methods
|
||||
|
||||
| Method | Purpose |
|
||||
|--------|---------|
|
||||
| `_format_sse_event(event_type, event_id, data)` | Formats data per SSE protocol spec |
|
||||
| `_broadcast_to_flight(flight_id, event)` | Sends event to all clients of a flight |
|
||||
| `_buffer_event(flight_id, event)` | Stores event for replay to reconnecting clients |
|
||||
| `_get_buffered_events(flight_id, last_event_id)` | Retrieves events after given ID for replay |
|
||||
| `_prune_event_buffer(flight_id)` | Removes old events beyond retention window |
|
||||
| `_generate_event_id(frame_id, event_type)` | Creates sequential event ID for ordering |
|
||||
|
||||
## Event Types
|
||||
|
||||
| Event | Format |
|
||||
|-------|--------|
|
||||
| `frame_processed` | `{frame_id, gps, altitude, confidence, heading, timestamp}` |
|
||||
| `frame_refined` | `{frame_id, gps, refined: true}` |
|
||||
| `search_expanded` | `{frame_id, grid_size, status}` |
|
||||
| `user_input_needed` | `{request_id, frame_id, candidate_tiles}` |
|
||||
| `:heartbeat` | SSE comment (no data payload) |
|
||||
|
||||
## Buffering Strategy
|
||||
|
||||
- Events buffered per flight with configurable retention (default: 1000 events or 5 minutes)
|
||||
- Events keyed by sequential ID for replay ordering
|
||||
- On reconnection with `last_event_id`, replay all events after that ID
|
||||
|
||||
## Unit Tests
|
||||
|
||||
| Test | Description |
|
||||
|------|-------------|
|
||||
| `test_send_frame_result_formats_correctly` | Event matches expected JSON structure |
|
||||
| `test_send_refinement_includes_refined_flag` | Refined events marked appropriately |
|
||||
| `test_send_search_progress_valid_status` | Status field correctly populated |
|
||||
| `test_send_user_input_request_has_request_id` | Request includes unique identifier |
|
||||
| `test_send_heartbeat_sse_comment_format` | Heartbeat uses `:heartbeat` comment format |
|
||||
| `test_buffer_event_stores_in_order` | Events retrievable in sequence |
|
||||
| `test_buffer_pruning_removes_old_events` | Buffer doesn't grow unbounded |
|
||||
| `test_replay_from_last_event_id` | Correct subset returned for replay |
|
||||
| `test_broadcast_returns_false_no_connections` | Graceful handling when no clients |
|
||||
| `test_event_id_generation_sequential` | IDs increase monotonically |
|
||||
|
||||
## Integration Tests
|
||||
|
||||
| Test | Description |
|
||||
|------|-------------|
|
||||
| `test_100_frames_all_received_in_order` | Stream 100 frame results, verify completeness |
|
||||
| `test_reconnection_replay` | Disconnect after 50 events, reconnect, receive 51-100 |
|
||||
| `test_multiple_event_types_interleaved` | Mix of frame_processed and frame_refined events |
|
||||
| `test_user_input_flow_roundtrip` | Send request, verify client receives |
|
||||
| `test_heartbeat_every_30_seconds` | Verify keepalive timing |
|
||||
| `test_event_latency_under_500ms` | Measure generation-to-receipt time |
|
||||
| `test_high_throughput_100_events_per_second` | Sustained event rate handling |
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
# SSE Event Streamer
|
||||
|
||||
## Interface Definition
|
||||
|
||||
**Interface Name**: `ISSEEventStreamer`
|
||||
|
||||
### Interface Methods
|
||||
|
||||
```python
|
||||
class ISSEEventStreamer(ABC):
|
||||
@abstractmethod
|
||||
def create_stream(self, flight_id: str, client_id: str) -> StreamConnection:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def send_frame_result(self, flight_id: str, frame_result: FrameResult) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def send_search_progress(self, flight_id: str, search_status: SearchStatus) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def send_user_input_request(self, flight_id: str, request: UserInputRequest) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def send_refinement(self, flight_id: str, frame_id: int, updated_result: FrameResult) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def send_heartbeat(self, flight_id: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def close_stream(self, flight_id: str, client_id: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_active_connections(self, flight_id: str) -> int:
|
||||
pass
|
||||
```
|
||||
|
||||
## Component Description
|
||||
|
||||
### Responsibilities
|
||||
- Real-time communication with clients.
|
||||
- Buffering events for disconnected clients.
|
||||
|
||||
### Callers
|
||||
- **F02.1 Flight Lifecycle Manager**: Calls `create_stream` (delegated from F01).
|
||||
- **F02.2 Flight Processing Engine**: Calls `send_user_input_request`, `send_search_progress`.
|
||||
- **F14 Result Manager**: Calls `send_frame_result`, `send_refinement`.
|
||||
|
||||
### Consistency Fix
|
||||
- Previously listed F11 as caller. **Correction**: F11 returns request objects to F02.2. **F02.2 is the sole caller** for user input requests and search status updates. F11 has no dependencies on F15.
|
||||
|
||||
### Scope
|
||||
- SSE protocol implementation
|
||||
- Event formatting and sending
|
||||
- Connection management
|
||||
- Client reconnection handling
|
||||
- Multiple concurrent streams per flight
|
||||
|
||||
## API Methods
|
||||
|
||||
### `create_stream(flight_id: str, client_id: str) -> StreamConnection`
|
||||
|
||||
**Description**: Establishes a server-sent events connection.
|
||||
|
||||
**Called By**: F01 REST API (GET /stream endpoint)
|
||||
|
||||
**Output**:
|
||||
```python
|
||||
StreamConnection:
|
||||
stream_id: str
|
||||
flight_id: str
|
||||
client_id: str
|
||||
last_event_id: Optional[str]
|
||||
```
|
||||
|
||||
**Event Types**:
|
||||
- `frame_processed`
|
||||
- `frame_refined`
|
||||
- `search_expanded`
|
||||
- `user_input_needed`
|
||||
- `processing_blocked`
|
||||
- `route_api_updated`
|
||||
- `route_completed`
|
||||
|
||||
**Test Cases**:
|
||||
1. Create stream → client receives keepalive pings
|
||||
2. Multiple clients → each gets own stream
|
||||
|
||||
---
|
||||
|
||||
### `send_frame_result(flight_id: str, frame_result: FrameResult) -> bool`
|
||||
|
||||
**Description**: Sends frame_processed event.
|
||||
|
||||
**Called By**: F14 Result Manager
|
||||
|
||||
**Event Format**:
|
||||
```json
|
||||
{
|
||||
"event": "frame_processed",
|
||||
"id": "frame_237",
|
||||
"data": {
|
||||
"frame_id": 237,
|
||||
"gps": {"lat": 48.123, "lon": 37.456},
|
||||
"altitude": 800.0,
|
||||
"confidence": 0.95,
|
||||
"heading": 87.3,
|
||||
"timestamp": "2025-11-24T10:30:00Z"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Test Cases**:
|
||||
1. Send event → all clients receive
|
||||
2. Client disconnected → event buffered for replay
|
||||
|
||||
---
|
||||
|
||||
### `send_search_progress(flight_id: str, search_status: SearchStatus) -> bool`
|
||||
|
||||
**Description**: Sends search_expanded event.
|
||||
|
||||
**Called By**: F02.2 Flight Processing Engine (via F11 status return)
|
||||
|
||||
**Event Format**:
|
||||
```json
|
||||
{
|
||||
"event": "search_expanded",
|
||||
"data": {
|
||||
"frame_id": 237,
|
||||
"grid_size": 9,
|
||||
"status": "searching"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `send_user_input_request(flight_id: str, request: UserInputRequest) -> bool`
|
||||
|
||||
**Description**: Sends user_input_needed event.
|
||||
|
||||
**Called By**: F02.2 Flight Processing Engine (via F11 request object)
|
||||
|
||||
**Event Format**:
|
||||
```json
|
||||
{
|
||||
"event": "user_input_needed",
|
||||
"data": {
|
||||
"request_id": "uuid",
|
||||
"frame_id": 237,
|
||||
"candidate_tiles": [...]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `send_refinement(flight_id: str, frame_id: int, updated_result: FrameResult) -> bool`
|
||||
|
||||
**Description**: Sends frame_refined event.
|
||||
|
||||
**Called By**: F14 Result Manager
|
||||
|
||||
**Event Format**:
|
||||
```json
|
||||
{
|
||||
"event": "frame_refined",
|
||||
"data": {
|
||||
"frame_id": 237,
|
||||
"gps": {"lat": 48.1235, "lon": 37.4562},
|
||||
"refined": true
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### `send_heartbeat(flight_id: str) -> bool`
|
||||
|
||||
**Description**: Sends heartbeat/keepalive to all clients subscribed to flight.
|
||||
|
||||
**Called By**:
|
||||
- Background heartbeat task (every 30 seconds)
|
||||
- F02.2 Flight Processing Engine (periodically during processing)
|
||||
|
||||
**Event Format**:
|
||||
```
|
||||
:heartbeat
|
||||
```
|
||||
|
||||
**Behavior**:
|
||||
- Sends SSE comment (`:heartbeat`) which doesn't trigger event handlers
|
||||
- Keeps TCP connection alive
|
||||
- Client can use to detect connection health
|
||||
|
||||
**Test Cases**:
|
||||
1. Send heartbeat → all clients receive ping
|
||||
2. Client timeout → connection marked stale
|
||||
|
||||
---
|
||||
|
||||
### `close_stream(flight_id: str, client_id: str) -> bool`
|
||||
|
||||
**Description**: Closes SSE connection.
|
||||
|
||||
**Called By**: F01 REST API (on client disconnect)
|
||||
|
||||
---
|
||||
|
||||
### `get_active_connections(flight_id: str) -> int`
|
||||
|
||||
**Description**: Returns count of active SSE connections for a flight.
|
||||
|
||||
**Called By**:
|
||||
- F02.2 Flight Processing Engine (monitoring)
|
||||
- Admin tools
|
||||
|
||||
**Test Cases**:
|
||||
1. No connections → returns 0
|
||||
2. 5 clients connected → returns 5
|
||||
|
||||
## Integration Tests
|
||||
|
||||
### Test 1: Real-Time Streaming
|
||||
1. Client connects
|
||||
2. Process 100 frames
|
||||
3. Client receives 100 frame_processed events
|
||||
4. Verify order and completeness
|
||||
|
||||
### Test 2: Reconnection with Replay
|
||||
1. Client connects
|
||||
2. Process 50 frames
|
||||
3. Client disconnects
|
||||
4. Process 50 more frames
|
||||
5. Client reconnects with last_event_id
|
||||
6. Client receives frames 51-100
|
||||
|
||||
### Test 3: User Input Flow
|
||||
1. Processing blocks
|
||||
2. send_user_input_request()
|
||||
3. Client receives event
|
||||
4. Client responds with fix
|
||||
5. Processing resumes
|
||||
|
||||
## Non-Functional Requirements
|
||||
|
||||
### Performance
|
||||
- **Event latency**: < 500ms from generation to client
|
||||
- **Throughput**: 100 events/second
|
||||
- **Concurrent connections**: 1000+ clients
|
||||
|
||||
### Reliability
|
||||
- Event buffering for disconnected clients
|
||||
- Automatic reconnection support
|
||||
- Keepalive pings every 30 seconds
|
||||
|
||||
## Dependencies
|
||||
|
||||
### Internal Components
|
||||
- None (receives calls from other components)
|
||||
|
||||
### External Dependencies
|
||||
- **FastAPI** or **Flask** SSE support
|
||||
- **asyncio**: For async event streaming
|
||||
|
||||
## Data Models
|
||||
|
||||
### StreamConnection
|
||||
```python
|
||||
class StreamConnection(BaseModel):
|
||||
stream_id: str
|
||||
flight_id: str
|
||||
client_id: str
|
||||
created_at: datetime
|
||||
last_event_id: Optional[str]
|
||||
```
|
||||
|
||||
### SSEEvent
|
||||
```python
|
||||
class SSEEvent(BaseModel):
|
||||
event: str
|
||||
id: Optional[str]
|
||||
data: Dict[str, Any]
|
||||
```
|
||||
Reference in New Issue
Block a user