dami - bus 之流事件（stream）

1、发送（流事件）与监听（流事件）

public class DemoApp {
    static String topic = "demo.hello";

    public static void main(String[] args) {
        //监听流事件
        Dami.bus().<String, String>listen(topic, (event, att, data, sink) -> {
            System.err.println(data);
            sink.onNext("hi");
            sink.onComplete();
        });

        //发送流事件
        Flux.from(Dami.bus().<String, String>stream(topic, "hello")).doOnNext(item -> {
            System.err.println(item);
        }).subscribe();
    }
}

提醒：事件监听的参数差别（开发工具通过参数个数，推断为不同的类型）：

2、流事件的内部处理

流事件的内部同样是“通用事件”（仅是一种体验简化）。改成“通用事件”如下：

public class DemoApp {
    static String topic = "demo.hello";

    public static void main(String[] args) {
        //监听流事件
        Dami.bus().<StreamPayload<String, ? super String>>listen(topic, event -> {
            System.err.println(event.getPayload().getData());
            event.getPayload().getSink().onNext("hi");
            event.getPayload().getSink().onComplete();
        });

        //发送流事件
        Flux.<String>from(subscriber -> {
            Dami.bus().<StreamPayload<String, ? super String>>send(topic, new StreamPayload<>("hello", subscriber));
        }).doOnNext(item -> {
            System.err.println(item);
        }).subscribe();
    }
}