Class 10: Subscriptions

Duration: 55 minutes | Difficulty: Intermediate | Prerequisites: Class 9 completed

---

What You'll Learn

By the end of this class, you will:

• Understand how GraphQL subscriptions differ from queries and mutations
• Set up WebSocket transport for subscriptions
• Use Project Reactor's Flux and Sinks.Many to build an event stream
• Create a ReviewEventPublisher that emits events when reviews are posted
• Implement a @SubscriptionMapping that streams events to connected clients
• Filter subscription events by movie ID

How Subscriptions Differ from Queries

Queries and mutations follow a request/response pattern: the client sends a request, the server processes it, and the server sends back a single response. The connection ends there.

Subscriptions are fundamentally different. The client sends a subscription request, and the server keeps the connection open, pushing events to the client whenever something relevant happens. The client doesn't poll - it listens. This is ideal for real-time features like live notifications, chat messages, or activity feeds.

The key difference is who initiates communication. In queries, the client asks and the server answers. In subscriptions, the client subscribes and the server pushes. This inversion needs a transport that can keep a connection open and let the server send messages whenever events happen, not just when the client asks.

Why WebSocket?

Plain HTTP request/response cannot deliver server-initiated messages on its own: one request, one response, connection closes. To keep a channel open for pushes, GraphQL implementations use one of two transports on top of HTTP. WebSocket is a full-duplex persistent connection (the graphql-ws protocol is the modern standard) and Server-Sent Events (SSE) is a one-way server-to-client stream over plain HTTP (the graphql-sse protocol). Spring for GraphQL supports both.

We use WebSocket in this course because it is the most widely deployed option, the default in Spring for GraphQL examples, and the only transport supported by the Apollo Client subscription link without extra packages. SSE is a great fit when you only need one-way pushes and want to avoid the operational baggage of WebSocket (no special proxy config, works over HTTP/2 multiplexing), but it is one direction only and not all clients ship a GraphQL-over-SSE link out of the box.

SSE subscriptions are built in

You do not have to choose WebSocket to get streaming. Spring for GraphQL already serves subscriptions over Server-Sent Events at the same HTTP /graphql endpoint, with zero extra dependency and zero extra configuration - no spring-boot-starter-websocket, no spring.graphql.websocket.path. A client that sends Accept: text/event-stream on its POST to /graphql receives the subscription as an event stream, and your resolver code is identical: the very same Flux<ReviewNotification> from Step 7 drives both transports. WebSocket earns its keep when you need full-duplex or a client library that only speaks graphql-ws; when a one-way server-to-client push is all you need, the built-in SSE path is the lower-friction option.

Once the WebSocket handshake completes, both the client and server can send messages to each other at any time. The client opens a WebSocket connection to the server, sends a subscription document, and then receives a stream of events over that same connection until it disconnects or unsubscribes.

Transport	Direction	Connection	Use Case
HTTP	Client to server, server responds	Closes after response	Queries, mutations
WebSocket	Both directions, any time	Stays open	Subscriptions (course default)
SSE	Server to client only	Stays open	Subscriptions, built into Spring for GraphQL at the HTTP /graphql endpoint (send Accept: text/event-stream)

The Event Architecture

Before writing code, let's understand the architecture we're building. Subscriptions need three components:

1. Event publisher: something that emits events when interesting things happen (a review is posted)
2. Event stream: a reactive pipeline that carries events from the publisher to subscribers
3. Subscription resolver: the @SubscriptionMapping method that connects the event stream to a specific client

When a user posts a review, ReviewService tells the ReviewEventPublisher. The publisher pushes the event into a Sinks.Many, which is a Reactor primitive that acts like a broadcasting channel. Every connected subscriber receives the event through their own Flux - a reactive stream that Spring GraphQL sends over WebSocket.

Step 1: Add WebSocket Dependency

Add the WebSocket starter to your pom.xml:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

