go 55 lines · 10 steps

Streaming NDJSON ingestion in Gin

A Gin handler decodes a newline-delimited JSON stream record by record and flushes to storage in fixed-size batches.

Explained by highlit
Gin Advanced streaming batching json-decoding input-validation backpressure
1type SensorReading struct {
2 DeviceID string `json:"device_id" binding:"required"`
3 Temperature float64 `json:"temperature"`
4 RecordedAt time.Time `json:"recorded_at" binding:"required"`
5}
6 
7func (h *IngestHandler) StreamReadings(c *gin.Context) {
8 if ct := c.ContentType(); ct != "application/x-ndjson" && ct != "application/jsonlines" {
9 c.JSON(http.StatusUnsupportedMediaType, gin.H{"error": "expected newline-delimited JSON"})
10 return
11 }
12 
13 dec := json.NewDecoder(c.Request.Body)
14 dec.DisallowUnknownFields()
15 
16 var (
17 accepted int
18 batch = make([]SensorReading, 0, 256)
19 )
20 
21 for {
22 var r SensorReading
23 if err := dec.Decode(&r); err != nil {
24 if errors.Is(err, io.EOF) {
25 break
26 }
27 c.JSON(http.StatusBadRequest, gin.H{
28 "error": "malformed record",
29 "accepted": accepted,
30 "detail": err.Error(),
31 })
32 return
33 }
34 
35 batch = append(batch, r)
36 accepted++
37 
38 if len(batch) == cap(batch) {
39 if err := h.store.InsertReadings(c.Request.Context(), batch); err != nil {
40 c.JSON(http.StatusInternalServerError, gin.H{"error": "persist failed", "accepted": accepted - len(batch)})
41 return
42 }
43 batch = batch[:0]
44 }
45 }
46 
47 if len(batch) > 0 {
48 if err := h.store.InsertReadings(c.Request.Context(), batch); err != nil {
49 c.JSON(http.StatusInternalServerError, gin.H{"error": "persist failed", "accepted": accepted - len(batch)})
50 return
51 }
52 }
53 
54 c.JSON(http.StatusAccepted, gin.H{"accepted": accepted})
55}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Streaming a decoder over the request body avoids buffering the whole payload in memory.
  2. 2Flushing at a fixed batch capacity bounds memory while keeping database writes efficient.
  3. 3Reporting the accepted count on failure gives clients a precise resume point after partial ingestion.

Related explainers

go 
package theme
 
import (
	"fmt"

Per-tenant HTML templates with Gin's renderer

multi-tenancy concurrency html-templates
Advanced 8 steps
php 
<?php
 
namespace App\Http\Controllers;
 

Streaming a filtered CSV export in Laravel

streaming csv-export query-builder
Intermediate 9 steps
go 
package auth
 
import (
	"net/http"

Setting and reading secure session cookies in Go

cookies session-management security
Intermediate 6 steps
go 
package middleware
 
import (
	"net/http"

A maintenance-mode gate in Gin

middleware atomic-flag concurrency
Intermediate 8 steps
javascript 
const MAX_BATCH_SIZE = 20;
const FLUSH_INTERVAL = 5000;
const ENDPOINT = '/api/analytics';
 

Batching analytics events in the browser

batching buffering sendbeacon
Intermediate 10 steps
go 
func ServeVideo(c *gin.Context) {
	id := c.Param("id")
	video, err := videoRepo.FindByID(c.Request.Context(), id)
	if err != nil {

Streaming video files with Gin

http streaming range requests file serving
Intermediate 6 steps

Share this explainer

Here's the card — post it anywhere.

Streaming NDJSON ingestion in Gin — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code