Как объединить потоки Kotlin только тогда, когда первый выдает (не нулевое) значение, отписываясь от второго, если это не так?

Question

Я пытаюсь добиться того, чтобы просто (повторно) подписываться, когда выдает значение, отличное от . Но нет стандартного способа сделать это (как я вижу).

В моем сценарии продолжать выдавать значения без необходимости дорого, поэтому мне нужен способ отписаться от него, когда он равен , но повторно подписаться, когда он имеет (не ) значение.

Если я использую что-то вродеtakeWhile(проверка значений) передcombine, происходит то, что он завершается с первым отрицательным результатом и больше не будет получать обновления.

СfilterNotNullЯ также не могу добиться того, что пытаюсь сделать, потому что потоковая передача будет продолжать работать с предыдущими значениями (она не остановится при получении a).

У кого-нибудь есть предложения о том, как добиться этого с помощью стандартных операторов?

Или это должно быть сделано как новый (пользовательский) оператор (например,combineWhenNotNull), например? Любые советы о том, как добиться такого поведения?

      flowA
  .combine(flowB) { a, b ->
    if (a != null) {
      Log.d(TAG, "$a -> $b")
    }
  }

Это базовая логика потоковой передачи, но таким образомflowBпродолжает выдавать значения, даже когдаflowAизлучаетnull(это то, чего я хочу избежать).