如何向ConnectableFlowable发送取消信号？

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.observables.ConnectableFlowable;

public class ConnectableFlowableExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建两个源 Flowable
        Flowable<Integer> source1 = Flowable.interval(1, TimeUnit.SECONDS).take(5);
        Flowable<Integer> source2 = Flowable.interval(1, TimeUnit.SECONDS).take(5);

        // 将两个源 Flowable 合并为一个 ConnectableFlowable
        ConnectableFlowable<Integer> connectableFlowable = Flowable.merge(source1, source2).publish();

        // 订阅 ConnectableFlowable
        connectableFlowable.subscribe(
            data -> System.out.println("Received: " + data),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Completed")
        );

        // 连接 ConnectableFlowable 以开始发射数据
        connectableFlowable.connect();

        // 等待一段时间后发送取消信号
        Thread.sleep(3000);
        connectableFlowable.unsubscribe();

        // 等待足够的时间以确保程序结束前所有操作完成
        Thread.sleep(2000);
    }
}

Flowable<Integer> source = Flowable.interval(1, TimeUnit.SECONDS).takeUntil(signal -> signal.isDisposed());