asmhatre/spring-messaging-demo
1
0
Fork 0
Code Activity

Add the protocol-comparison module

Browse Source
This commit is contained in:
asmhatre
2026-09-04 00:52:18 +05:30
parent 5224afdad2
commit d56c60824e
32 changed files with 1710 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
protocol-comparison/src/main/java/com/ankurm/protocols/GrpcQuoteService.java Normal file
View File

@@ -0,0 +1,121 @@
package com.ankurm.protocols;
import com.ankurm.protocols.grpc.Quote;
import com.ankurm.protocols.grpc.QuoteRequest;
import com.ankurm.protocols.grpc.QuoteServiceGrpc;
import com.ankurm.protocols.grpc.StreamRequest;
import io.grpc.Context;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.stereotype.Service;
/**
* The gRPC side. No registration code: the generated {@code ImplBase} is a
* {@code BindableService} and Boot registers every such bean with the server.
*
* <p>The cancellation check in {@link #streamQuotes} is not decoration. gRPC does not interrupt
* your thread when a client goes away &mdash; it sets a flag on the {@link Context} &mdash; so a
* loop that never looks keeps producing into a stream nobody is reading. That is covered at
* length in <a href="https://ankurm.com/spring-grpc-spring-boot-4/">the Spring gRPC article</a>;
* here it matters because the back-pressure benchmark deliberately abandons a stream.
*
* @see <a href="../../../../../docs/04-backpressure.md">docs/04-backpressure.md</a>
*/
@Service
public class GrpcQuoteService extends QuoteServiceGrpc.QuoteServiceImplBase {
/** Counts what the SERVER produced, comparable with the RSocket and WebSocket counters. */
static final AtomicLong PRODUCED = new AtomicLong();
private final QuoteSource source;
GrpcQuoteService(QuoteSource source) {
this.source = source;
}
@Override
public void getQuote(QuoteRequest request, StreamObserver<Quote> observer) {
observer.onNext(toProto(source.at(request.getSymbol(), 0)));
observer.onCompleted();
}
/**
* An effectively unbounded stream, for the back-pressure comparison. A consumer that stops
* calling {@code next()} closes the HTTP/2 flow-control window, and this loop blocks inside
* {@code onNext} &mdash; so the server stops producing, but it stops by blocking a thread
* rather than by being asked to stop.
*/
@Override
public void streamUnbounded(QuoteRequest request, StreamObserver<Quote> observer) {
PRODUCED.set(0);
while (!Context.current().isCancelled()) {
observer.onNext(toProto(source.at(request.getSymbol(), PRODUCED.get())));
PRODUCED.incrementAndGet();
}
}
/**
* The fix, and the reason the unfixed version is worth measuring.
*
* <p>{@code StreamObserver.onNext} on a gRPC server <strong>never blocks</strong>. If the
* client is not reading, the message is queued in the server's outbound buffer and the loop
* keeps going &mdash; HTTP/2 flow control governs the wire, not your code. The only thing
* that connects the two is {@link ServerCallStreamObserver#isReady()}, and using it means
* restructuring the handler around {@code setOnReadyHandler} rather than writing a loop.
*
* <p>The spin here is deliberately the simplest possible demonstration rather than the
* shape you would ship; a real implementation registers an on-ready handler and returns.
*/
@Override
public void streamUnboundedReady(QuoteRequest request, StreamObserver<Quote> observer) {
PRODUCED.set(0);
ServerCallStreamObserver<Quote> ready = (ServerCallStreamObserver<Quote>) observer;
while (!ready.isCancelled()) {
if (!ready.isReady()) {
Thread.onSpinWait();
continue;
}
ready.onNext(toProto(source.at(request.getSymbol(), PRODUCED.get())));
PRODUCED.incrementAndGet();
}
}
@Override
public void produced(QuoteRequest request, StreamObserver<com.ankurm.protocols.grpc.ProducedCount> observer) {
observer.onNext(com.ankurm.protocols.grpc.ProducedCount.newBuilder()
.setCount(PRODUCED.get()).build());
observer.onCompleted();
}
@Override
public void streamQuotes(StreamRequest request, StreamObserver<Quote> observer) {
for (int i = 0; i < request.getCount(); i++) {
if (Context.current().isCancelled()) {
// Do NOT call onCompleted/onError here: the stream is already closed and
// touching it throws IllegalStateException. Just return.
return;
}
observer.onNext(toProto(source.at(request.getSymbol(), i)));
if (request.getDelayMs() > 0) {
try {
Thread.sleep(request.getDelayMs());
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
return;
}
}
}
observer.onCompleted();
}
static Quote toProto(com.ankurm.protocols.Quote q) {
return Quote.newBuilder()
.setSymbol(q.symbol()).setSeq(q.seq())
.setBid(q.bid()).setAsk(q.ask())
.setEpochMicros(q.epochMicros())
.build();
}
}

22
protocol-comparison/src/main/java/com/ankurm/protocols/ProtocolComparisonApplication.java Normal file
View File

