Reactor Flux中向现有流动态发送消息的策略与挑战

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.UnicastProcessor;
import java.util.Arrays;
import java.util.List;

// 模拟外部库的MappedType和MyObj
class MappedType {
    private String value;
    // 私有构造函数，模拟外部库的限制
    private MappedType(String value) { this.value = value; }
    // 静态工厂方法，或者通过其他方式创建
    public static MappedType fromString(String s) { return new MappedType("Converted: " + s); }
    @Override
    public String toString() { return "MappedType{" + "value='" + value + '\'' + '}'; }
}

class MyObj {
    String data;
    public MyObj(String data) { this.data = data; }
    @Override
    public String toString() { return "MyObj{" + "data='" + data + '\'' + '}'; }
}

// 模拟外部库
class Library {
    // 假设这个Flux会从某个内部源（比如一个队列）发出MappedType
    // 并且它的底层可能是一个UnicastProcessor，只允许被订阅一次
    public static Flux<MappedType> createMappingToMappedType() {
        // 模拟一个已存在的、独立的Flux
        return Flux.just("ExistingData1", "ExistingData2")
                   .map(MappedType::fromString)
                   .doOnSubscribe(s -> System.out.println("Library Flux subscribed once."));
    }
}

public class FluxInjectionTutorial {

    // 假设您有能力将MyObj转换为MappedType
    private static MappedType convertMyObjToMappedType(MyObj obj) {
        // 这里是您的转换逻辑
        return MappedType.fromString(obj.data + "-transformed");
    }

    public static void main(String[] args) {
        // 步骤1: 创建一个可控的FluxSink来发出您的原始数据 (MyObj)
        UnicastProcessor<MyObj> myObjProcessor = UnicastProcessor.create();
        FluxSink<MyObj> myObjSink = myObjProcessor.sink();

        // 步骤2: 将您的原始MyObj数据流转换为MappedType流
        // 重要的是，这个转换是在您自己的控制下完成的，而不是依赖aFluxMap的内部机制
        Flux<MappedType> yourConvertedFlux = myObjProcessor
            .map(FluxInjectionTutorial::convertMyObjToMappedType)
            .doOnNext(m -> System.out.println("Your Converted Flux emitted: " + m));

        // 步骤3: 获取外部库的现有Flux
        Flux<MappedType> aFluxMap = Library.createMappingToMappedType();

        // 步骤4: 合并两个Flux的输出
        // Flux.merge操作符将并发地从两个源中取数据并合并到一个新的Flux中
        Flux<MappedType> combinedFlux = Flux.merge(aFluxMap, yourConvertedFlux);

        // 步骤5: 订阅合并后的Flux以处理所有MappedType数据
        System.out.println("--- Subscribing to combined Flux ---");
        combinedFlux.doOnNext(converted -> System.out.println("Combined Flux received: " + converted))
                    .doOnComplete(() -> System.out.println("Combined Flux completed."))
                    .subscribe();

        // 步骤6: 动态发送数据到您的FluxSink
        // 这些数据会经过您的转换逻辑，然后与aFluxMap的数据合并
        System.out.println("\n--- Emitting custom data ---");
        List<MyObj> customObjects = Arrays.asList(new MyObj("CustomDataA"), new MyObj("CustomDataB"));
        customObjects.forEach(myObjSink::next);
        myObjSink.complete(); // 发送完毕后，完成您的数据流
    }
}