java 39 lines · 9 steps

Streaming market quotes with Spring WebFlux

A reactive controller pushes live price ticks over server-sent events with backpressure and timeout safeguards.

Explained by highlit
Spring Advanced reactive-streams backpressure server-sent-events flux error-handling
1@RestController
2@RequestMapping("/api/quotes")
3public class QuoteStreamController {
4 
5 private final QuoteRepository quoteRepository;
6 private final MarketFeedClient marketFeedClient;
7 
8 public QuoteStreamController(QuoteRepository quoteRepository, MarketFeedClient marketFeedClient) {
9 this.quoteRepository = quoteRepository;
10 this.marketFeedClient = marketFeedClient;
11 }
12 
13 @GetMapping(value = "/{symbol}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
14 public Flux<QuoteEvent> streamQuotes(@PathVariable String symbol,
15 @RequestParam(defaultValue = "25") int maxPerSecond) {
16 return marketFeedClient.subscribe(symbol)
17 .filter(tick -> tick.volume() > 0)
18 .onBackpressureBuffer(512,
19 dropped -> log.warn("Dropped stale tick for {} at {}", symbol, dropped.timestamp()),
20 BufferOverflowStrategy.DROP_OLDEST)
21 .limitRate(maxPerSecond)
22 .map(tick -> new QuoteEvent(tick.symbol(), tick.price(), tick.timestamp()))
23 .timeout(Duration.ofSeconds(30))
24 .doOnCancel(() -> log.info("Client cancelled stream for {}", symbol))
25 .onErrorResume(TimeoutException.class,
26 ex -> Flux.error(new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, "feed idle")));
27 }
28 
29 @GetMapping
30 public Flux<QuoteEvent> recentQuotes(@RequestParam List<String> symbols) {
31 return Flux.fromIterable(symbols)
32 .flatMap(quoteRepository::findLatestBySymbol, 8)
33 .map(QuoteEvent::from)
34 .sort(Comparator.comparing(QuoteEvent::timestamp).reversed())
35 .take(100);
36 }
37 
38 private static final Logger log = LoggerFactory.getLogger(QuoteStreamController.class);
39}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Returning a Flux lets a single endpoint push an open-ended stream instead of one response.
  2. 2Backpressure operators like onBackpressureBuffer and limitRate keep a fast producer from overwhelming a slow client.
  3. 3Reactive pipelines fold timeouts, cancellation, and error mapping into the same declarative chain.

Related explainers

ruby 
class Order
  class InvalidTransition < StandardError; end
 
  TRANSITIONS = {

A state machine for order transitions in Ruby

state-machine data-driven error-handling
Intermediate 8 steps
java 
public class SlidingLogRateLimiter {
 
    private final int maxRequests;
    private final long windowMillis;

How a sliding-log rate limiter works

rate-limiting concurrency sliding-window
Advanced 8 steps
java 
public final class Slugifier {
 
    private static final Pattern NON_LATIN = Pattern.compile("[^\\w-]");
    private static final Pattern WHITESPACE = Pattern.compile("[\\s]+");

Building a URL slugifier in Java

regex unicode-normalization string-processing
Intermediate 8 steps
rust 
use std::cmp::Ordering;
use std::str::FromStr;
 
#[derive(Debug, Clone, PartialEq, Eq)]

Parsing and ordering semantic versions in Rust

parsing trait-implementation ordering
Intermediate 8 steps
javascript 
'use client';
 
import { useEffect } from 'react';
import * as Sentry from '@sentry/nextjs';

How a Next.js error boundary recovers

error-boundary error-handling observability
Intermediate 8 steps
ruby 
class Registration < ApplicationRecord
  belongs_to :event
 
  validates :email, presence: true, format: { with: URI::MailTo::EMAIL_REGEXP }

Validating registrations in Rails

validations i18n error-handling
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Streaming market quotes with Spring WebFlux — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code