@@ -0,0 +1,22 @@
package com.ankurm.protocols;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* One application, three protocols, one set of data.
*
* <p>gRPC on 9090, RSocket on 7000, WebSocket on the servlet port. Serving the same two
* operations from the same JVM is the only way to make a benchmark mean anything: the heap, the
* JIT state, the CPU and the data generator are shared, so a difference in the numbers is a
* difference in the transports.
*
* @see <a href="../../../../../docs/01-three-protocols.md">docs/01-three-protocols.md</a>
*/
@SpringBootApplication
public class ProtocolComparisonApplication {
public static void main(String[] args) {
SpringApplication.run(ProtocolComparisonApplication.class, args);
}
}

8
protocol-comparison/src/main/java/com/ankurm/protocols/Quote.java Normal file
View File

@@ -0,0 +1,8 @@
package com.ankurm.protocols;
/**
* The JSON/CBOR shape, field-for-field identical to the protobuf {@code Quote} message, so the
* encoded sizes in {@code docs/output/payload-sizes.txt} compare like with like.
*/
public record Quote(String symbol, long seq, double bid, double ask, long epochMicros) {
}

22
protocol-comparison/src/main/java/com/ankurm/protocols/QuoteSource.java Normal file
View File

@@ -0,0 +1,22 @@
package com.ankurm.protocols;
import java.time.Instant;
import org.springframework.stereotype.Component;
/**
* Deterministic quote generation, shared by all three protocol adapters.
*
* <p>Deliberately cheap and deliberately allocation-free apart from the record: a benchmark of
* three transports must not spend its time in the thing behind them.
*/
@Component
public class QuoteSource {
public Quote at(String symbol, long seq) {
double base = 100.0 + (seq % 97) * 0.01;
Instant now = Instant.now();
return new Quote(symbol, seq, base, base + 0.02,
now.getEpochSecond() * 1_000_000L + now.getNano() / 1_000L);
}
}

72
protocol-comparison/src/main/java/com/ankurm/protocols/RSocketQuoteController.java Normal file
View File

@@ -0,0 +1,72 @@
package com.ankurm.protocols;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* The RSocket side. The same two operations, and one extra that exists only to make
* back-pressure visible.
*
* <p>Note what the signatures say that the other two protocols' cannot: {@code Mono} means
* request/response, {@code Flux} means request/stream, and the choice of interaction model is
* the return type rather than a separate declaration. RSocket has four (fire-and-forget,
* request/response, request/stream, channel) and Spring picks from the method shape.
*
* @see <a href="../../../../../docs/04-backpressure.md">docs/04-backpressure.md</a>
*/
@Controller
public class RSocketQuoteController {
private static final Logger log = LoggerFactory.getLogger(RSocketQuoteController.class);
/** Counts what the SERVER produced, which is the number the back-pressure test cares about. */
static final AtomicLong PRODUCED = new AtomicLong();
private final QuoteSource source;
RSocketQuoteController(QuoteSource source) {
this.source = source;
}
@MessageMapping("quote")
public Mono<Quote> quote(String symbol) {
return Mono.just(source.at(symbol, 0));
}
@MessageMapping("quotes")
public Flux<Quote> quotes(StreamSpec spec) {
Flux<Quote> flux = Flux.range(0, spec.count()).map(i -> source.at(spec.symbol(), i));
return spec.delayMs() > 0 ? flux.delayElements(Duration.ofMillis(spec.delayMs())) : flux;
}
/**
* An unbounded generator. It emits only when the transport asks, so the count it reaches is
* a direct read-out of how much the consumer requested &mdash; which is the whole argument
* for RSocket.
*/
@MessageMapping("quotes.unbounded")
public Flux<Quote> unbounded(String symbol) {
PRODUCED.set(0);
return Flux.generate(() -> 0L, (seq, sink) -> {
sink.next(source.at(symbol, seq));
PRODUCED.incrementAndGet();
return seq + 1;
});
}
@MessageMapping("produced")
public Mono<Long> produced() {
return Mono.just(PRODUCED.get());
}
/** Request payload for {@code quotes}. */
public record StreamSpec(String symbol, int count, int delayMs) {
}
}

27
protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketConfig.java Normal file
View File

@@ -0,0 +1,27 @@
package com.ankurm.protocols;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
/**
* {@code @EnableWebSocket}, not {@code @EnableWebSocketMessageBroker}: this module wants the raw
* transport, with no broker in the path, so the benchmark measures a socket rather than a routing
* layer. The {@code sse-websocket} module in this repository is the STOMP version.
*/
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
private final WebSocketQuoteHandler handler;
WebSocketConfig(WebSocketQuoteHandler handler) {
this.handler = handler;
}
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
registry.addHandler(handler, "/quotes").setAllowedOriginPatterns("*");
}
}

94
protocol-comparison/src/main/java/com/ankurm/protocols/WebSocketQuoteHandler.java Normal file
View File

