Определите количество эмиссий характеристических уведомлений по первой эмиссии

Можно добиться желаемого.

Самый простой способ - заменить Observable.combineLatest(...) на:

Observable.merge(
        rxBleConnection.writeCharacteristic(AP_SCAN_DATA, writeValue).ignoreElements(), // send the request but ignore the returned value
        apScanDataNotificationObservable.takeUntil(newResponseEndWatcher()) // take the response notifications until the response end watcher says so
);

Где newResponseEndWatcher() должен содержать логику для определения того, являются ли полученные значения ожидаемыми. Это может выглядеть так:

private Func1<byte[], Boolean> newResponseEndWatcher() {
    return new Func1<byte[], Boolean>() {

        private static final int NOT_INITIALIZED = -1;

        private int totalLength = NOT_INITIALIZED;
        private int receivedLength = NOT_INITIALIZED;

        @Override
        public Boolean call(byte[] bytes) {
            if (isNotInitialized(totalLength)) { // if it is the first received value
                // parse totalLength from the header
            }
            // update receivedLength
            return receivedLength >= totalLength;
        }

        private boolean isNotInitialized(int value) {
            return value == NOT_INITIALIZED;
        }
    };
}

Просто имейте в виду, что Func1, который является результатом newResponseEndWatcher(), имеет состояние. Если бы кто-то сохранил в переменной наблюдаемую, которая является результатом apScanDataNotificationObservable.takeUntil(newResponseEndWatcher()), следующие подписки могут закончиться преждевременно.

Чтобы смягчить эту проблему, можно использовать функцию Observable.using(), которая будет вызывать newResponseEndWatcher() каждый раз при подписке, а затем создавать новую apScanDataNotificationObservable.takeUntil(newResponseEndWatcher):

Observable.using(
        () -> newResponseEndWatcher(), // create a new response end watcher on each subscription
        responseEndWatcher -> apScanDataNotificationObservable.takeUntil(responseEndWatcher), // create the response observable that will complete properly
        responseEndWatcher -> { /* ignored, responseEndWatcher will get GCed eventually */ }
);