Как можно разделить излучение наблюдаемого RxJava<Map<String, List> на несколько?

#java #android #functional-programming #rx-java2

#java #Android #функциональное программирование #rx-java2

Вопрос:

Я пытаюсь разделить излучение an RxJava.Observable<Map<String, List<Integer>> на несколько RxJava.Observable<Map<String, List<Integer>> . Каждое излучение должно включать раздел исходного списка в исходном порядке. Гарантируется, что каждый список имеет одинаковый размер и (list.size() % partitionSize) == 0 .

Пример:

 { 
    "a": [0, 1, 2, 3, 4, 5, 6, 7, 8], 
    "b": [0, 1, 2, 3, 4, 5, 6, 7, 8] 
}

Для

 // First emission
{ 
    "a": [0,1,2], 
    "b": [0,1,2] 
}
// Second emission
{ 
    "a": [3,4,5], 
    "b": [3,4,5] 
}
// Third emission
{ 
    "a": [6,7,8], 
    "b": [6,7,8] 
}

Заранее спасибо!

1. Вы хотите создать List новый разделенный Maps ?

2. Нет, я хочу испускать новые Map s с разделенными List s.

3. Я понимаю это, но мои вопросы в том, хотите ли вы, чтобы a List содержал новый Maps : List<Map<String, List<Integer>>

Ответ №1:

Я бы использовал streams для этой работы. Одним из способов было бы:

 Map<String, List<Integer>> map = Map.of(
        "a", List.of(0, 1, 2, 3, 4, 5, 6, 7, 8),
        "b", List.of(0, 1, 2, 3, 4, 5, 6, 7, 8));

int size = map.values().stream().mapToInt(list -> list.size()).max().getAsInt();

int partitionSize = IntStream.iterate(2, i -> i   1)
        .limit(size)
        .filter(i -> size % i == 0)
        .findFirst()
        .orElse(1);

int splitSize = Math.round(size / partitionSize);

List<Map<String, List<Integer>>> maps = IntStream.range(0, partitionSize)
        .mapToObj(i -> map.entrySet().stream().collect(
                Collectors.toMap(Map.Entry::getKey, 
                                entry -> entry.getValue().subList(i * splitSize, (i   1) * splitSize))))
        .collect(Collectors.toList());

maps.forEach(System.out::println);

Вывод:

 {a=[0, 1, 2], b=[0, 1, 2]}
{a=[3, 4, 5], b=[3, 4, 5]}
{a=[6, 7, 8], b=[6, 7, 8]}

Примечание: я предполагаю, что вы не знаете partitionSize , что все Lists они не являются ни null пустыми, ни пустыми, и все имеют одинаковый размер и list.size() > 1 .

1. Спасибо за ваш ответ! На самом деле я забыл упомянуть, что я хочу использовать RxJava.Observable which должен испускать (глядя на данный пример) 3 излучения для каждого нового Map , вместо того, чтобы испускать список новых Map s .

2. Извините, я не знаком RxJava.Observable и не могу помочь с этой частью. Надеюсь, кто-то другой сможет взять это отсюда 😉

Ответ №2:

Вы должны buffer(size) каждый список для данного аргумента partitionSize затем сопоставить для каждого отправленного буферизованного списка из каждого элемента в наборе ключей.

  fromIterable(map.entries).flatMap { entry ->
        fromIterable(entry.value).buffer(Math.min(entry.value.size, partionSize))
    }.flatMap {
        fromIterable(map.keys).flatMap { key -> just(key to it) }
    }.subscribe(::println)

Вывод:

 (a, [0, 1, 2])
(b, [0, 1, 2])
(a, [3, 4, 5])
(b, [3, 4, 5])
(a, [6, 7, 8])
(b, [6, 7, 8])
(a, [0, 1, 2])
(b, [0, 1, 2])
(a, [3, 4, 5])
(b, [3, 4, 5])
(a, [6, 7, 8])
(b, [6, 7, 8])