流中项目的 rxjava 过滤器不匹配流中的任何其他过滤器

Question

我的应用程序中的 CommandManager 通过网络协议接收命令并将它们发布到命令 PublishSubject 上。特定 CommandHandler 类订阅命令 PublishSubject，这样每个 CommandHandler 只接收它设计用来处理的命令。我还希望有一个默认命令处理程序，它只接收那些没有与其他 CommandHandler 的过滤谓词匹配的命令。

// Each CommandHandler subscribes to a filtered 
// stream of incoming commands matching this CommanHandler

commandManager.getCommandsPublishSubject()
.filter(getMessageFilterPredicate())
.subscribe(this::onNewMessage, 
     e -> LOGGER.error("Error getting new message", e));

有没有办法应用一个过滤器来匹配任何 CommandHandler 都不匹配的项目？

Answer 1

没有执行此操作的标准运算符。您必须创建一个自定义组件，而不是运算符，它承载过滤器 lambda 以及每个案例的更多主题和默认主题。

public final class FilteringDispatcher<T> {

    static final class FilterAndSubject<T> {
         Predicate<T> predicate;
         Subject<T> subject;
    }

    final ConcurrentMap<String, FilterAndSubject<T>> filters = new ConcurrentHashMap<>();

    final Subject<T> defaultSubject = PublishSubject.<T>create().toSerialized();

    public Observable<T> registerPredicate(String name, Predicate<T> predicate) {
         Subject<T> subject = PublishSubject.<T>create().toSerialized();
         FilterAndSubject fs = new FilterAndSubject();
         fs.predicate = predicate;
         fs.subject = subject;

         FilterAndSubject old = filters.putIfAbsent(name, fs);
         return old != null ? old.subject : subject;
    }

    public Observable<T> getObservable(String name) {
        return filters.get(name).subject;
    }

    public Observable<T> getDefaultObservable() {
        return defaultSubject;
    }

    public void onNext(T item) {
        for (FilterAndSubject fs : filters.values()) {
            if (fs.predicate.test(item)) {
                fs.subject.onNext(item);
                return;
            }
        }
        defaultSubject.onNext(item);
    }
}

流中项目的 rxjava 过滤器不匹配流中的任何其他过滤器

rxjava filter for items in stream not matching any other filters on stream

rx-java