我正在使用 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);
}
}