Selectively throttles items in an observable stream

rxThrottleMatchingItems.kt

fun <T> Observable<T>.throttleMatchingItems(
        windowDuration: Long,
        unit: TimeUnit,
        predicate: (T) -> Boolean
): Observable<T> {
        return this.compose { stream ->
                val throttledItems = stream
                        .filter(predicate)
                        .throttleFirst(windowDuration, unit)
                val allOtherItems = stream.filter { !predicate(it) }
                Observable.merge(throttledItems, allOtherItems)
        }
}