package httpstream
import (
"bufio"
"database/sql"
"fmt"
"io"
)
type rowEncoder func(rows *sql.Rows, w io.Writer) error
func StreamRowsAsJSON(dst io.Writer, rows *sql.Rows, encode rowEncoder) error {
bw := bufio.NewWriterSize(dst, 32*1024)
defer bw.Flush()
if _, err := bw.WriteString("["); err != nil {
return err
}
first := true
for rows.Next() {
if !first {
if _, err := bw.WriteString(","); err != nil {
return err
}
}
first = false
if err := encode(rows, bw); err != nil {
return fmt.Errorf("encode row: %w", err)
}
}
if err := rows.Err(); err != nil {
return fmt.Errorf("row iteration: %w", err)
}
if _, err := bw.WriteString("]"); err != nil {
return err
}
return bw.Flush()
}
package httpstream
import (
"database/sql"
"encoding/json"
"io"
"log"
"net/http"
)
type User struct {
ID int64 `json:"id"`
Email string `json:"email"`
Name string `json:"name"`
}
type UsersHandler struct {
DB *sql.DB
}
func (h *UsersHandler) ListUsers(w http.ResponseWriter, r *http.Request) {
const q = `SELECT id, email, name FROM users ORDER BY id`
rows, err := h.DB.QueryContext(r.Context(), q)
if err != nil {
http.Error(w, "query failed", http.StatusInternalServerError)
return
}
defer rows.Close()
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
var u User
encodeRow := func(rows *sql.Rows, out io.Writer) error {
if err := rows.Scan(&u.ID, &u.Email, &u.Name); err != nil {
return err
}
return json.NewEncoder(out).Encode(u)
}
if err := StreamRowsAsJSON(w, rows, encodeRow); err != nil {
// Response is already committed; can only log the broken stream.
log.Printf("stream users: %v", err)
}
}
Loading an entire result set into a slice before serializing it works fine for small tables but falls apart when a query returns hundreds of thousands of rows: memory usage spikes, garbage collection thrashes, and the client waits for the whole set before receiving a single byte. This snippet shows how to stream rows straight from the database driver to the HTTP response, encoding each row as it is scanned so peak memory stays roughly constant regardless of result size.
In streamer.go, StreamRowsAsJSON accepts an io.Writer, a *sql.Rows, and a rowEncoder callback. It writes the opening [ by hand, then iterates with rows.Next(), calling the encoder for each row and manually inserting , separators between elements. Writing the array brackets and commas directly — rather than building a []interface{} and handing it to json.Marshal — is what keeps the whole set from ever living in memory at once. A bufio.Writer wraps the destination so many small Write calls coalesce into fewer syscalls, and Flush is deferred to push the final buffered bytes. Crucially, rows.Err() is checked after the loop, because rows.Next() returning false can mean either normal completion or a mid-stream failure, and the two must be distinguished.
A subtle correctness point handled here is the header-versus-body problem: once the first bytes are flushed, the HTTP status code is already sent, so an error surfacing halfway through cannot become a clean 500. The code addresses this by only returning an error before any write happens where possible, and otherwise surfacing the failure to the caller for logging while the connection is simply broken.
In users_handler.go, ListUsers issues the query, defers rows.Close(), sets the Content-Type to application/json, and delegates to StreamRowsAsJSON with a closure that scans each User and writes it with a per-row json.Encoder. Reusing scan destinations across iterations avoids per-row allocation churn. Passing r.Context() into QueryContext ensures a client disconnect cancels the underlying database query rather than letting it run to completion pointlessly. This pattern is the go-to when exporting reports, feeding data pipelines, or serving any endpoint whose result size is unbounded and driven by user input.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
# Vulnerable: user input is concatenated directly into SQL.
email = params[:email]
password = params[:password]
sql = "SELECT * FROM users WHERE email = '#{email}' AND password_hash = '#{password}'"
user = ActiveRecord::Base.connection.execute(sql).firstSQL injection prevention with unsafe and safe query patterns
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