@@ -0,0 +1,94 @@
package com.ankurm.protocols;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import tools.jackson.databind.ObjectMapper;
/**
* The WebSocket side, with no STOMP and no framework on top: a two-word text protocol, so the
* benchmark measures the transport and not a broker.
*
* <pre>
* QUOTE &lt;symbol&gt; -&gt; one JSON quote
* STREAM &lt;symbol&gt; &lt;count&gt; &lt;delayMs&gt; -&gt; count JSON quotes, then "END"
* UNBOUNDED &lt;symbol&gt; -&gt; quotes until the session dies
* </pre>
*
* <p>The absence worth noticing is in {@code UNBOUNDED}: there is nowhere for a consumer to say
* how many it wants. {@code sendMessage} either succeeds, blocks, or eventually throws when
* Spring's send buffer limit is exceeded. That is not a gap in this handler &mdash; it is what a
* raw WebSocket offers.
*
* @see <a href="../../../../../docs/04-backpressure.md">docs/04-backpressure.md</a>
*/
@Component
public class WebSocketQuoteHandler extends TextWebSocketHandler {
/** Counts what the SERVER wrote, comparable with RSocket's PRODUCED. */
static final AtomicLong PRODUCED = new AtomicLong();
private final QuoteSource source;
private final ObjectMapper mapper;
WebSocketQuoteHandler(QuoteSource source, ObjectMapper mapper) {
this.source = source;
this.mapper = mapper;
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
String[] parts = message.getPayload().trim().split("\\s+");
switch (parts[0]) {
case "QUOTE" -> send(session, source.at(parts[1], 0));
case "STREAM" -> {
int count = Integer.parseInt(parts[2]);
int delay = parts.length > 3 ? Integer.parseInt(parts[3]) : 0;
for (int i = 0; i < count; i++) {
send(session, source.at(parts[1], i));
if (delay > 0) {
Thread.sleep(delay);
}
}
session.sendMessage(new TextMessage("END"));
}
case "UNBOUNDED" -> {
PRODUCED.set(0);
// No requestN. Nothing here can be told to slow down: the loop writes until the
// socket, the send buffer or the send-time limit stops it. Catching the write
// failure keeps an abandoned stream from filling the log with one stack trace
// per run -- and note that catching it is the ONLY notification that arrives.
try {
while (session.isOpen()) {
send(session, source.at(parts[1], PRODUCED.get()));
PRODUCED.incrementAndGet();
}
}
catch (Exception ex) {
// Broken pipe. The client left; there was no other way to find out.
}
}
case "PRODUCED" -> session.sendMessage(new TextMessage(Long.toString(PRODUCED.get())));
default -> session.sendMessage(new TextMessage("ERR unknown command"));
}
}
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) {
// Where an over-buffered slow consumer surfaces: as a transport error on the writer,
// long after the consumer stopped reading.
PRODUCED.addAndGet(0);
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
}
private void send(WebSocketSession session, Quote quote) throws Exception {
session.sendMessage(new TextMessage(mapper.writeValueAsString(quote)));
}
}

45
protocol-comparison/src/main/proto/quotes.proto Normal file
View File

@@ -0,0 +1,45 @@
syntax = "proto3";
package quotes;
option java_multiple_files = true;
option java_package = "com.ankurm.protocols.grpc";
// The same two operations the RSocket and WebSocket sides expose, so the comparison is between
// transports rather than between designs.
service QuoteService {
// Request/response.
rpc GetQuote (QuoteRequest) returns (Quote);
// Server streaming: count quotes, as fast as the transport allows.
rpc StreamQuotes (StreamRequest) returns (stream Quote);
// Never completes on its own. Used to show what stops a gRPC server producing.
rpc StreamUnbounded (QuoteRequest) returns (stream Quote);
// The same unbounded stream, but the server checks ServerCallStreamObserver.isReady()
// before every write. Same API, different memory profile.
rpc StreamUnboundedReady (QuoteRequest) returns (stream Quote);
// How many messages the server has produced on the current unbounded stream.
rpc Produced (QuoteRequest) returns (ProducedCount);
}
message ProducedCount {
int64 count = 1;
}
message QuoteRequest {
string symbol = 1;
}
message StreamRequest {
string symbol = 1;
int32 count = 2;
// Milliseconds to sleep between emissions; 0 means "as fast as possible".
int32 delay_ms = 3;
}
message Quote {
string symbol = 1;
int64 seq = 2;
double bid = 3;
double ask = 4;
int64 epoch_micros = 5;
}

22
protocol-comparison/src/main/resources/application.yaml Normal file
View File

@@ -0,0 +1,22 @@
spring:
application:
name: protocol-comparison
rsocket:
server:
# RSocket gets its own port and its own Reactor Netty server. TCP here; "websocket" is the
# other transport and is what you use when the client is a browser.
port: 7000
transport: tcp
grpc:
server:
# spring.grpc.server.port -- note the spring. prefix and that "port: -1" means in-process
# only. See the property table in ankurm.com/spring-grpc-spring-boot-4/
port: 9090
server:
port: 8080
logging:
level:
io.grpc: WARN
io.rsocket: WARN