¿Cómo crear un intervalo infinito observable que emitirá un nuevo object cada intervalo de time?

Estoy tratando de crear un Observable que emitirá nuevo object cada segundo. Así que por ahora solo estoy comprimiendo un Observable, que emite un número finito de objects de una list, con un intervalo Observable.

val list1 = mutableListOf<SomeClass1>( SomeClass1("1", 1), SomeClass1("2", 2), SomeClass1("3", 3), SomeClass1("4", 4), SomeClass1("5", 5), SomeClass1("6", 6), SomeClass1("7", 7), SomeClass1("8", 8), SomeClass1("9", 9) ) val someClass1Observable = Observable .fromIterable(list1) .zipWith(Observable.interval(2, TimeUnit.SECONDS), BiFunction { item: SomeClass1, _: Long -> item }) 

Entonces, ¿cómo hacer que esto sea infinito, de manera que cada 2 segundos habrá un nuevo object de class de la class SomeClass1 con algunos parameters?

Aquí está la respuesta para mi pregunta, que satisface mi:

  override fun onCreate(savedInstanceState: Bundle?) { super.onCreate(savedInstanceState) setContentView(R.layout.activity_main) var counter = 0 val periodicSomeClass1 = Observable.interval(1, TimeUnit.SECONDS) .flatMap( { counter++ Observable.just(SomeClass1("$counter", counter)) } ) periodicSomeClass1.subscribe { t: SomeClass1 -> Log.v("periodicSomeClass1", t.toString()) } } 
  • Cómo burlarse del repository reactivo que devuelve Observable
  • Observables opcionales en combinar
  • Cómo controlar el flujo sin .flatMap, que rompe una stream reactiva que impide que operadores como distinctUntilChanged trabajen en toda la secuencia
  • Cómo escribir la testing adecuada para el repository de interfaz reactiva que devuelve Observable solo cuando hay algún evento, cómo simular el desencadenamiento de ese evento
  • ¿Cómo especificar la versión de RxJava al usar RxKotlin?
  • Cambie Flowable <List <Obj1 >> a Flowable <List <Obj2 >> en la habitación
  • Single.zip completando antes de Success
  • Completable.andThen resultados en "other is null"
  • No se puede comprimir Rxjava Observables
  • Extraño comportamiento de componer y ObservableTransformer en RxJava cuando el genérico pasado se extiende
  • ¿Comportamiento incorrecto de Maybe <List <T >> en Room?