重构嵌套的 Observables

Question

这是一个初学者的问题。我正在将多个进程的输出流回客户端，并且我想将每个流存储到一个单独的文件中。

所以我想要的是（大写字母表示嵌套流结束）

S: -a-a-a-b-b-a-A-c-b-B-c-c-...-d--C...D-e--...-z-E-z--z--...

R:  a-a-a-----a-A (complete)
          b-b-------b-B (complete)
                  c-----c-c--------C (complete)
                                d------D (complete)
                                         e--------E (complete)

                     .
                     .
                     . 
       (end many more nested streams coming)
                     .

所以我想要一个像 Observables 的动态工厂这样的东西。类似于using()，但据我了解using() 创建的Observable 与原始Observable 一样长，而我想在每次嵌套流完成时完成并关闭文件。

重要 - 我不想在内存中缓冲，因为这些是非常长的流（长时间运行的进程的输出）。所以我想避免buffer()，groupBy()。

Answer 1

如果您可以从一个值确定它是否是特定字母的最终值，则可以在内部组中使用 groupBy 和 takeUntil（您可能需要在 lambda 中进行一些尝试捕获）：

source
.groupBy(v -> v.type)
.flatMap(g -> 
     Observable.using(
         () -> createOutput(g.getKey()), 
         f -> g.takeUntil(v -> v.isEnd).doOnNext(v -> f.write(v))),
         f -> f.close()
     )
)
.subscribe(...)

TakeUntil 确保 groupBy 只保留子流，只要我们期望它的值（如果源正确排序，groupBy 将不会重新创建组）。

如果你真的想避免groupBy，你必须手动跟踪每个打开的文件并在适当的时候关闭它们：

Observable.using(
    HashMap::new,
    map -> source
       .doOnNext(v -> {
           Output o = map.get(v.type);
           if (o == null) {
               o = new Output(v.type);
               map.put(v.type, o);
           }
           o.write(v);
           if (v.isEnd) {
               o.close();
               map.remove(v.type);
           }
       }),
    map -> map.values().forEach(e -> e.close())
)
.subscribe(...);