Как сделать группу и собрать с помощью RxJava и Kotlin?

Question

У меня Observable<Rates> и скорость всего лишь простой объект:

Rate(val value:String){} 
Rates(val rates: List<Rate>) 

и я хочу изменить, что Observable<Rates> в Observable<HashMap<String,Long>.

так, например, для ставок Rates(arrayOf(Rate("1"),Rate("2"), Rate("3"),Rate("3"), Rate("2"),Rate("2"))) я ожидать результат:

(1 -> 1) 
(2 -> 3) 
(3 -> 2) 
(4 -> 0) 
(5 -> 0) 

я начинаю создавать что-то вроде этого:

service.getRates() 
     .flatMap {it-> Observable.from(it.rates) } 
     .filter { !it.value.isNullOrEmpty() } 
     .groupBy {it -> it.value} 
     .collect({ HashMap<String,Long>()}, { b, t -> b.put(t.key, t.count???)} 

, но я застрял здесь, и я не знаю, рассчитывать все значения? и я не знаю, как добавить пустые значения (0), если нет 5 из 4. Есть ли способ сделать это с помощью rx?

Answer 1

Хитрость заключается в том, чтобы использовать count на GroupedObservable, поскольку она излучает только одно значение, когда источник наблюдаемых завершается:

Отсюда следует:

rates 
    .flatMap { Observable.from(it.rates) } 
    .filter { !it.value.isNullOrEmpty() } 
    .groupBy { it.value } 
    .flatMap { group -> group.count().map { group.key to it } } // list "1"->1, "2"->3, ... 
    .mergeWith(Observable.from((1..5).map { it.toString() to 0 })) // defaults "4"->0 
    .reduce(mutableMapOf<String, Int>()) { acc, cur -> 
     acc.apply { 
     val (key, count) = cur 
     this[key] = (this[key] ?: 0) + count // add counts 
     } 
    }.subscribe { countedRates -> 
    println(countedRates) 
} 

Answer 2

Пожалуйста, смотрите на комментарии в коде для ответов на ваш вопрос.

import rx.Observable 

fun main(args: Array<String>) { 
    val service = Service() 

    // This adds all keys with each key mapped to zero 
    val referenceKeyCounts = Observable 
     .just("1", "2", "3", "4", "5") 
     .map { it to 0 } 

    val keyCountsFromService = service.getRates() 
     .flatMap { Observable.from(it.rates) } 
     .filter { !it.value.isNullOrEmpty() } 
     .map { it.value to 1 } // map each occurrence of key to 1 

    Observable.concat(referenceKeyCounts, keyCountsFromService) 
     .groupBy { it.first } 
     .flatMap { group -> // this converts GroupedObservable to final values 
      group.reduce(0, { acc, pair -> acc + pair.second }) // add instead of counting 
       .map { group.key to it } 
     } 
     .subscribe(::println) 

} 

class Service { 
    fun getRates(): Observable<Rates> = Observable.just(Rates(listOf(
     Rate("1"), Rate("2"), Rate("3"), Rate("3"), Rate("2"), Rate("2") 
    ))) 
} 

class Rate(val value: String) 

class Rates(val rates: List<Rate>) 

Answer 3

Я думаю, что это больше о функциональном программировании, а не RxJava связанного Вопрос

Реализовать функцию отображения из Rates -> Map<String,Int>.

хитрости: Объединить два списка Pair<String,Int> для формирования Map<String,Int>

val ratesToMapWithEmptyValues: (Rates) -> Map<String, Int> = { source -> 

    //TODO: Just for demo 
    val validRatesValue = arrayOf("1","2","3","4","5") 

    mapOf(
    *validRatesValue.map { it to 0 }.toTypedArray(), 
    *source.rates.groupBy(Rate::value).mapValues { it.value.size }.toList().toTypedArray() 
) 

} 

Применить функцию в Observable.map

service.getRates() 
     .map(ratesToMapWithEmptyValues)