5

我正在使用 JDK 9 Flow API 创建一个用户事件系统，所以我有一个房间（它实现了Flow.Subscriber<Notification>），它可能有很多用户，每个用户都可以随时提供（调度）更新。

当用户进入房间时，我订阅房间的更新user.subscribe(this)。但是没有退订，用户离开房间时如何退订？

public abstract class Room implements Flow.Subscriber<Notification> {
    private Flow.Subscription subscription;

    public void addUser(User user) {
        user.subscribe(this);
    }

    public void removeUser(User user) {
        // How can I unsubscribe the user?
    }

    @Override
    public void onSubscribe(final Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onError(final Throwable throwable) {
        // ...
    }

    @Override
    public void onNext(final Notification notification) {
        // ...
        subscription.request(1);
    }

    @Override
    public void onComplete() {
        // User left
    }
}

用户等级：

public class User extends SubmissionPublisher<Notification> {

    ....

    public int offer(Notification item) {
        return super.offer(item, (sub, msg) -> false);
    }
}
4

0 回答 0