Spring WebFlux Flux Tutorial Examples

by

Discover the capabilities of Spring WebFlux Flux with this comprehensive tutorial. Gain insights into creating, manipulating, and transforming Flux streams effectively using practical examples. Develop proficiency in asynchronous, non-blocking programming principles and elevate your Spring application development expertise.

Flux: It’s a reactive stream in Spring WebFlux that can emit 0 or N items over time. It represents a sequence of data elements and is commonly used for handling streams of data that may contain multiple elements or continuous data flows.

Example: 

Flux<User> fluxUsers = Flux.just(Arrays.asList(new User("John"),new User("Alice"),new User("Bob")));

Set up a Spring Boot project using Spring Initializr or any IDE of your choice, and make sure to include the following dependency:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

Example: Create FluxService.java class

This class, FluxService, offers various methods illustrating different functionalities of Flux streams.

  • getFlux(): Generates a Flux stream with predefined strings, serving as static data or example data.
  • getFluxList(): Converts collections of User objects into reactive streams, enabling integration with reactive programming.
  • filterFlux(): Filters elements in the Flux stream based on specific conditions, enabling selective processing.
  • flatMapExample(): Demonstrates asynchronous processing of each element in the Flux stream using flatMap.
  • tranformFluxExample(): Illustrates Flux transformation using the transform operator, promoting modularity and maintainability.
  • defaultIfEmptyExample(String str): Handles empty Flux streams gracefully by providing a default value based on a provided condition.
  • getBlankFlux(): Returns an empty Flux stream, useful as a placeholder or starting point for further processing.
@Service
public class FluxService {

    public Flux<String> getFlux() {
        return Flux.just("First", "Second", "Third", "Fourth", "Fifth");
    }

    public Flux<User> getFluxList() {
        return Flux.fromIterable(Arrays.asList(new User("Pavan", "pavan@123"),
                        new User("Kiran", "kiran@123")))
                .cast(User.class);
    }
    
    public Flux<String> filterFlux() {
        return getFlux().filter(data -> data.equals("CCC"));
    }

    public Flux<String> flatMapExample() {
        return getFlux().flatMap(data -> Flux.just(data)).delayElements(Duration.ofMillis(3000));
    }

    /*
        while you can achieve similar results without transform by chaining operators directly on the Flux,
        transform provides a cleaner, more modular approach to defining and applying Flux transformations,
        promoting code reuse, readability, and maintainability.
     */

    public void tranformFluxExample() {
        Flux<Integer> originalFlux = Flux.range(1, 10);
        //without transform method
        Flux<Integer> integerFlux = originalFlux.map(i -> i * 2).filter(i -> i % 3 != 0).publishOn(Schedulers.parallel());
        integerFlux.subscribe(data -> System.out.println(data)); //2 4 6 8 10 14 16 20

        //with transform method.
        Flux<Integer> transformedFlux = originalFlux.transform(flux -> {
            return flux.map(i -> i * 2).filter(i -> i % 3 != 0).subscribeOn(Schedulers.parallel());
        });
        transformedFlux.subscribe(data -> System.out.println(data));
    }

    public Flux<String> defaultIfEmptyExample(String str) {
        return getFlux().filter(data -> data.contains(str)).defaultIfEmpty("Doesn't contians: " + str);
    }

    public Flux<Object> getBlankFlux() {
        return Flux.empty();
    }

}

Test the functionality of the FluxService class by running the following test methods for Spring WebFlux Flux. 

@SpringBootTest
public class FluxServiceTest {

    @Autowired
    private FluxService fluxService;

    @Test
    void testFlux() {
        fluxService.getFlux().subscribe(data -> {
            System.out.println(data);
        });
    }

    @Test
    void testGetFluxList() {
        fluxService.getFluxList().subscribe(data -> {
            System.out.println(data);
        });
    }

    @Test
    void testFilter() {
        Disposable subscribe = fluxService.filterFlux().subscribe(System.out::println);
        System.out.println("filtered text: " + subscribe); // CCC
    }

    @Test
    void testFlatMap() {
        fluxService.flatMapExample().subscribe(data -> {
            System.out.println("FlatMap: " + data);
        });
    }

    @Test
    void tranformFluxExample() {
        fluxService.tranformFluxExample();
    }

    @Test
    void ifExample() {
        Flux<String> flux = fluxService.defaultIfEmptyExample("Third");
        flux.subscribe(data->{
            System.out.println("data: "+data);
        });
        StepVerifier.create(flux).expectNext("Third").verifyComplete();
    }

    @Test
    void getBlankFlux() {
        Flux<Object> blankFlux = fluxService.getBlankFlux();
        StepVerifier.create(blankFlux).expectNext().verifyComplete();
    }
}

Output

Spring WebFlux Flux

Conclusion

In short, this article gave a hands-on look at Spring WebFlux Flux, showing how it works with easy examples. By getting a grasp of Flux’s role in reactive programming and trying out its functions, developers can use it to make Spring apps that react quickly and handle big loads. We made sure our code was solid by testing it well, setting the stage for effective and sturdy software building.

Leave a Comment