Reactor – How to convert Flux into List, Map

Tutorial: “Flux to List – How to convert Reactor Flux to List/Map”

What is a Reactor Flux? A Flux is a Reactive Streams Publisher , augmented with a lot of operators that can be used to generate, transform, orchestrate Flux sequences. It can emit 0 to n elements ( onNext event) then either completes or errors ( onComplete and onError terminal events).
In this tutorial, I introduces ways to convert Reactor Flux into List/Map.

Related post:

– Reactor – Handle Error
– Reactor – How to Create Flux (Publisher) with Interval

I. Ways to convert Flux to List, Map

We will use Flux methods such as:
– collectList(): accumulate sequence into a Mono<List>.
-> Collect all elements emitted by this Flux into a List that is emitted by the resulting Mono when this sequence completes.

– collectSortedList(): accumulate sequence and sort into a Mono<List>.
-> Collect all elements emitted by this Flux until this sequence completes, and then sort them in natural order into a List that is emitted by the resulting Mono.

– collectMap(): convert sequence into a Mono<Map>.
-> Collect all elements emitted by this Flux into a hashed Map that is emitted by the resulting Mono when this sequence completes. The key is extracted from each element by applying the keyExtractor Function. In case several elements map to the same key, the associated value will be the most recently emitted element.

– collectMultimap(): convert sequence into a Mono<Map> that each Map’s key can be paired with multi-value (in a Collection).
-> Collect all elements emitted by this Flux into a multimap that is emitted by the resulting Mono when this sequence completes. The key is extracted from each element by applying the keyExtractor Function, and every element mapping to the same key is stored in the List associated to said key.

Then the Mono result above will be converted into a real List, Map using block() method.

Initialization Reactor Flux

Reactor installation in Maven

– First, import the BOM by adding the following to pom.xml:

<dependencyManagement>
	<dependencies>
		<dependency>
			<groupId>io.projectreactor</groupId>
			<artifactId>reactor-bom</artifactId>
			<version>Aluminium-SR1</version>
			<type>pom</type>
			<scope>import</scope>
		</dependency>
	</dependencies>
</dependencyManagement>

– Next, add dependency:

<dependencies>
	<dependency>
		<groupId>io.projectreactor</groupId>
		<artifactId>reactor-core</artifactId>
	</dependency>
</dependencies>

Declare & Initialize Reactor Flux

Flux<String> flux = Flux.just(
						"Site_0:loizenai.com", 
						"Description_0:Java Technology",
						"Description_1:Spring Framework");

Convert Flux to List

Flux collectList() method

List<String> list1 = flux.collectList().block();
list1.forEach(System.out::println);

// Site_0:loizenai.com
// Description_0:Java Technology
// Description_1:Spring Framework

Reactor Flux collectSortedList() method

List<String> list2 = flux.collectSortedList().block();
list2.forEach(System.out::println);

// Description_0:Java Technology
// Description_1:Spring Framework
// Site_0:loizenai.com

Convert Flux to Map

Flux collectMap method

Map<String, String> map1 = flux
							.collectMap(
									item -> {return item.split(":")[0];},
									item -> {return item.split(":")[1];})
							.block();
map1.forEach((key, value) -> System.out.println(key + " -> " + value));

// Site_0 -> loizenai.com
// Description_1 -> Spring Framework
// Description_0 -> Java Technology

Flux collectMultimap() method

Map<String, Collection<String>> map2 = flux
										.collectMultimap(
												item -> {return item.split("_[0-9]+:")[0];},
												item -> {return item.split(":")[1];})
										.block();
map2.forEach((key, value) -> System.out.println(key + " -> " + value));

// Site -> [loizenai.com]
// Description -> [Java Technology, Spring Framework]

II. Source Code

1. Technology

– Java 8
– Maven 3.6.1
– Reactor Core 3.0.4, with the Aluminium release train.

2. Code – Reactor Flux to List

package com.loizenai.fluxcollection;

import java.util.Collection;
import java.util.List;
import java.util.Map;

import reactor.core.publisher.Flux;

public class MainApp {

	public static void main(String[] args) {

		Flux<String> flux = Flux.just(
								"Site_0:loizenai.com", 
								"Description_0:Java Technology",
								"Description_1:Spring Framework");

		System.out.println("=== flux.collectList() ===");
		List<String> list1 = flux.collectList().block();
		list1.forEach(System.out::println);

		System.out.println("=== flux.collectSortedList() ===");
		List<String> list2 = flux.collectSortedList().block();
		list2.forEach(System.out::println);

		System.out.println("=== flux.collectMap() ===");
		Map<String, String> map1 = flux
									.collectMap(
											item -> {return item.split(":")[0];},
											item -> {return item.split(":")[1];})
									.block();
		map1.forEach((key, value) -> System.out.println(key + " -> " + value));
		
		System.out.println("=== flux.collectMultimap() ===");
		Map<String, Collection<String>> map2 = flux
												.collectMultimap(
														item -> {return item.split("_[0-9]+:")[0];},
														item -> {return item.split(":")[1];})
												.block();
		map2.forEach((key, value) -> System.out.println(key + " -> " + value));
		
	}
}

3. Results

=== flux.collectList() ===
Site_0:loizenai.com
Description_0:Java Technology
Description_1:Spring Framework
=== flux.collectSortedList() ===
Description_0:Java Technology
Description_1:Spring Framework
Site_0:loizenai.com
=== flux.collectMap() ===
Site_0 -> loizenai.com
Description_1 -> Spring Framework
Description_0 -> Java Technology
=== flux.collectMultimap() ===
Site -> [loizenai.com]
Description -> [Java Technology, Spring Framework]

Further Reading – Reactor Programming

– Flux Reator

Last updated on March 13, 2021.