This gives Spring the ability to handle WebSocket connections. Without it, subscription requests will fail because the server has no WebSocket support.

Step 2: Configure the WebSocket Path

Tell Spring GraphQL where to listen for WebSocket connections:

Add to src/main/resources/application.yaml:

spring:
  graphql:
    websocket:
      path: /graphql
      keep-alive: 30s

This configures the WebSocket endpoint at the same path as the HTTP GraphQL endpoint (/graphql). The server distinguishes between HTTP and WebSocket requests by the connection upgrade header - a regular HTTP POST goes through the normal query/mutation pipeline, while a WebSocket upgrade request establishes a persistent connection for subscriptions.

The keep-alive value is a Duration. When set, the server sends a graphql-ws PING frame at that interval whenever no other message has gone out, and a healthy client answers with a PONG. Without it, a subscription that sits idle between events (which is the normal state - reviews are posted rarely) can be silently dropped by a load balancer, reverse proxy, or NAT gateway that reaps connections it believes are dead. Thirty seconds is a common starting point. The counterpart for SSE subscriptions, spring.graphql.sse.keep-alive, arrived in a later Spring Boot 3.x release; on WebSocket the property has been available since Spring for GraphQL 1.3.

Same Path, Different Protocols

Having the same /graphql path for both HTTP and WebSocket is the convention. The client library (Apollo, urql, etc.) knows to use HTTP for queries/mutations and WebSocket for subscriptions. You don't need separate paths.

WebSocket does not inherit the HTTP security context

Worth filing away for later: a WebSocket upgrade is not a normal HTTP request, so the JwtAuthFilter servlet filter from Class 6 never runs on it. The token that authenticates your queries and mutations never reaches a subscription. The reviewAdded stream we build in this class is public by design, so nothing here breaks, but the moment a subscription needs to know who is listening, you authenticate the connection itself. Spring for GraphQL provides AuthenticationWebSocketInterceptor (a servlet variant and a WebFlux variant, both extending AbstractAuthenticationWebSocketInterceptor) that reads an Authorization token from the connection_init payload the client sends when the socket opens, validates it, and propagates the resulting SecurityContext to every subscription on that connection. Class 13 returns to this when it wires request context across both transports.

Step 3: Create the ReviewNotification Record

We need a type to carry subscription events. This is the payload that subscribers receive:

Create src/main/java/com/graphqlguy/moviedb/review/ReviewNotification.java:

package com.graphqlguy.moviedb.review;

public record ReviewNotification(
    Review review,
    Long movieId,
    String movieTitle
) {}

This record bundles the review with its movie context. The movieId is important for filtering - subscribers can choose to only receive notifications for a specific movie. The movieTitle is a convenience field so clients can display "New review on The Dark Knight" without a follow-up query.

Step 4: Add the Schema Type

Add the subscription type and notification type to your schema:

Add to src/main/resources/graphql/schema.graphqls:

"""Notification published when a review is posted"""
type ReviewNotification {
  review: Review!
  movieId: ID
  movieTitle: String
}

type Subscription {
  """
  Receive live notifications when reviews are posted.
  Optionally filter by movieId to only receive reviews for a specific movie.
  """
  reviewAdded(movieId: ID): ReviewNotification!
}

A few things to notice about this schema definition:

The movieId argument is optional (no !). When omitted, the subscriber receives notifications for all movies. When provided, they only receive notifications for that specific movie. This makes the subscription flexible: a movie detail page subscribes with a movieId, while an admin dashboard subscribes without one to see everything.

The return type is ReviewNotification!, not [ReviewNotification!]!. Unlike queries that return complete results, subscriptions return one event at a time. Each event is a single ReviewNotification. The stream delivers many events over time, but each delivery is a single object.

Step 5: Create the ReviewEventPublisher

This is the heart of the subscription system. The publisher uses Reactor's Sinks.Many to broadcast events to all subscribers:

Create src/main/java/com/graphqlguy/moviedb/review/ReviewEventPublisher.java:

