RxAsyncAwait.java

AsyncAwait.dart

// ref. http://news.dartlang.org/2015/03/dart-19-release-youve-been-await-ing-for.html?m=1
Future<int> getCount(bool showCustomers) async {
    try {
        var database = await connectDb();
        var count = await database.queryEmployees().length;
        if (showCustomers) {
            count += await database.queryCustomers().length;
        }
        return count;
    } on DbError catch (error, stack) {
        print("DB Error: $error, $statck");
        return null;
    }
}

Future<Database> connectDb();
Future<List<User>> Database.queryEmployees();
Future<List<User>> Database.queryCustomers();

RxAsyncAwait.java

Observable<Integer> getCount(boolean showCustomers) {
    try {
        Database database = connectDb().toBlocking().single();
        Observable<User> users = database.queryEmployees();
        if (showCustomers) users.concatWith(database.queryCustomers());
        return users.count();
    } catch (DbError e) {
        e.printStackTrace();
        return Observable.just(null);
    }
    return Observable.empty();
}

Observable<Database> connectDb();
Observable<User> Database.queryEmployees();
Observable<User> Database.queryCustomers();

RxAsyncAwait2.java

// await vs. toBlocking().single()
Observable<Integer> getCount(boolean showCustomers) {
    return Observable.just(showCustomers).map(b -> {
        Database database = connectDb().toBlocking().single();
        int count = database.queryEmployees().count().toBlocking().single();
        if (b) {
            count += database.queryCustomers().count().toBlocking().single();
        }
        return count;
    }).onErrorResumeNext(e -> {
        if (e instanceof DbError) {
            e.printStackTrace();
            return Observable.just(null);
        }
        return Observable.error(e);
    });
}

RxAsyncAwait3.java

// with flatMap()
Observable<Integer> getCount(boolean showCustomers) {
    return Observable.just(showCustomers)
        .flatMap(b -> {
            Database database = connectDb().toBlocking().single();
            Observable<User> users = database.queryEmployees();
            if (b) users.concatWith(database.queryCustomers());
            return users.count();
        })
        .onErrorResumeNext(e -> {
            if (e instanceof DbError) {
                e.printStackTrace();
                return Observable.just(null);
            }
            return Observable.error(e);
        });
}

RxAsyncAwait4.java

// without toBlocking().single()
Observable<Integer> getCount(final boolean showCustomers) {
    return connectDb()
        .flatMap(database -> {
            Observable<User> users = database.queryEmployees();
            if (showCustomers) users.concatWith(database.queryCustomers());
            return users.count();
        })
        .onErrorResumeNext(e -> {
            if (e instanceof DbError) {
                e.printStackTrace();
                return Observable.just(null);
            }
            return Observable.error(e);
        });
}

RxAsyncAwait5.java

// without final by zip()
Observable<Integer> getCount(boolean showCustomers) {
    return Observable.zip(Observable.just(showCustomers), connectDb(), (b, database) -> {
            Observable<User> users = database.queryEmployees();
            if (b) users.concatWith(database.queryCustomers());
            return users.count();
        })
        .flatMap(count -> count)
        .onErrorResumeNext(e -> {
            if (e instanceof DbError) {
                e.printStackTrace();
                return Observable.just(null);
            }
            return Observable.error(e);
        });
}