Netflix DGS Subscriptions

Question

I'm following the documentation of DGS subscriptions and I'm not getting any errors but not getting back any data either.

The setup is very simple. In the schema.graphqls file I have defined the subscription:

type Subscription {
    ratings: Rating
}

type Rating {
    stars: Int
}

And the Java code is as follows:

@DgsComponent
public class SubscriptionDataFetcher {
    @DgsData(parentType = "Subscription", field = "ratings")
    public Publisher<Rating> ratings() {
        return Flux.interval(Duration.ofSeconds(1)).map(t -> new Rating(4));
    }
}

If I now connect a websocket to my backend it connects just fine but not getting back any data as expected (doesn't matter how I do it, also tried using JavaScript, also connects just fine but not getting any data).

For example using curl to connect (but using JavaScript the result is the same, connects but no data):

curl -o - --http1.1 \
    --include \
    --no-buffer \
    --header "Connection: Upgrade" \
    --header "Upgrade: websocket" \
    --header "Host: localhost:8443" \
    --header "Origin: https://localhost:8443" \
    --header "Sec-WebSocket-Key: SGVsbG8sIHdvcmxkIQ==" \
    --header "Sec-WebSocket-Version: 13" \
    https://localhost:8443/subscriptions\?query\=ewoicXVlcnkiOiAic3Vic2NyaXB0aW9uIHsgIHN0b2NrcyB7bmFtZX0gfSIKfQ==

I have tried connecting via the Graphiql interface as well but that gives an error:

subscription {
  ratings {
    stars
  }
}

Error message:

{
  "message": "response is not defined",
  "stack": "ReferenceError: response is not defined\n    at https://localhost:8443/graphiql:46:35\n    at async Object.onRun (https://unpkg.com/graphiql/graphiql.min.js:1:540500)"
}

Another thing that's not clear to me from the examples in the link is how to actually manage subscriptions. So for example let's say I want to publish a notification if a mutation takes place. Any pointers how that would be done using the Netflix DGS framework would also be much appreciated.

Answer 1

Unfortunately, the graphiql interface that comes with DGS does not seems to handle subscriptions properly - if you add playground-spring-boot-starter to your project, a more polished tool will be available at /playground, which fully supports subscriptions. If you try your subscription there it should work (assuming you have already added graphql-dgs-subscriptions-websockets-autoconfigure as per the docs).

Regarding your second question about how to publish a notification if a mutation takes place - this is unfortunately missing from the documentation, but there is an example in the examples repo.

I have stripped down the example a bit here. If you want to support a subscription & mutation like this:

@DgsSubscription
public Publisher<Review> reviewAdded() {
    return reviewsService.getReviewsPublisher();
}

@DgsMutation
public Review addReview(@InputArgument SubmittedReview review) {
    return reviewsService.saveReview(review);
}

Inside your service, you would create a Flux (to return to subscribers) and keep a reference to its emitter so you can call next on it whenever a mutation occurs.

@Service
public class ReviewsService {

    private FluxSink<Review> reviewsStream;
    private ConnectableFlux<Review> reviewsPublisher;

    @PostConstruct
    public void init() {
        Flux<Review> publisher = Flux.create(emitter -> {
            reviewsStream = emitter;
        });

        reviewsPublisher = publisher.publish();
        reviewsPublisher.connect();
    }

    public Review saveReview(SubmittedReview reviewInput) {
        Review review = Review.newBuilder()
                .username(reviewInput.getUsername())
                .starScore(reviewInput.getStarScore())
                .submittedDate(OffsetDateTime.now()).build();

        // Save to the database, etc.

        reviewsStream.next(review); // publishes the review to subscribers
        return review;
    }

    public Publisher<Review> getReviewsPublisher() {
        return reviewsPublisher;
    }
}