Netflix DGS 订阅

Question

我正在按照 DGS subscriptions 的文档进行操作，我没有收到任何错误，但也没有取回任何数据。

设置非常简单。在 schema.graphqls 文件中我定义了订阅：

type Subscription {
    ratings: Rating
}

type Rating {
    stars: Int
}

而Java代码如下：

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

如果我现在将一个 websocket 连接到我的后端，它连接得很好，但没有按预期取回任何数据（不管我怎么做，也尝试使用 JavaScript，也连接得很好但没有得到任何数据）。

例如使用curl连接（但是使用Java脚本结果是一样的，连接但是没有数据）：

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==

我也尝试通过 Graphiql 接口连接，但出现错误：

subscription {
  ratings {
    stars
  }
}

错误信息：

{
  "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)"
}

我在 link 中的示例中不清楚的另一件事是如何实际管理订阅。因此，例如，假设我想在发生突变时发布通知。任何有关如何使用 Netflix DGS 框架完成此操作的指示也将不胜感激。

Answer 1

不幸的是，DGS 附带的 graphiql 界面似乎无法正确处理订阅 - 如果您按照文档添加 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。

关于你的第二个问题，如果发生突变，如何发布通知 - 不幸的是，文档中缺少这个，但 examples repo.

中有一个示例

我在这里稍微简化了示例。如果你想支持这样的订阅和变更：

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

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

在您的服务中，您将创建一个 Flux（return 订阅者）并保留对其发射器的引用，以便在发生突变时调用 next。

@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;
    }
}

Netflix DGS 订阅

Netflix DGS Subscriptions

graphql-java

netflix

netflix-dgs