24 - Streaming and Real-Time Data - Python

32 min read

Streaming and Real-Time Data — Python

Quote

“Turning the database inside out: take the implementation detail that was previously hidden inside the database, and make it a first-class citizen.”

— Martin Kleppmann, Making Sense of Stream Processing (2016)

Technologies Overview

Comparison of the five streaming and transfer protocols used in this notebook — from lowest-latency local TCP to managed cloud services and batch file transfer.

TechnologyProtocolDirectionLatencyUse Case
WebSocketTCP (upgrade from HTTP)Full-duplex (bi-directional)Sub-ms to ~10msLive trading, gaming, collaborative editing
SSE (Server-Sent Events)HTTP/1.1 long-livedServer → Client only~10-50msDashboards, notifications, AI chat streaming
Google Cloud Pub/SubgRPCDecoupled (pub/sub)50-200msEvent-driven pipelines, microservices, IoT
Firestore ListenergRPC (bi-directional stream)Server → Client push100-500msMobile sync, live dashboards, cache invalidation
Batch (GCS file)HTTPSRequest/ResponseSecondsETL, data lake loads, archival

WebSocket opens a persistent TCP connection where both sides can send messages at any time — the lowest-latency option. The protocol starts as an HTTP request (Upgrade: websocket) then switches to a raw binary frame protocol.

SSE is a simpler one-way alternative: the server holds an HTTP connection open and pushes text/event-stream lines. The browser (or client) automatically reconnects on failure. It works through CDNs and proxies that block WebSocket.

Pub/Sub is a managed message bus — publishers and subscribers are fully decoupled. Messages are durably stored until acknowledged. Supports fan-out (one message → many subscribers), dead-letter queues, and exactly-once delivery. Higher latency than direct connections but infinitely more scalable.

Firestore Listener (on_snapshot) uses gRPC bidirectional streaming under the hood. The server pushes document-level change events (ADDED, MODIFIED, REMOVED) as they happen. Built-in offline support and optimistic concurrency — designed for mobile/web apps that need real-time sync without managing connections.

Batch transfer (file upload/download via GCS) is included as a baseline. It has the highest per-message latency but the highest throughput for bulk data — the right choice when freshness is measured in minutes, not milliseconds.

Render the message-flow sequence diagram.


sequenceDiagram
    participant P as Publisher / Client
    participant S as Server / Broker
    participant L as Listener / Subscriber

    rect rgb(41, 46, 66)
    Note over P,L: WebSocket — full-duplex TCP
    P->>S: Upgrade: websocket
    S-->>L: tick stream (sub-ms latency)
    L-->>S: commands / acks
    end

    rect rgb(26, 27, 38)
    Note over P,L: SSE — server → client only
    L->>S: GET /stream (HTTP)
    S-->>L: text/event-stream (~10–50ms)
    end

    rect rgb(41, 46, 66)
    Note over P,L: Pub/Sub — decoupled message bus
    P->>S: Publish message (gRPC)
    Note over S: Durable queue
    S-->>L: StreamingPull delivery (50–200ms)
    L->>S: Ack
    end

    rect rgb(26, 27, 38)
    Note over P,L: Firestore — document change push
    P->>S: Write document (gRPC)
    S-->>L: on_snapshot callback (100–500ms)
    end

Related pattern

For the architectural context of where streaming fits within the broader data platform — including how real-time feeds connect to batch pipelines — see streaming-architecture.

Python Streaming and Real-Time Data Setup

Kernel configuration, package imports, asyncio patching, GCP client initialization, and shared data-generation utilities used across all streaming patterns below.

Setup | Jupyter | imports, asyncio, environment init

Configures the Jupyter kernel and loads all required packages and GCP clients.

Load imports and configure Jupyter async support

All external libraries used throughout the notebook: websockets and aiohttp for local streaming servers, httpx for async HTTP, google-cloud-pubsub and google-cloud-firestore for GCP streaming, and plotly for latency charts. nest_asyncio.apply() patches the event loop to allow asyncio.run() inside Jupyter, which normally forbids nested loops.

Load notebook dependencies in the Jupyter kernel.

import asyncio
import gc
import json
import os
import random
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
 
import aiohttp
from aiohttp import web
import httpx
import websockets
import nest_asyncio
 
import pandas as pd
 
from dotenv import load_dotenv
from google.cloud import pubsub_v1, firestore, storage
 
import plotly.graph_objects as go
from IPython.display import display
 
nest_asyncio.apply()
 
html_formatter = get_ipython().display_formatter.formatters['text/html'] # type: ignore
html_formatter.for_type(pd.DataFrame, lambda df: df.to_html())
_ = html_formatter.for_type(pd.Series, lambda s: s.to_frame().to_html())
 
import logging
logging.getLogger("aiohttp.server").setLevel(logging.CRITICAL)
No output; imports and notebook patching only.

Load environment variables and create GCP clients

Loads .env variables, sets the GOOGLE_APPLICATION_CREDENTIALS path, and creates GCP clients for Pub/Sub, Firestore, and GCS. Publisher batching is disabled (max_messages=1, max_latency=0) to get accurate per-message latency measurements — the default batches up to 10ms, which would distort benchmarks. The final three lines print the active Project ID, Firestore database name, and GCS bucket path for verification.

Initialize the shared GCP clients and notebook settings.

load_dotenv(override=True)
 
PROJECT_ID   = 'seclab-dev-ap-26'
REGION       = 'europe-west1'
BUCKET_NAME  = f'{PROJECT_ID}-data'
FIRESTORE_DB = 'seclab-scores'
SA_KEY_PATH  = os.environ.get('GCP_SA_KEY_PATH', './gcp-sa-key.json')
DATA_DIR     = Path(r'C:\Users\aperi\DEV\LANG\data')
 
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = SA_KEY_PATH
 
