Working with Advanced Streaming Patterns
1. Implementing Batch Streaming
| Strategy | Detail |
|---|---|
| Batch size | 50–500 items per message; tune for latency vs CPU |
| Time + size triggers | Flush on whichever hits first |
| Bound message size | Stay under MaxRecvMsgSize |
Example: Batched server stream
batch := make([]*Event, 0, 100)
for e := range src {
batch = append(batch, e)
if len(batch) == cap(batch) {
_ = stream.Send(&EventBatch{Events: batch})
batch = batch[:0]
}
}2. Using Chunked File Upload
| Field | Purpose |
|---|---|
| First message | Metadata (filename, mime, size) |
| Subsequent | bytes data = 1; chunks (e.g. 64 KB) |
| Final | Server returns checksum + URI |
3. Using Chunked File Download
| Pattern | Detail |
|---|---|
| Server stream | Send fixed-size chunks |
| Range support | Include offset in request |
| Resume | Client tracks bytes received and resends offset on reconnect |
4. Implementing Stream Multiplexing
| Approach | Detail |
|---|---|
| Tag each message | uint64 stream_id field |
| Dispatch | Route by id on receive |
| Alternative | Use multiple gRPC streams over same connection (built-in HTTP/2 mux) |
5. Using Flow Control
| Mechanism | Detail |
|---|---|
| HTTP/2 window | Default 64 KB per stream; tuned via InitialWindowSize |
InitialWindowSize / InitialConnWindowSize | Server option; raise to 4 MB for high-throughput streams |
| Slow consumer | Send blocks — natural backpressure |
6. Implementing Stream Cancellation
Example: Cancel from client
ctx, cancel := context.WithCancel(context.Background())
stream, _ := client.Subscribe(ctx, req)
go func() { time.Sleep(5*time.Second); cancel() }()| Side | Detail |
|---|---|
| Client | cancel() ctx |
| Server sees | stream.Context().Err() == context.Canceled |
7. Using Stream Heartbeats
| Approach | Detail |
|---|---|
| Application heartbeat | Send empty / dedicated Heartbeat message every N seconds |
| HTTP/2 keep-alive | Detect dead TCP; not application-level |
| Detect liveness | Reset timer on each received message |
8. Implementing Stream Reconnection
Example: Reconnect with backoff
backoff := time.Second
for {
stream, err := client.Subscribe(ctx, &Req{LastSeq: last})
if err == nil {
last, err = consume(stream)
if ctx.Err() != nil { return }
}
time.Sleep(backoff)
if backoff < 30*time.Second { backoff *= 2 }
}9. Handling Partial Failures
| Pattern | Detail |
|---|---|
| Per-item status | Response message includes repeated ItemStatus |
| Continue on error | Don't fail entire stream for one bad item |
| Trailer summary | Counts of OK/failed via trailer metadata |
10. Using Stream Buffering
| Buffer Type | Use |
|---|---|
| Channel buffer (Go) | Decouple producer/consumer |
| Bounded queue | Drop or block on overflow |
| Disk spill | Very large or critical streams |