Need a variation on throttle that does not throttle first item

[dalewking]

Scenario is I have a screen that has a search field and the search requires making a web service call. We do not want to do a search on each character that is typed so we use throttle on the Observable from the search field so that we only process it after they stop typing. The issue is when we initially create the screen and subscribe to the Observable on the search field the very first value is also throttled. Resulting in an initial delay before retrieving results. Would like a way to pass a parameter to throttle telling it that the very first item after subscription should not be delayed.

[headinthebox]

You can build this from the existing operators. Just split the stream into the first and rest and throttle the rest.

val ys = xs.publish((_xs: Observable[Int]) => _xs.take(1)++_xs.WhateverThrottleYouWant())

[akarnokd]

I came up with the following construct:

public class ThrottleNotFirst {
  public static void main(String[] args) throws Exception {
      for (int firstWait = 0; firstWait < 2; firstWait++) {
          PublishSubject<String> source = PublishSubject.create();

          PublishSubject<Observable<String>> immediates = PublishSubject.create();

          Observable<Observable<String>> first = source.take(1)
                  .observeOn(Schedulers.io())
                  .map(s -> 
                       Observable.just("First " + s).delay(100, TimeUnit.MILLISECONDS));

          Observable<Observable<String>> rest = source.skip(1)
                  .debounce(100, TimeUnit.MILLISECONDS, Schedulers.io())
                  .map(s -> 
                       Observable.just("Rest " + s).delay(100, TimeUnit.MILLISECONDS));

          Observable.merge(first, rest).subscribe(immediates);

          Observable<String> result = Observable.switchOnNext(immediates);

          result.subscribe(System.out::println, 
            Throwable::printStackTrace, () -> System.out.println("Done"));

          System.out.println("Typed: One");
          source.onNext("One");
          if (firstWait == 1) {
              System.out.println("Waiting after one");
              Thread.sleep(300);
          }
          source.onNext("Two");
          source.onNext("Three");

          Thread.sleep(300);

          System.out.println("-----");
      }
  }
}

[dalewking]

I realize that there are ways to work around it, but it would be much more convenient to just have an overload with an extra parameter to specify whether to delay the first value after subscription

[headinthebox]

Then the next person wants the second value delayed, and the third one every other value.

Rx is a library that provides you with primitives that you can combine together to get what you want.
Every additional operator makes the library overweight and age faster. Ideally, the primitives are tricky to implement correctly and efficiently, look for instance at merge, but on top of that you plug things together like Lego bricks. That's the power of Rx.

My biggest nightmare is that Rx turns into a framework like Angular, that is floor wax, desert topping, and hair gel.

[benjchristensen]

Based on answer from @headinthebox I'm closing this out.

Here is another similar example (bufferedDebounce) of how a specific use case is accomplished by composing operators: https://gist.github.com/benjchristensen/e4524a308456f3c21c0b