package com.graphqlguy.moviedb.review;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

@Component
@Slf4j
public class ReviewEventPublisher {

    private final Sinks.Many<ReviewNotification> sink =
        Sinks.many().multicast().onBackpressureBuffer(256);

    public void publish(ReviewNotification notification) {
        Sinks.EmitResult result = sink.tryEmitNext(notification);
        if (result.isFailure()) {
            log.warn("Failed to emit review notification: {}", result);
        } else {
            log.info("Published review notification for: {}",
                     notification.movieTitle());
        }
    }

    public Flux<ReviewNotification> getStream() {
        return sink.asFlux();
    }
}

Let's break down the Reactor concepts:

What Is Sinks.Many?

Sinks.Many is a programmatic event emitter. Think of it as a pipe: you push events in one end with tryEmitNext(), and subscribers read events from the other end as a Flux. It's the bridge between imperative code (our ReviewService calling publish()) and reactive streams (the Flux that @SubscriptionMapping returns).

Why multicast()?

Reactor provides three flavors of Sinks.Many:

Flavor	Behavior	Use Case
unicast()	Only one subscriber allowed	Single consumer processing
multicast()	Multiple subscribers, each gets events published after they subscribe	Live notifications - you see new events, not old ones
replay()	Multiple subscribers, new subscribers receive some/all past events	Chat history, event log

We use multicast() because each WebSocket connection is a separate subscriber, and there can be many simultaneously. A subscriber should only see reviews posted after they connected - there's no need to replay old reviews.

What Is onBackpressureBuffer(256)?

If events are published faster than subscribers can consume them, the buffer holds up to 256 events. In practice, review posting is infrequent enough that this buffer will never fill. But it's good practice to set a limit rather than allowing unbounded memory growth.

Why tryEmitNext() Instead of emitNext()?

tryEmitNext() is non-blocking and returns an EmitResult that tells you whether the emission succeeded. emitNext() takes a failure handler and can retry, but for our use case, logging a warning and moving on is sufficient. A failed emission means the sink is in a bad state (closed, full, etc.), which is unlikely in normal operation.

Step 6: Publish Events from ReviewService

Update ReviewService to publish a notification after saving a review:

Update src/main/java/com/graphqlguy/moviedb/review/ReviewService.java:

@Service
@RequiredArgsConstructor
public class ReviewService {

    private final ReviewRepository reviewRepository;
    private final MovieRepository movieRepository;
    private final UserRepository userRepository;
    private final ReviewEventPublisher reviewEventPublisher;

    @Transactional
    public Review createMovieReview(CreateMovieReviewInput input,
                                    String username) {
        AppUser user = userRepository.findByUsername(username)
            .orElseThrow(() -> new IllegalStateException("Authenticated user has no matching record: " + username));
        Long movieId = Long.parseLong(input.movieId());
        Movie movie = movieRepository.findById(movieId)
            .orElseThrow(() -> new EntityNotFoundException("Movie", movieId));

        Review saved = reviewRepository.save(Review.builder()
            .user(user).movie(movie)
            .score(input.score()).comment(input.comment()).build());

        // Publish event for subscribers
        reviewEventPublisher.publish(
            new ReviewNotification(saved, movieId, movie.getTitle())
        );

        return saved;
    }
}

The two lookups fail with deliberately different exceptions. A missing movie is a true NOT_FOUND - the client referenced a row that isn't there - so it throws EntityNotFoundException. A missing user is a different situation: the username came from the authenticated session, so an empty result does not mean the client asked for something absent - it means an authenticated identity no longer resolves to a real account, a data-integrity problem rather than anything the client did wrong. We surface that as an IllegalStateException, which Spring GraphQL classifies as INTERNAL_ERROR - the same way Class 7 treats this guard, since both mutations sit behind @PreAuthorize("isAuthenticated()").

The publish call happens after the call to save(). This is important - if the save fails before we reach the publish line (validation error, missing parent entity), we never notify subscribers about a review that does not exist.