from google.cloud.pubsub_v1.types import BatchSettings
publisher    = pubsub_v1.PublisherClient(
    batch_settings=BatchSettings(max_messages=1, max_latency=0))
subscriber   = pubsub_v1.SubscriberClient()
fs_client    = firestore.Client(project=PROJECT_ID, database=FIRESTORE_DB)
gcs_client   = storage.Client(project=PROJECT_ID)
bucket       = gcs_client.bucket(BUCKET_NAME)
 
PROJECT_ID
FIRESTORE_DB
print(f'  GCS:       gs://{BUCKET_NAME}')
  seclab-dev-ap-26
  seclab-scores
  gs://seclab-dev-ap-26-data

Setup | data generation | formatting helpers and OHLCV tick simulation

Shared utility functions and a synthetic OHLCV tick generator used as the data source for all four streaming patterns.

Format time and rate values as human-readable strings

Two utility functions used throughout the notebook. fmt_time converts milliseconds to the most readable unit (µs, ms, s, or min). fmt_rate converts a message count and elapsed duration into a throughput rate (msg/s, K msg/s, or M msg/s).

Define the time and rate formatting helpers.

def fmt_time(ms):
    if ms < 1: return f'{ms*1000:.0f}µs'
    if ms < 1000: return f'{ms:.0f}ms'
    if ms < 60_000: return f'{ms/1000:.1f}s'
    m, s = divmod(ms / 1000, 60)
    if s == 0: return f'{int(m)}min'
    return f'{int(m)}m{s:.0f}s'
 
def fmt_rate(n, ms):
    if ms <= 0: return '-'
    rate = n / (ms / 1000)
    if rate < 1000: return f'{rate:.0f} msg/s'
    if rate < 1_000_000: return f'{rate/1000:.1f}K msg/s'
    return f'{rate/1_000_000:.1f}M msg/s'
No output; helper function definitions only.

Generate simulated OHLCV ticks with random-walk price movement

Produces synthetic tick data for 5 European equity symbols (ASML, SAP, Siemens, LVMH, TotalEnergies). Each call to generate_tick() applies a Gaussian random-walk price change (0.2% standard deviation) and returns a dictionary with symbol, timestamp, price, volume, bid, and ask fields. This function is the data source for all streaming patterns below.

Generate the synthetic OHLCV tick stream.

SYMBOLS = ['ASML.AS', 'SAP.DE', 'SIE.DE', 'MC.PA', 'TTE.PA']
PRICES = {s: random.uniform(50, 900) for s in SYMBOLS}
 
def generate_tick():
    """Generate one simulated tick with random walk price movement."""
    symbol = random.choice(SYMBOLS)
    price = PRICES[symbol]
    change = price * random.gauss(0, 0.002)
    PRICES[symbol] = price + change
    return {
        'symbol': symbol,
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'price': round(PRICES[symbol], 4),
        'volume': random.randint(100, 10_000),
        'bid': round(PRICES[symbol] - random.uniform(0.01, 0.5), 4),
        'ask': round(PRICES[symbol] + random.uniform(0.01, 0.5), 4),
    }
 
sample = [generate_tick() for _ in range(5)]
display(pd.DataFrame(sample))
Sample tick DataFrame rendered inline below.
symboltimestamppricevolumebidask
0SAP.DE2026-03-28T01:42:24.808669+00:00895.42242710895.3121895.6138
1ASML.AS2026-03-28T01:42:24.808669+00:00857.32986499857.2716857.4697
2SIE.DE2026-03-28T01:42:24.808669+00:00625.59575149625.2853625.6701
3SIE.DE2026-03-28T01:42:24.808669+00:00627.20628418626.9840627.4667
4MC.PA2026-03-28T01:42:24.808669+00:00107.82188086107.4289108.1191

WebSocket Streaming

Full-duplex, persistent TCP connection. The server pushes ticks as they occur — no polling. Used by every real-time trading platform (Binance, Bloomberg Terminal, Refinitiv).

WebSocket | websockets | server and client

Local WebSocket server broadcasting simulated ticks, with a client that measures one-way latency using time.perf_counter_ns() embedded in each message.

Start local WebSocket tick server

Starts a local WebSocket server in a background thread that broadcasts ticks at ~100 msg/s. The asyncio.sleep(0) yields control to the event loop between sends without adding delay. The client connects, receives ticks for 3 seconds, and collects them into a DataFrame.

Start the local WebSocket server used for latency measurement.

Scenario: Real-time price dashboards, algorithmic trading, live order book feeds. When NOT to use: One-shot request/response patterns — use REST instead.

Run the WebSocket server benchmark cell.

WS_PORT = 8765
 
async def ws_handler(websocket):
    try:
        while True:
            t = generate_tick()
            t['send_ts'] = time.perf_counter_ns()
            await websocket.send(json.dumps(t))
            await asyncio.sleep(0)
    except websockets.ConnectionClosed:
        pass
 
ws_server = await websockets.serve(ws_handler, 'localhost', WS_PORT)
WS_PORT
  WebSocket server running on ws://localhost:8765

Receive ticks and measure one-way latency

Connects to the local WebSocket server and receives 10,000 ticks (after 100 warmup messages). Each tick carries a send_ts from time.perf_counter_ns(), allowing one-way latency measurement in microseconds without clock synchronization. Garbage collection is disabled during measurement to avoid GC pauses inflating tail latency.

Run the WebSocket client benchmark and collect latency samples.

ws_latencies_us = []
WARMUP_WS = 100
NUM_WS = 10_000
 
