如何向现有Reactor Flux注入自定义事件流

Flux<MappedType> aFluxMap = Library.createMappingToMappedType();

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.UnicastProcessor;

// 假设我们有某种RawType和MappedType
class RawType { String data; public RawType(String data) { this.data = data; } @Override public String toString() { return "RawType(" + data + ")"; } }
class MappedType { String mappedData; public MappedType(String mappedData) { this.mappedData = mappedData; } @Override public String toString() { return "MappedType(" + mappedData + ")"; } }

public class CustomFluxEmitter {

    public static void main(String[] args) {
        // 1. 创建一个UnicastProcessor作为我们自定义事件的源
        UnicastProcessor<RawType> customRawProcessor = UnicastProcessor.create();
        // 2. 获取FluxSink，用于向customRawProcessor发射事件
        FluxSink<RawType> rawSink = customRawProcessor.sink();

        // 3. 将自定义的RawType流转换为MappedType流
        //    这里假设我们有一个转换函数，或者MappedType可以直接从RawType构建
        Flux<MappedType> yourCustomMappedFlux = customRawProcessor
                .map(raw -> new MappedType("Mapped(" + raw.data + ")"));

        // 此时 yourCustomMappedFlux 是一个可以由 rawSink 控制的 MappedType 流
        yourCustomMappedFlux.subscribe(
            mapped -> System.out.println("Custom Mapped Type: " + mapped),
            error -> System.err.println("Error in custom flux: " + error),
            () -> System.out.println("Custom flux completed")
        );

        // 4. 模拟发射自定义事件
        rawSink.next(new RawType("Input A"));
        rawSink.next(new RawType("Input B"));
        // rawSink.complete(); // 可以在适当时候完成流
    }
}

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.UnicastProcessor;
import java.time.Duration;

// 假设 MappedType 已经定义，并且 Library 提供了 createMappingToMappedType 方法
// 模拟外部库的Flux
class Library {
    public static Flux<MappedType> createMappingToMappedType() {
        // 模拟一个每秒发出一个 MappedType 的外部 Flux
        return Flux.interval(Duration.ofSeconds(1))
                   .take(3) // 只发出3个元素
                   .map(i -> new MappedType("Library Mapped " + i));
    }
}

public class MergeFluxExample {

    public static void main(String[] args) throws InterruptedException {
        // 1. 获取外部库的 Flux
        Flux<MappedType> aFluxMap = Library.createMappingToMappedType();

        // 2. 创建一个可控的 Flux，用于发射你的自定义 MappedType 事件
        UnicastProcessor<MappedType> customProcessor = UnicastProcessor.create();
        FluxSink<MappedType> customSink = customProcessor.sink();
        Flux<MappedType> yourCustomFlux = customProcessor;

        // 3. 使用 Flux.merge 合并两个 Flux
        // merge操作符会将两个或更多Publisher的元素交错合并到一个新的Flux中
        Flux<MappedType> combinedFlux = Flux.merge(aFluxMap, yourCustomFlux);

        // 4. 订阅合并后的 Flux 并处理事件
        combinedFlux.doOnNext(mapped -> System.out.println("Received: " + mapped))
                    .doOnComplete(() -> System.out.println("Combined Flux Completed"))
                    .subscribe();

        // 5. 模拟在运行时发射自定义 MappedType 事件
        System.out.println("Emitting custom events...");
        Thread.sleep(500); // 等待一下，让library的flux先开始
        customSink.next(new MappedType("Custom A"));
        Thread.sleep(1200);
        customSink.next(new MappedType("Custom B"));
        Thread.sleep(1200);
        customSink.next(new MappedType("Custom C"));
        customSink.complete(); // 完成自定义流

        // 等待一段时间观察输出
        Thread.sleep(5000);
    }
}