Publishing inside a transaction is not enough

The publish() call is still inside the @Transactional method, and that is the part to be careful about. reviewRepository.save() writes to the JPA persistence context; the actual database flush typically happens at commit, and even after flush the transaction can still roll back (a constraint violation on commit, a connection drop, a thrown exception further up the call stack). If any of that happens, subscribers will already have received a notification for a review that was never committed.

The robust pattern is to publish only after the transaction commits. Instead of calling publisher.publish(...) directly, publish a Spring ApplicationEvent and listen for it with @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT). Spring guarantees the listener fires only if (and after) the surrounding transaction commits successfully. The simplest sketch:

// In ReviewService - replace the direct publish call with:
applicationEventPublisher.publishEvent(
    new ReviewCreatedEvent(saved, movieId, movie.getTitle())
);

// In a separate listener bean:
@Component
@RequiredArgsConstructor
class ReviewEventListener {
    private final ReviewEventPublisher reviewEventPublisher;

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    void onReviewCreated(ReviewCreatedEvent event) {
        reviewEventPublisher.publish(
            new ReviewNotification(event.review(), event.movieId(), event.movieTitle())
        );
    }
}

For this tutorial we keep the inline publish() call to focus on the subscription mechanics, but in production code you should treat any "notify the world" side effect as something that runs after commit, not during.

Step 7: Create the Subscription Controller

Now wire everything together with a @SubscriptionMapping:

Create src/main/java/com/graphqlguy/moviedb/review/ReviewSubscriptionController.java:

package com.graphqlguy.moviedb.review;

import lombok.RequiredArgsConstructor;
import org.springframework.graphql.data.method.annotation.Argument;
import org.springframework.graphql.data.method.annotation.SubscriptionMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;

@Controller
@RequiredArgsConstructor
public class ReviewSubscriptionController {

    private final ReviewEventPublisher publisher;

    @SubscriptionMapping
    Flux<ReviewNotification> reviewAdded(@Argument Long movieId) {
        Flux<ReviewNotification> flux = publisher.getStream();
        if (movieId != null) {
            flux = flux.filter(n -> movieId.equals(n.movieId()));
        }
        return flux;
    }
}

@SubscriptionMapping works like @QueryMapping and @MutationMapping - the method name matches the field name in the schema (reviewAdded). But instead of returning a single value, it returns a Flux - a reactive stream that emits values over time.

Spring GraphQL takes the returned Flux and bridges it to the WebSocket connection. Every time the Flux emits a ReviewNotification, Spring serializes it as a GraphQL response and sends it over the WebSocket to the client. The client receives a stream of individual responses, each containing one ReviewNotification.

There is a mental model worth making explicit here. The @SubscriptionMapping runs once to set up the stream, but each emitted event then triggers a fresh execution of the subscription's selection set, with the emitted ReviewNotification as the source object. So any nested @SchemaMapping resolvers, @BatchMapping loaders, and per-field authorization checks re-run for every event, exactly as they would for a normal query on that payload - which is why a subscription that selects deep, expensive fields can be far more costly than the one-line resolver suggests. (One more rule from the spec: a subscription operation may have exactly one root field.)

The Optional Filter

The movieId argument is optional. When provided, we apply a filter() operator to the Flux. This is a Reactor operator that drops any events that don't match the predicate. The filtering happens on the server - events for other movies are discarded before being sent over the wire, saving bandwidth.

Without the filter, a movie detail page subscribed to reviewAdded would receive notifications about reviews on every movie - irrelevant noise. The filter lets the client say "I only care about movie 42" and receive only those events.

Step 8: Understanding the Reactive Flow

Let's trace through the complete lifecycle of a subscription:

1. Client subscribes:

subscription {
  reviewAdded(movieId: "3") {
    review { score comment user { username } }
    movieTitle
  }
}

