Need inclusive predicate variant of takeUntil/takeWhile

[samueltardieu]

It has been two times I've been missing a takeUntil(filter) operator on observables. The last time was in cgeo while implementing a low-power mode: I want to receive location updates through an observable and stop as soon as a location arrives with a precision of 20m or less, but I want this location to be returned. Using takeWhile(!filter) would drop the matching item.

Of course, this can be worked around, but I was surprised not to get this operator which is the dual of takeWhile. Would you consider a contribution adding it?

[benjchristensen]

Good question. I'd like to find out from @headinthebox what the canonical solution is to this as Haskell, various Rx flavors and others either must have or need this solution as well.

The takeUntil(filter) solution seems like it is a reasonable approach. My only question is about the semantics as I could interpret "until" either way (inclusive or exclusive).

[benjchristensen]

/cc @spodila who also needs a solution to this

[spodila]

Nice. I could use that. Thank you.

[headinthebox]

We cannot add every possible operator that people need in the library. Instead we must make sure you can achieve waht you want by composing a set of orthogonal primitive operators. In this case the solution is as follows:

object MainScala {

  def main(args: Array[String]): Unit = {

    val xs = Observable.items(0,1,2,3,4)

    val ys = xs.takeWhile(x => x < 3)
    ys.subscribe(y => println(y)) // 0,1,2

    val zs = xs.publish[Int]((xs: Observable[Int]) => xs.takeUntil(xs.dropWhile(x => x < 3)))
    zs.subscribe(z => println(z)) // 0,1,2,3
    readLine()
  }
}

[spodila]

Maybe I am missing something...
I don't see dropWhile() in 0.20.3. Is it already being added?
That makes it look a bit convoluted, but looks like my use case can work with it.

[samueltardieu]

@headinthebox Is there a guarantee that your example will work in every case? I may be missing something, but where in the documentation is it described that the xs.dropWhile(…) observable will consider the respective elements of xs before or after xs.takeUntil(…) one? Isn't there a race condition here? Is it guaranteed on every scheduler?

In your example, couldn't xs.dropWhile(x => x < 3) emit "4" only after xs.takeUntil(…) have already let the "4" pass?

[zsxwing]

I don't see dropWhile() in 0.20.3. Is it already being added?

dropWhile is here: https://github.com/ReactiveX/RxScala/blob/0.x/src/main/scala/rx/lang/scala/Observable.scala#L1981

[samueltardieu]

@headinthebox Your example with RxJava 0.20.4 gives to me "0 1 2 0 1 2", not "0 1 2 0 1 2 3" as shown in your comments. Your code seems to be the strict equivalent to takeWhile (drop as soon as the negation of the condition is met), which does not do what is requested.

The proposed takeUntil(_ >= 3) would properly give "0 1 2 3".

[headinthebox]

I'm without a proper computer right now, but did you run exactly the same snippet using RxScala?

[samueltardieu]

Yup, I only added the import:

import rx.lang.scala._

object MainScala {

  def main(args: Array[String]): Unit = {

  val xs = Observable.items(0, 1, 2, 3, 4)

    val ys = xs.takeWhile(x => x < 3)
    ys.subscribe(y => println(y)) // 0,1,2

    val zs = xs.publish[Int]((xs: Observable[Int]) => xs.takeUntil(xs.dropWhile(x => x < 3)))
    zs.subscribe(z => println(z)) // 0,1,2,3
    readLine()
  }

build.sbt:

scalaVersion := "2.11.2"

libraryDependencies += "com.netflix.rxjava" % "rxjava-scala" % "0.20.4"

And run:

% sbt
[info] Set current project to x (in build file:/tmp/x/)
> run
[info] Updating {file:/tmp/x/}x...
[info] Resolving jline#jline;2.12 ...
[info] Done updating.
[info] Running MainScala 
0
1
2
0
1
2

[success] Total time: 34 s, completed Sep 16, 2014 7:41:26 PM
>

But from what I read, this is expected, as takeUntil will stop as soon as the dropWhile fires, while we want to keep one more element. Maybe adding a tail after dropWhile would get the right one, but then comes my former question about ordering guarantees, and the code becomes convoluted.

[samueltardieu]

@headinthebox Any news on this one?

[headinthebox]

Was blocked because of broken scala build.

[headinthebox]

Yup, the tail is missing. There is no issue with reordering. Using .publish and takeUntil on yourself is a very common pattern that more people should use. For example you can also define switch that same way.

import rx.lang.scala._

object MainScala {
  def main(args: Array[String]): Unit = {
  val xs = Observable.items(0, 1, 2, 3, 4)
  val zs = xs.publish[Int]((xs: Observable[Int]) => xs.takeUntil(xs.dropWhile(x => x < 3).tail))
  zs.subscribe(z => println(z)) // 0,1,2,3
  readLine()
}