flatMapFirst: flatMap, который добавляет новую наблюдаемую только в том случае, если предыдущая закончилась

Вам нужен не новый оператор, а комбинация существующих:

source.onBackpressureLatest().flatMap(function, 1)

FlatMap будет запускать 1 внутренний источник одновременно, а onBackpressureLatest будет продолжать отбрасывать значения внешнего источника (кроме последних), если нет спроса, пока flatMap запускает 1 внутренний источник.

Если вы не хотите продолжать использовать последнюю версию, доступную из исходного кода, рассмотрите возможность использования вместо этого onBackpressureDrop.

Дополнение к ответу Дэвида. Если вы ищете способ иметь flatMapFirst в RxJava 2 Observable, а не Flowable, вот быстрая реализация Kotlin:

fun <T, R> Observable<T>.flatMapFirst(transform: (T) -> Observable<R>) =
    toFlowable(BackpressureStrategy.DROP)
        .flatMap({ transform(it).toFlowable(BackpressureStrategy.BUFFER) }, 1)
        .toObservable()

UPD. Альтернативная реализация, основанная на предложении Дэвида Карнока:

fun <T, R> Observable<T>.flatMapFirst(transform: (T) -> Observable<R>) =
    Observable.defer {
        val busy = AtomicBoolean()
        return@defer this
                .filter { busy.compareAndSet(false, true) }
                .flatMap {
                    transform(it).doAfterTerminate { busy.set(false) }
                }
    }

Мне удалось решить это с помощью этого:

/**
 * Flatmaps upstream items into [source] items.
 * Ignores upstream items if there is any [source] instance currently running.
 *
 * ```
 * upstream ----u-----u---u-------u---------------|-->
 *              ↓                 ↓               ↓
 * source       ---s-------|->    ---s-------|->  ↓
 *                 ↓                 ↓            ↓
 * result   -------s-----------------s------------|-->
 * ```
 */
fun <T, R> Observable<T>.flatMapWithDrop(source: Observable<R>): Observable<R> {
  return this.toFlowable(BackpressureStrategy.DROP)
    .flatMap({ source.toFlowable(BackpressureStrategy.MISSING) }, 1)
    .toObservable()
}