The client sends this over WebSocket. Spring GraphQL calls reviewAdded(movieId: 3L). The method returns a Flux<ReviewNotification> filtered to movie 3. Spring GraphQL subscribes to this Flux and begins listening.

2. Nothing happens yet. The Flux is "cold" from the subscriber's perspective - it only emits when the publisher pushes an event. The WebSocket connection is open but idle.

3. Someone posts a review on movie 3: ReviewService.createMovieReview() saves the review and calls reviewEventPublisher.publish(notification). The publisher pushes the notification into the Sinks.Many.

4. The Flux emits. The notification passes through the filter (movieId matches), and Spring GraphQL serializes it and sends it over WebSocket:

{
  "data": {
    "reviewAdded": {
      "review": { "score": 8, "comment": "Great movie!", "user": { "username": "john" } },
      "movieTitle": "The Dark Knight"
    }
  }
}

5. Someone posts a review on movie 7. The publisher emits the notification, but the filter drops it because movieId doesn't match 3. The client sees nothing.

6. The client disconnects (navigates away, closes the tab). Spring GraphQL cancels the Flux subscription. The reactive pipeline is cleaned up automatically.

Step 9: Testing Subscriptions

Subscriptions are harder to test than queries and mutations because they're asynchronous and long-lived. You can't simply send a request and assert on the response - you need to subscribe, then trigger an event, then verify the event arrives.

Testing Subscriptions

The GraphiQL page that ships with Spring for GraphQL runs graphql-ws subscriptions directly: it points its subscription fetcher at the WebSocket URL for you. Enable it with spring.graphql.graphiql.enabled=true (it is off by default, served at /graphiql) and make sure spring.graphql.websocket.path is set from Step 2 so the socket has somewhere to connect. Type a subscription, press the play button, and events stream into the response pane as they arrive. For manual testing you also have:

• The frontend we'll build in a later class has subscription support built in
• Altair GraphQL Client (a desktop app) supports WebSocket subscriptions
• websocat or similar CLI tools can connect to WebSocket endpoints
• Automated tests using GraphQlTester with Reactor's StepVerifier (shown below)

