合并通量的 Spring Webflux 类型安全性答案

【问题标题】：Spring Webflux type safety of merged fluxes合并通量的 Spring Webflux 类型安全性
【发布时间】：2020-03-20 17:43:51
【问题描述】：

我已经使用 Project Reactor 实现了服务器发送事件，用于设置从服务器到客户端的单向通信。

在服务器上，我添加了一个字符串类型的磁通连接：

Flux<String> addDataConnection() {
        return Flux.create(emitter -> {
            fluxSinks.add(emitter);
        });
    }

每当我收到消息时，它都会传播消息：

void propagateMessage(String messageContent) {
        fluxSinks.parallelStream()
                .forEach(fluxSink -> fluxSink.next(messageContent));
    }

此数据来自外部来源，因此传入事件的频率超出了我的控制范围。

因此，我添加了一个额外的心跳流，我将它与我的数据连接流合并。这样，我既可以保持连接活动，也可以检测终止的连接，例如当浏览器窗口关闭时：

Flux.interval(Duration duration) 实际上返回一个 Long，但是我将它映射到一个 String 以匹配数据连接的类型安全。这个字符串实际上并没有被任何人使用

Flux<String> addHeartbeat() {
    return Flux.interval(Duration.ofSeconds(2))
            .map(i -> "heartbeat");
}

我的问题是关于助焊剂的元素类型。由于我现在在我的数据连接上返回字符串，我可以简单地从心跳返回一个字符串，因此从通量的合并中返回一个 Flux<String>：

Flux<String> initiateFluxStreams() {
        return Flux.merge(addDataConnection(), addHeartbeat());
    }

但是，如果我的数据流返回一个更复杂的对象并且我想确保通量合并方法中的类型安全怎么办？我总是可以返回一个类型 Flux<?> 来让我的编译器满意，但我对这个解决方案并不满意，因为它允许我（或其他人）返回他们想要的任何东西。通量的合并是否支持更详细的编译安全性？

编辑：我在https://stackoverflow.com/a/49082329/3432964 看到了答案，以及它如何使用<? extends someMarkerInterface> 作为类型。这是我唯一的选择吗？将整个内容包装在ServerSentEvent-object 中是否更好？

【问题讨论】：

标签： java spring-webflux project-reactor

【解决方案1】：

对于不同类型的合并 Flux，没有“神奇”的方法可以做到这一点，因为您无法保证这些类型将具有 any 共同的祖先（除了 Object，但是当您指出，这不是太有用。）

如果你的类型是Foo，你可以有一个Foo 的“特殊”实例（或子类），只是表示一个空的心跳类型，但我并不特别喜欢这种方法——它可能涉及到设计牺牲只是为了撬开那里的某种类型。

我的首选方法（注意，在其他地方没有看到，所以我不能将其称为“常用方法”）是使用Flux<Optional<Foo>>，其中“心跳”触发器只是空选项。比如：

Flux<Foo> addDataConnection() {
    return Flux.create(emitter -> {
        fluxSinks.add(emitter);
    });
}

Flux<Optional<Foo>> initiateFluxStreams() {
    return Flux.merge(addDataConnection().map(f -> Optional.of(f)), addHeartbeat().map(h -> Optional.empty()));
}

Flux<Long> addHeartbeat() {
    return Flux.interval(Duration.ofSeconds(2));
}

然后您可以过滤和映射生成的 Flux 以在处理之前清除心跳：

fluxOfOptionalFoo
        .filter(Optional::isPresent)
        .map(Optional::get);

出于多种原因，我更喜欢这种方法：

不需要对底层对象结构进行任何更改
更具描述性，您有一个“可能存在也可能不存在”的Flux 类型
很容易过滤掉空的选项（如果你想用心跳功能做一些特别的事情，也可以不过滤）

如果您愿意，如果您认为 optional 不够“描述性”，您当然可以实现自己的“心跳”泛型类，但我个人认为没有必要。

【讨论】：

这是一个有趣的观点。暂时，我只会使用Flux<String>，但我以后一定会记住你的解决方案。

【解决方案2】：

像许多协议（rest api 等）一样，我会实现一个包含元数据和正文有效负载的类。

// This is just some type of pseudo code

{
    private type: HEARTBEAT
}

{
    private type: DATA
    private payload: {
        // Some type of payload
    }
}

所以用泛型定义一个可以处理多种类型数据的类。

new Request<Foo>();

【讨论】：