I have following `@Dao`, that provides `Flowable<User>` stream:

```
@Dao
interface UsersDao {
  @Query("SELECT * FROM users")
  fun loadUsers(): Flowable<List<User>>
}

```

I want the subscriber of the stream to receive updates of the database as soon as some change happens there. Subscribing to Room's `Flowable` I will get that feature out of the box.

What I want is following: if database is empty I want to perform a web request and save users into database. The subscriber will automatically receive new updates that had just happened. 

Now I want the client of the repository not to be aware all of the initialization logics: all he does - he performs `usersRepository.loadUsers()`. And all of these magic should take place inside the repository class:

```
class UsersRepository @Inject constructor(
    private val api: Api,
    private val db: UsersDao
) {

  fun loadUsers(): Flowable<List<User>> {
    ...
  }
}

```

Of course I can use following approach:

```
fun loadUsers(): Flowable<List<User>> {
  return db.loadTables()
      .doOnSubscribe {
        if (db.getCount() == 0) {
          val list = api.getTables().blockingGet()
          db.insert(list)
        }
      }
}

```

But I would like to construct the stream without using side-effects (`doOn...` operators). I've tried [`composing()`](http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#compose-io.reactivex.FlowableTransformer-) but that didn't help much. Been stuck on this for a while.