For automated testing, Spring provides GraphQlTester (not HttpGraphQlTester, since subscriptions don't go over HTTP) combined with Reactor's StepVerifier:

@SpringBootTest
@AutoConfigureGraphQlTester
class SubscriptionTest {

    @Autowired
    private GraphQlTester graphQlTester;

    @Autowired
    private ReviewEventPublisher eventPublisher;

    @Test
    void shouldReceiveReviewNotification() {
        var subscription = graphQlTester
            .document("""
                subscription {
                    reviewAdded {
                        movieTitle
                        review { score }
                    }
                }
                """)
            .executeSubscription()
            .toFlux("reviewAdded", ReviewNotification.class);

        StepVerifier.create(subscription.take(1))
            .then(() -> eventPublisher.publish(
                new ReviewNotification(testReview, 1L, "Test Movie")
            ))
            .assertNext(notification -> {
                assertThat(notification.movieTitle())
                    .isEqualTo("Test Movie");
            })
            .verifyComplete();
    }
}

The StepVerifier pattern works like this:

1. .create(subscription.take(1)) - subscribe to the Flux, expecting exactly 1 event
2. .then(() -> ...) - after subscribing, publish a test event
3. .assertNext(...) - verify the received event
4. .verifyComplete() - verify the Flux completes (because we used take(1))

The ordering matters: we start listening before publishing, so we don't miss the event. take(1) ensures the test doesn't hang waiting for more events that will never come.

Two practical notes. StepVerifier lives in the reactor-test dependency (io.projectreactor:reactor-test, test scope), so add it if it isn't already on the classpath. And this test uses the in-process @AutoConfigureGraphQlTester, which drives the Flux directly without a real socket - fast and focused, but it does not exercise the WebSocket transport. To test over an actual WebSocket connection end-to-end, point a WebSocketGraphQlTester at your running server.

Step 10: Run and Verify

Restart your application. The WebSocket endpoint is now active at ws://localhost:8080/graphql.

The quickest way to see it end-to-end is two GraphiQL tabs (with spring.graphql.graphiql.enabled=true from the testing tip above). Open the second tab first, so the subscription is already listening when you post the review:

1. In one GraphiQL tab, run the subscription and press play. It stays open, waiting:

subscription {
  reviewAdded {
    review { score comment }
    movieTitle
  }
}

2. In a second GraphiQL tab, post a review (with authentication):

mutation {
  createMovieReview(input: {
    movieId: "1", score: 9, comment: "Amazing!"
  }) {
    id score
  }
}

3. Switch back to the first tab: the notification should already be sitting in the response pane.

If you prefer a dedicated tool, point Altair or a websocat session at ws://localhost:8080/graphql, send the same subscription, and fire the mutation from GraphiQL - the event arrives the same way.

Exercises

Exercise 1: Add a tvShowReviewAdded Subscription

Create a second subscription that notifies when reviews are posted on TV shows. You'll need to update ReviewService.addTvShowReview() to publish events too. Consider whether you need a separate Sinks.Many or can reuse the existing one with a type discriminator.

Exercise 2: Include the Reviewer Username

Update ReviewNotification to include the reviewer's username directly (not nested inside Review). This makes it easy for clients to display "john reviewed The Dark Knight" without navigating nested objects.

Exercise 3: Think About Scaling

Our Sinks.Many is in-memory - it only works within a single server instance. If you deployed two server instances behind a load balancer, a review posted on server A would only notify subscribers connected to server A. Research how you'd use Redis Pub/Sub or a message broker (RabbitMQ, Kafka) to distribute events across instances. You don't need to implement it - just sketch the architecture.

Common Issues

Issue: WebSocket connection refused

Error: Client gets "connection refused" or "WebSocket handshake failed" Solution: Verify the WebSocket dependency is in pom.xml and the spring.graphql.websocket.path is set in application.yaml. Also check that your SecurityConfig doesn't block WebSocket upgrade requests - the permitAll() configuration from Class 6 should allow them.

Issue: Subscription never receives events

Error: The subscription connects but no events arrive when reviews are posted Solution: Verify that ReviewService calls reviewEventPublisher.publish() after saving. Check the log output - you should see "Published review notification for: ..." in the console. If you see the log but no events arrive, the subscriber might have connected after the event was published (a race condition in testing).

Issue: Sinks.Many emit failures

Error: Log shows "Failed to emit review notification: FAIL_OVERFLOW" Solution: The buffer is full (256 events). This shouldn't happen in normal use. If it does, either increase the buffer size or investigate why subscribers aren't consuming events fast enough.

Issue: CORS errors with WebSocket

Error: Browser blocks the WebSocket connection with a CORS error Solution: WebSocket CORS is configured separately from HTTP CORS. Add the frontend origin to the allowed origins in your CORS configuration. Spring GraphQL respects the same spring.graphql.cors configuration for WebSocket connections.

Summary

In this class, you learned:

• GraphQL subscriptions maintain a persistent connection where the server pushes events to the client - unlike queries and mutations which follow request/response
• WebSocket is the transport protocol for subscriptions because HTTP doesn't support server-initiated messages
• Sinks.Many with multicast() acts as a broadcasting channel - multiple subscribers each receive every event published after they connect
• ReviewEventPublisher bridges imperative code (publish()) with reactive streams (getStream() returns a Flux)
• @SubscriptionMapping returns a Flux that Spring GraphQL streams over WebSocket to the client
• Optional filtering (by movieId) uses Reactor's filter() operator to reduce noise for clients that only care about specific movies
• Events are published after successful saves to ensure subscribers only hear about reviews that actually exist in the database

What's Next?

In Class 11, we'll explore custom scalars and file handling:

• Creating custom DateTime scalar types
• Handling file uploads in GraphQL
• Validating custom types

---