async def ws_bench():
    async with websockets.connect(f'ws://localhost:{WS_PORT}') as ws:
        msg_count = 0
        gc_disabled = False
        async for msg in ws:
            msg_count += 1
            if msg_count <= WARMUP_WS:
                continue
            if not gc_disabled:
                gc.disable()
                gc_disabled = True
            recv_ts = time.perf_counter_ns()
            tick = json.loads(msg)
            send_ts = tick['send_ts']
            ws_latencies_us.append((recv_ts - send_ts) / 1000)
            if len(ws_latencies_us) >= NUM_WS:
                break
        if gc_disabled:
            gc.enable()
 
asyncio.run(ws_bench())
ws_server.close()
 
ws_latencies_us.sort()
ws_p50 = ws_latencies_us[len(ws_latencies_us) // 2]
ws_p99 = ws_latencies_us[int(len(ws_latencies_us) * 0.99)]
ws_p999 = ws_latencies_us[int(len(ws_latencies_us) * 0.999)]
print(f'  {len(ws_latencies_us)} one-way measurements')
print(f'  p50: {ws_p50:.0f}µs  p99: {ws_p99:.0f}µs  p99.9: {ws_p999:.0f}µs')
  10000 one-way measurements
  p50: 83µs  p99: 139µs  p99.9: 233µs

Server-Sent Events (SSE)

One-directional server→client push over HTTP. Simpler than WebSocket — works through proxies/CDNs, auto-reconnects, text-only. Used by ChatGPT, GitHub notifications, stock tickers.

SSE | aiohttp + httpx | server and client

Local aiohttp SSE server streaming ticks as text/event-stream, with an httpx async client that parses events and measures one-way latency using the same perf_counter_ns() approach as WebSocket.

Start local SSE tick server

Starts a local aiohttp server that streams ticks as text/event-stream. The asyncio.sleep(0) yields control without delay. The client reads events using httpx async streaming.

Start the local SSE server used for latency measurement.

Scenario: Live dashboards, notification feeds, AI chat token streaming. When NOT to use: Bi-directional communication — use WebSocket. Binary data — use gRPC.

Run the SSE server benchmark cell.

SSE_PORT = 8766
sse_running = True
 
async def sse_handler(request):
    resp = web.StreamResponse()
    resp.content_type = 'text/event-stream'
    resp.headers['Cache-Control'] = 'no-cache'
    resp.headers['Connection'] = 'keep-alive'
    await resp.prepare(request)
    while sse_running:
        t = generate_tick()
        t['send_ts'] = time.perf_counter_ns()
        tick = json.dumps(t)
        await resp.write(f'data: {tick}\n\n'.encode())
        await asyncio.sleep(0)  
    return resp
 
async def start_sse_server():
    app = web.Application()
    app.router.add_get('/ticks', sse_handler)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', SSE_PORT)
    await site.start()
    while sse_running:
        await asyncio.sleep(0.1)
    await runner.cleanup()
 
sse_loop = asyncio.new_event_loop()
sse_thread = threading.Thread(target=lambda: sse_loop.run_until_complete(start_sse_server()), daemon=True)
sse_thread.start()
time.sleep(0.5)
print(f'  SSE server running on http://localhost:{SSE_PORT}/ticks')
  SSE server running on http://localhost:8766/ticks

Receive SSE events and measure one-way latency

Connects to the SSE endpoint and reads 10,000 data: lines (after 100 warmup). Each line is parsed from JSON and the embedded send_ts is compared to a perf_counter_ns() at receive time to compute one-way latency in microseconds. SSE is server→client only, so no echo-based RTT is possible. Warmup messages are skipped before disabling garbage collection to avoid GC pauses inflating tail latency.

Run the SSE client benchmark and collect latency samples.

sse_latencies_us = []
WARMUP_SSE = 100
NUM_SSE = 10_000
 
async def sse_bench():
    async with httpx.AsyncClient() as client:
        async with client.stream('GET', f'http://localhost:{SSE_PORT}/ticks') as resp:
            msg_count = 0
            gc_disabled = False
            async for line in resp.aiter_lines():
                if not line.startswith('data: '):
                    continue
                msg_count += 1
                if msg_count <= WARMUP_SSE:
                    continue
                if not gc_disabled:
                    gc.disable()
                    gc_disabled = True
                recv_ts = time.perf_counter_ns()
                tick = json.loads(line[6:])
                send_ts = tick['send_ts']
                sse_latencies_us.append((recv_ts - send_ts) / 1000)
                if len(sse_latencies_us) >= NUM_SSE:
                    break
            if gc_disabled:
                gc.enable()
 
asyncio.run(sse_bench())
sse_running = False
 
sse_latencies_us.sort()
sse_p50 = sse_latencies_us[len(sse_latencies_us) // 2]
sse_p99 = sse_latencies_us[int(len(sse_latencies_us) * 0.99)]
sse_p999 = sse_latencies_us[int(len(sse_latencies_us) * 0.999)]
print(f'  {len(sse_latencies_us)} one-way measurements')
print(f'  p50: {sse_p50:.0f}µs  p99: {sse_p99:.0f}µs  p99.9: {sse_p999:.0f}µs')
  10000 one-way measurements
  p50: 127µs  p99: 524µs  p99.9: 654µs

Google Cloud Pub/Sub

Managed message bus with at-least-once delivery, auto-scaling, and dead-letter queues. Decouples publishers from subscribers — the backbone of event-driven architectures in GCP. For topic/subscription setup, dead-letter configuration, and operational patterns via gcloud, see pubsub-messaging.

Pub/Sub | google-cloud-pubsub | setup, streaming, latency

Enables the Pub/Sub API, creates a topic and subscription, starts a streaming subscriber, publishes 500 ticks at a steady ~50 msg/s rate, and measures end-to-end delivery latency using a custom send_ts attribute (same-machine clock, no NTP drift).

Enable Pub/Sub API and grant IAM permissions

Enables the Pub/Sub API and grants the roles/pubsub.admin role to the notebook service account. Both operations are idempotent. The output dumps the full IAM policy confirming the binding was added.

Enable Pub/Sub access and print the resulting IAM policy.

!gcloud services enable pubsub.googleapis.com --project=seclab-dev-ap-26
!gcloud projects add-iam-policy-binding seclab-dev-ap-26 --member=serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com --role=roles/pubsub.admin --condition=None --quiet
    bindings:
    - members:
      - serviceAccount:service-922174528852@gcp-sa-artifactregistry.iam.gserviceaccount.com
      role: roles/artifactregistry.serviceAgent
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/bigquery.admin
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/cloudkms.cryptoKeyEncrypterDecrypter
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/cloudsql.admin
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/compute.instanceAdmin.v1
    - members:
      - serviceAccount:922174528852@cloudservices.gserviceaccount.com
      role: roles/compute.instanceGroupManagerServiceAgent
    - members:
      - serviceAccount:service-922174528852@compute-system.iam.gserviceaccount.com
      role: roles/compute.serviceAgent
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/datastore.owner
    - members:
      - serviceAccount:922174528852-compute@developer.gserviceaccount.com
      role: roles/editor
    - members:
      - serviceAccount:service-922174528852@firebase-rules.iam.gserviceaccount.com
      role: roles/firebaserules.system
    - members:
      - serviceAccount:service-922174528852@gcp-sa-firestore.iam.gserviceaccount.com
      role: roles/firestore.serviceAgent
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/iam.serviceAccountTokenCreator
    - members:
      - user:alexper.recovery@gmail.com
      role: roles/owner
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/pubsub.admin
    - members:
      - serviceAccount:service-922174528852@gcp-sa-pubsub.iam.gserviceaccount.com
      role: roles/pubsub.serviceAgent
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/secretmanager.admin
    - members:
      - serviceAccount:notebook-sa@seclab-dev-ap-26.iam.gserviceaccount.com
      role: roles/storage.admin
    etag: BwZOCmEOLbA=
    version: 1
 
    Updated IAM policy for project [seclab-dev-ap-26].

Create topic and subscription

Creates the topic and subscription used for tick streaming. Both operations are idempotent — if the resource already exists, the get_topic/get_subscription call succeeds and creation is skipped. Each resource is first fetched with a get call; if the get raises an exception (not found), the create call runs instead.

Create the Pub/Sub topic and subscription for the benchmark run.

TOPIC_ID = 'tick-feed'
SUB_ID   = 'tick-feed-sub'
topic_path = publisher.topic_path(PROJECT_ID, TOPIC_ID)
sub_path   = subscriber.subscription_path(PROJECT_ID, SUB_ID)
 
try:
    publisher.get_topic(request={'topic': topic_path})
    print(f'  Topic exists: {topic_path}')
except Exception:
    publisher.create_topic(request={'name': topic_path})
    print(f'  Created topic: {topic_path}')
 
try:
    subscriber.get_subscription(request={'subscription': sub_path})
    print(f'  Subscription exists: {sub_path}')
except Exception:
    subscriber.create_subscription(request={'name': sub_path, 'topic': topic_path, 'ack_deadline_seconds': 10})
    print(f'  Created subscription: {sub_path}')
  Topic exists: projects/seclab-dev-ap-26/topics/tick-feed
  Subscription exists: projects/seclab-dev-ap-26/subscriptions/tick-feed-sub

Start streaming subscriber

Starts the subscriber before publishing so the gRPC stream is established when messages arrive. This measures true transport latency, not queue wait time.

Start the streaming subscriber before publishing messages.

Scenario: Event-driven pipelines, microservice communication, IoT telemetry. When NOT to use: Sub-millisecond latency requirements — use direct TCP/WebSocket.

Run the streaming subscriber benchmark cell.

NUM_TICKS = 1000
received = []
pull_lock = threading.Lock()
 
def callback(message):
    receive_time = time.time()
    tick = json.loads(message.data.decode('utf-8'))
    tick['receive_ts'] = receive_time
    with pull_lock:
        received.append(tick)
    message.ack()
 
streaming_pull = subscriber.subscribe(sub_path, callback=callback)
time.sleep(2)
print(f'  Streaming subscriber connected on {sub_path}')
  Streaming subscriber connected on projects/seclab-dev-ap-26/subscriptions/tick-feed-sub

Publish 1000 ticks to topic

Publishes 1000 ticks with wall-clock timestamps. The subscriber callback receives them in real-time.

Publish the measured Pub/Sub workload at a steady rate.

t0 = time.perf_counter()
futures = []
for _ in range(NUM_TICKS):
    tick = generate_tick()
    tick['publish_ts'] = time.time()
    data = json.dumps(tick).encode('utf-8')
    futures.append(publisher.publish(topic_path, data))
for f in futures:
    f.result()
pub_ms = (time.perf_counter() - t0) * 1000
print(f'  Published {NUM_TICKS} ticks in {fmt_time(pub_ms)} ({fmt_rate(NUM_TICKS, pub_ms)})')
  Published 1000 ticks in 737ms (1.4K msg/s)

Measure end-to-end delivery latency

Publishes 500 messages at a steady ~50 msg/s rate (after 50 warmup), measures delivery latency per message. Uses a custom send_ts attribute with same-machine time.time() instead of message.publish_time (which uses Google’s server clock and introduces NTP offset errors). Warmup messages are excluded from the latency measurements. The publish loop sleeps 20ms between sends to maintain a steady ~50 msg/s rate.

Collect the Pub/Sub latency samples from the streaming subscriber.

NUM_PS = 500
WARMUP_PS = 50
ps_latencies_ms = []
ps_lock = threading.Lock()
ps_count = [0]
ps_done = threading.Event()
 
def ps_callback(message):
    recv_time = time.time()
    send_ts = float(message.attributes.get('send_ts', '0'))
    if send_ts > 0:
        latency = (recv_time - send_ts) * 1000
        with ps_lock:
            ps_count[0] += 1
            if ps_count[0] > WARMUP_PS:
                ps_latencies_ms.append(latency)
            if len(ps_latencies_ms) >= NUM_PS:
                ps_done.set()
    message.ack()
 
sub_client = subscriber.subscribe(sub_path, callback=ps_callback)
time.sleep(2)
 
total_msgs = WARMUP_PS + NUM_PS
t0 = time.perf_counter()
for n in range(total_msgs):
    payload = json.dumps(generate_tick()).encode('utf-8')
    publisher.publish(topic_path, payload, send_ts=str(time.time()))
    time.sleep(0.02)
pub_ms = (time.perf_counter() - t0) * 1000
print(f'  Published {total_msgs} messages in {fmt_time(pub_ms)}')
 
ps_done.wait(timeout=60)
sub_client.cancel()
 
ps_latencies_ms.sort()
if ps_latencies_ms:
    avg_latency = sum(ps_latencies_ms) / len(ps_latencies_ms)
    ps_p50 = ps_latencies_ms[len(ps_latencies_ms) // 2]
    ps_p99 = ps_latencies_ms[int(len(ps_latencies_ms) * 0.99)]
    print(f'  {len(ps_latencies_ms)} delivery latency measurements')
    print(f'  p50: {ps_p50:.0f}ms  p99: {ps_p99:.0f}ms  avg: {avg_latency:.0f}ms')
else:
    avg_latency = 0
    print('  No messages received')
  Published 550 messages in 11.5s
  225 delivery latency measurements
  p50: 45ms  p99: 52ms  avg: 45ms

Delete subscription and topic

Deletes the subscription and topic created for the latency benchmark. Both operations are wrapped in try/except — if the resource was already deleted, the error is silently caught.

Clean up the Pub/Sub benchmark resources.

try:
    subscriber.delete_subscription(request={'subscription': sub_path})
    print(f'  Deleted subscription: {sub_path}')
except Exception:
    print(f'  Subscription already deleted')
 
try:
    publisher.delete_topic(request={'topic': topic_path})
    print(f'  Deleted topic: {topic_path}')
except Exception:
    print(f'  Topic already deleted')
  Deleted subscription: projects/seclab-dev-ap-26/subscriptions/tick-feed-sub
  Deleted topic: projects/seclab-dev-ap-26/topics/tick-feed

Firestore Real-Time Listener

Firestore’s on_snapshot pushes document changes to the client in real-time over gRPC. The same mechanism that powers live sync in Firebase mobile apps and dashboards.

Firestore | google-cloud-firestore | listener, writes, cleanup

Registers an on_snapshot callback, writes 550 documents at ~50 doc/s, measures write-to-notification latency, and cleans up. Same measurement pattern as Pub/Sub — custom send_ts field with same-machine clock.

Register real-time listener with on_snapshot

Registers a callback that fires on every document change (ADDED, MODIFIED, REMOVED). Runs in a background thread.

Scenario: Live dashboards, mobile sync, collaborative editing, cache invalidation. When NOT to use: High-throughput ingestion (>1K writes/s) — use Pub/Sub or streaming inserts.

Register the Firestore real-time listener.

FS_RT_COLLECTION = 'realtime_ticks'
NUM_FS = 500
WARMUP_FS = 50
fs_latencies_ms = []
fs_lock = threading.Lock()
fs_count = [0]
fs_done = threading.Event()
 
def on_snapshot(col_snapshot, changes, read_time):
    recv_time = time.time()
    for change in changes:
        if change.type.name == 'ADDED':
            doc = change.document.to_dict()
            send_ts = doc.get('send_ts', 0)
            if send_ts > 0:
                latency = (recv_time - send_ts) * 1000
                with fs_lock:
                    fs_count[0] += 1
                    if fs_count[0] > WARMUP_FS:
                        fs_latencies_ms.append(latency)
                    if len(fs_latencies_ms) >= NUM_FS:
                        fs_done.set()
 
col_ref = fs_client.collection(FS_RT_COLLECTION)
listener = col_ref.on_snapshot(on_snapshot)
time.sleep(1)
print(f'  Listener registered on {FS_RT_COLLECTION}')
  Listener registered on realtime_ticks

Write documents at steady rate

Writes 550 documents (50 warmup + 500 measured) one at a time at a steady ~50 doc/s rate (20ms sleep between writes, matching the Pub/Sub benchmark). Each document carries a send_ts field from the same-machine clock (no NTP drift) for latency measurement. Individual set() calls are used instead of batch writes so each document triggers a separate on_snapshot notification.

Write the Firestore benchmark documents at a steady rate.

total_fs = WARMUP_FS + NUM_FS
t0 = time.perf_counter()
for i in range(total_fs):
    tick = generate_tick()
    tick['send_ts'] = time.time()
    tick['seq'] = i
    col_ref.document(f'tick_{i:04d}').set(tick)
    time.sleep(0.02)
write_ms = (time.perf_counter() - t0) * 1000
print(f'  Wrote {total_fs} documents in {fmt_time(write_ms)}')
  Wrote 550 documents in 30.5s

Measure listener delivery latency

Waits for the background listener to receive all change events, then computes write-to-receive latency per document.

Collect the Firestore listener latency samples.

fs_done.wait(timeout=60)
listener.unsubscribe()
 
fs_latencies_ms.sort()
if fs_latencies_ms:
    fs_avg = sum(fs_latencies_ms) / len(fs_latencies_ms)
    fs_p50 = fs_latencies_ms[len(fs_latencies_ms) // 2]
    fs_p99 = fs_latencies_ms[int(len(fs_latencies_ms) * 0.99)]
    print(f'  {len(fs_latencies_ms)} delivery latency measurements')
    print(f'  p50: {fs_p50:.0f}ms  p99: {fs_p99:.0f}ms  avg: {fs_avg:.0f}ms')
else:
    fs_avg = 0
    print('  No notifications received')
  500 delivery latency measurements
  p50: 44ms  p99: 63ms  avg: 44ms

Delete test documents from real-time collection

Deletes all test documents created during the listener benchmark to leave the collection empty.

Delete the Firestore benchmark documents.

for i in range(WARMUP_FS + NUM_FS):
    fs_client.collection(FS_RT_COLLECTION).document(f'tick_{i:04d}').delete()
print(f'  Deleted {WARMUP_FS + NUM_FS} documents from {FS_RT_COLLECTION}')
  Deleted 550 documents from realtime_ticks

Latency Comparison

Two separate comparisons — local protocols vs GCP managed services — because mixing localhost (0ms network) with cross-continent GCP (300ms RTT) would be meaningless.

Latency | Plotly | local protocols

Both local protocols are sub-millisecond on localhost — network RTT dominates in production. WebSocket is ~1.5x faster at p50; SSE has worse tail latency due to HTTP chunked text parsing overhead.

Local protocols — WebSocket vs SSE throughput (localhost, no network)

Compares p50 one-way latency for both local protocols. WebSocket is ~1.5x faster at p50 — binary frames (2–6 byte header) have less per-message overhead than SSE’s HTTP chunked text encoding. SSE tail latency (p99) is significantly worse due to HTTP line parsing edge cases (partial reads, buffer boundaries) that don’t affect binary WebSocket framing. Both are sub-millisecond on localhost — in production, network RTT dominates. SSE trade-off: works through CDNs/proxies, built-in auto-reconnect, simpler to implement.

Plot the local p50 latency comparison.

local_data = {
    'WebSocket': ws_p50,
    'SSE': sse_p50,
}
 
print(f'  WebSocket: p50={ws_p50:.0f}µs  p99={ws_p99:.0f}µs  ({len(ws_latencies_us)} msgs)')
print(f'  SSE:       p50={sse_p50:.0f}µs  p99={sse_p99:.0f}µs  ({len(sse_latencies_us)} msgs)')
 
fig_local = go.Figure(go.Bar(
    x=list(local_data.keys()), y=list(local_data.values()),
    text=[f'{v:.0f}µs' for v in local_data.values()],
    textposition='outside', cliponaxis=False,
))
fig_local.update_layout(
    title='Local Streaming — p50 One-Way Latency (µs, lower = better)',
    yaxis_title='Latency (µs)',
    template='plotly_dark', height=400,
    margin=dict(t=60),
)
fig_local.show()
  WebSocket: p50=83µs  p99=139µs  (10000 msgs)
  SSE:       p50=127µs  p99=524µs  (10000 msgs)

Latency | Plotly | GCP managed services

Isolates network RTT from protocol overhead by measuring raw gRPC round-trip time to GCP as a baseline, then stacking Pub/Sub and Firestore delivery latency on top. Network RTT dominates ~75% of total latency for both services — both use gRPC to the same GCP region. On a VM in the same region (RTT ≈ 0), expect ~10–15ms pure service overhead.

Measure raw gRPC round-trip time to GCP

Measures raw gRPC RTT to GCP using a minimal Firestore metadata call (50 samples), then compares total delivery latency for Pub/Sub and Firestore against that baseline.

Measure the raw gRPC round-trip baseline to GCP.

rtt_samples = []
for _ in range(50):
    t0 = time.perf_counter_ns()
    list(fs_client.collection('rtt_probe').limit(1).stream())
    t1 = time.perf_counter_ns()
    rtt_samples.append((t1 - t0) / 1000)
 
rtt_samples.sort()
rtt_p50 = rtt_samples[len(rtt_samples) // 2]
rtt_p99 = rtt_samples[int(len(rtt_samples) * 0.99)]
rtt_ms = rtt_p50 / 1000
print(f'  gRPC RTT to GCP (50 samples): p50={rtt_p50/1000:.0f}ms  p99={rtt_p99/1000:.0f}ms')
  gRPC RTT to GCP (50 samples): p50=33ms  p99=49ms

Plot stacked latency breakdown for Pub/Sub and Firestore

Computes protocol overhead by subtracting the raw gRPC RTT baseline from total delivery latency, then renders a stacked bar chart showing the network RTT and protocol overhead components. Total latency labels are added on top of each bar.

Plot the GCP latency breakdown.

ps_overhead_ms = max(0, avg_latency - rtt_ms) if 'avg_latency' in dir() else 0
fs_total_ms = fs_avg if 'fs_avg' in dir() else 0
fs_overhead_ms = max(0, fs_total_ms - rtt_ms)
 
print(f'  Network RTT baseline:  {rtt_ms:.0f}ms')
print(f'  Pub/Sub total:         {avg_latency:.0f}ms  overhead: {ps_overhead_ms:.0f}ms')
print(f'  Firestore total:       {fs_total_ms:.0f}ms  overhead: {fs_overhead_ms:.0f}ms')
 
gcp_labels = ['Pub/Sub', 'Firestore']
gcp_rtt = [rtt_ms, rtt_ms]
gcp_overhead = [ps_overhead_ms, fs_overhead_ms]
 
fig_gcp = go.Figure()
fig_gcp.add_trace(go.Bar(name='Network RTT', x=gcp_labels, y=gcp_rtt,
    marker_color='#555'))
fig_gcp.add_trace(go.Bar(name='Protocol overhead', x=gcp_labels, y=gcp_overhead,
    marker_color='#636EFA'))
totals = [avg_latency, fs_total_ms]
fig_gcp.add_trace(go.Scatter(x=gcp_labels, y=[t + 2 for t in totals],
    text=[f'{t:.0f}ms' for t in totals], mode='text', showlegend=False))
fig_gcp.update_layout(
    title='GCP Managed Services — Latency Breakdown (ms)',
    yaxis_title='Latency (ms)',
    barmode='stack',
    template='plotly_dark', height=450,
    margin=dict(t=60),
)
fig_gcp.show()
  Network RTT baseline:  33ms (p50 of 50 gRPC probes)
  Pub/Sub total:         45ms  overhead: 12ms
  Firestore total:       44ms  overhead: 11ms

Enterprise Transfer & Streaming Patterns

Production patterns for large-scale data movement that go beyond what a notebook can demonstrate. Included as architecture reference — no runnable code.

Enterprise patterns | reference architecture

Architecture reference for large-scale data movement — MFT gateways, GCS Transfer Service, and dedicated interconnect options — with decision guidance for selecting the right pattern.

MFT (Managed File Transfer)

Dedicated gateways that handle large file transfers with multiplexing, packet-level resume, bandwidth routing, encryption, and audit logging. Examples: IBM Sterling, Axway, GoAnywhere.

GCS Transfer Service — scheduled cross-cloud transfers

Managed service for scheduled, recurring transfers between GCS buckets, S3, Azure, or HTTP endpoints. Handles retries, bandwidth throttling, and incremental sync.

Use this service when the transfer itself needs to be treated as an operational workflow rather than a one-off script. It is the right fit for scheduled S3 → GCS replication, multi-day large-volume migrations, and on-premises NAS ingestion where agent pools, managed retries, and transfer-state tracking matter more than hand-built orchestration.

Show the gcloud transfer-job command used for cross-cloud replication.

gcloud transfer jobs create \\
  --source-agent-pool=my-pool \\
  --source=posix:///data/exports \\
  --destination=gs://my-bucket/imports
Command example only; no live output captured.

Transfer Acceleration & Cloud Interconnect

Transfer Acceleration routes uploads through the cloud provider’s edge network (CDN PoPs) instead of the public internet. AWS S3 Transfer Acceleration, GCS has equivalent via CDN. Typical speedup: 2-5x for cross-continent transfers.

Cloud Interconnect / Direct Peering provides dedicated physical network links between your data center and the cloud provider. Consistent bandwidth (10-100 Gbps), lower latency, no public internet routing.

MethodBandwidthLatencyCostUse Case
Public internetVariableHighFreeDev, small transfers
Transfer Acceleration2-5x fasterMediumPer-GB feeCross-continent uploads
Dedicated Interconnect10-100 GbpsLowMonthly + port feeProduction pipelines
Partner Interconnect50 Mbps-50 GbpsLowMonthlySmaller dedicated link

Decision Criteria

Scenario-level guidance that replaces the prose decision table with direct reference blocks.

WebSocket for live price dashboards

Use WebSocket when the client must also send commands or acknowledgments back to the server.

Show the bidirectional live-dashboard selection rule.

def pick_dashboard_transport(client_sends_back: bool) -> str:
    return "WebSocket" if client_sends_back else "SSE"
 
print(pick_dashboard_transport(True))
WebSocket

SSE for AI chat token streaming

Use SSE when the flow is server → client only and the stream benefits from browser auto-reconnect.

Show the server-to-client-only selection rule.

def pick_streaming_transport(server_to_client_only: bool) -> str:
    return "SSE" if server_to_client_only else "WebSocket"
 
print(pick_streaming_transport(True))
SSE

Pub/Sub for event-driven microservices

Use Pub/Sub when publishers and subscribers must stay decoupled and the system needs fan-out, retries, and dead-letter handling.

Show the decoupled messaging selection rule.

def pick_backbone(decoupled_publishers: bool) -> str:
    return "Pub/Sub" if decoupled_publishers else "direct RPC"
 
print(pick_backbone(True))
Pub/Sub

Firestore listener for mobile live sync

Use Firestore listener when clients need document-level push updates without polling or custom connection management.

Show the listener-based live-sync selection rule.

def pick_sync_mode(document_level_push: bool) -> str:
    return "Firestore listener" if document_level_push else "batch polling"
 
print(pick_sync_mode(True))
Firestore listener

GCS + BigQuery load for nightly ETL batch

Use GCS + BigQuery load when freshness is measured in minutes and throughput matters more than per-event latency.

Show the batch-loading selection rule.

def pick_batch_pattern(freshness_minutes: bool) -> str:
    return "GCS + BigQuery load" if freshness_minutes else "streaming ingest"
 
print(pick_batch_pattern(True))
GCS + BigQuery load

Transfer Service for cross-cloud migration

Use Transfer Service when the transfer itself is an operational workflow that needs retries, scheduling, and checkpoints.

Show the managed-transfer selection rule.

def pick_transfer_path(operational_workflow: bool) -> str:
    return "Transfer Service" if operational_workflow else "manual copy"
 
print(pick_transfer_path(True))
Transfer Service

Enterprise MFT for regulated B2B exchange

Use Enterprise MFT when audit trails, encryption, and service-level controls are the main requirements.

Show the regulated-exchange selection rule.

def pick_exchange_mode(regulated_b2b: bool) -> str:
    return "Enterprise MFT" if regulated_b2b else "ad hoc file share"
 
print(pick_exchange_mode(True))
Enterprise MFT

Operational Warnings

asyncio.run() inside a running event loop

Use await directly in notebooks or call nest_asyncio.apply() before running a coroutine entry point.

Show the notebook-safe entry-point choice.

def choose_async_entrypoint(running_loop: bool) -> str:
    return "await main()" if running_loop else "asyncio.run(main())"
 
print(choose_async_entrypoint(True))
await main()

ack_deadline_seconds shorter than processing time

Set ack_deadline_seconds above the worst-case processing time, then extend it for long-running jobs.

Show the deadline-sizing rule for Pub/Sub consumers.

def size_ack_deadline(max_processing_s: int) -> str:
    return f"ack_deadline_seconds >= {int(max_processing_s * 1.5)}"
 
print(size_ack_deadline(45))
ack_deadline_seconds >= 67

Detached on_snapshot() listeners

Store the unsubscribe handle and call it explicitly when the listener is no longer needed.

Show the listener cleanup rule.

def cleanup_listener(has_handle: bool) -> str:
    return "unsubscribe()" if has_handle else "no-op"
 
print(cleanup_listener(True))
unsubscribe()

WebSocket reconnects without back-off

Use exponential back-off with jitter so reconnect storms do not amplify an outage.

Show the reconnect-backoff rule.

def reconnect_strategy(needs_jitter: bool) -> str:
    return "1s, 2s, 4s, ... with jitter" if needs_jitter else "immediate reconnect"
 
print(reconnect_strategy(True))
1s, 2s, 4s, ... with jitter

Enterprise Transfer and Streaming Patterns Recommendations

WebSocket only when the client must also send data back

Prefer SSE for one-way streams because it is simpler and auto-reconnecting.

Show the direction-based transport choice.

def choose_transport(client_sends_back: bool) -> str:
    return "WebSocket" if client_sends_back else "SSE"
 
print(choose_transport(False))
SSE

Pin google-cloud-pubsub in requirements.txt

Pin the dependency so transport and retry defaults do not drift across releases.

Show the pinned-dependency rule.

print("requirements.txt: google-cloud-pubsub==<pinned-version>")
requirements.txt: google-cloud-pubsub==<pinned-version>

Use StreamingPullFuture above ~10 messages/second

Use streaming pull for sustained throughput; synchronous pull adds avoidable round-trips.

Show the streaming-pull preference.

print("subscriber.subscribe(..., flow_control=FlowControl(...))")
subscriber.subscribe(..., flow_control=FlowControl(...))

Set max_messages on FlowControl

Cap in-flight messages to the amount the consumer can process without buffering spikes.

Show the flow-control cap.

print("FlowControl(max_messages=N)")
FlowControl(max_messages=N)

Prefer SSE for read-only dashboards

Use SSE when the UI only needs server push and network boundaries can block WebSocket upgrades.

Show the read-only dashboard preference.

print("SSE")
SSE

Log every Pub/Sub nack

Log each negative acknowledgment so dead-letter analysis remains traceable after the fact.

Show the logging rule for negative acknowledgments.

print('logger.warning("pubsub nack message_id=...")')
logger.warning("pubsub nack message_id=...")

Test Firestore callbacks under network partition conditions

Validate idempotency before production so offline replay does not duplicate side effects.

Show the replay-safe callback rule.

print("idempotent callback + offline replay check")
idempotent callback + offline replay check

Use gcloud transfer jobs create above 1 TB

Use the managed transfer service when checkpointing, retries, and audit logs matter more than ad hoc scripts.

Show the managed-transfer command choice.

print("gcloud transfer jobs create ...")
gcloud transfer jobs create ...

Enterprise Transfer and Streaming Patterns Troubleshooting

websockets.exceptions.ConnectionClosedError on connect

Check the URL scheme and confirm that the server is reachable on the expected port.

Show the connection-check rule.

print("verify ws:// vs wss:// and port reachability")
verify ws:// vs wss:// and port reachability

Pub/Sub subscriber receives every message twice

Extend the ack deadline and deduplicate on a stable message key or attribute.

Show the duplicate-delivery fix.

print("extend ack_deadline_seconds; deduplicate on message ID")
extend ack_deadline_seconds; deduplicate on message ID

Firestore on_snapshot never fires

Verify the read permissions and confirm the collection path is correct.

Show the listener-permission check.

print("roles/datastore.user and the collection path")
roles/datastore.user and the collection path

SSE disconnects every 30 seconds

Raise the proxy timeout and send a keepalive comment before the connection idles out.

Show the SSE timeout fix.

print("proxy_read_timeout 3600s; : keepalive\\n\\n")
proxy_read_timeout 3600s; : keepalive\n\n

google.api_core.exceptions.DeadlineExceeded on Pub/Sub publish

Increase the publisher timeout and verify network and service health.

Show the publish-timeout fix.

print("increase timeout in PublisherOptions; check network health")
increase timeout in PublisherOptions; check network health

WebSocket server accepts connections but sends no data

Await each coroutine and move blocking CPU work off the event loop.

Show the missing-await fix.

print("await every coroutine; move blocking work to executor")
await every coroutine; move blocking work to executor

DefaultCredentialsError when running locally

Configure Application Default Credentials before running Pub/Sub or Firestore code.

Show the local-auth fix.

print("gcloud auth application-default login")
gcloud auth application-default login

GCS Transfer Service job shows FAILED with no transferred objects

Grant the source bucket viewer role to the Transfer Service agent account.

Show the transfer-service permission fix.

print("grant roles/storage.objectViewer to the transfer service agent")
grant roles/storage.objectViewer to the transfer service agent

Graph View

Backlinks