Skip to content

F
FluxAndMonoErrorOperators
Commit ba13cc5e authored by Krishnakanth Balla's avatar Krishnakanth Balla
Browse files
Options

Initial Commint - Contains flux & mono error / functional operations

parents
Expand all Show whitespace changes
Inline Side-by-side
Showing with 928 additions and 0 deletions
.gitignore 0 → 100644
View file @ ba13cc5e
HELP.md
target/
!.mvn/wrapper/maven-wrapper.jar
!**/src/main/**/target/
!**/src/test/**/target/
### STS ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
.sts4-cache
### IntelliJ IDEA ###
.idea
*.iws
*.iml
*.ipr
### NetBeans ###
/nbproject/private/
/nbbuild/
/dist/
/nbdist/
/.nb-gradle/
build/
!**/src/main/**/build/
!**/src/test/**/build/
### VS Code ###
.vscode/
.mvn/wrapper/maven-wrapper.properties 0 → 100644
View file @ ba13cc5e
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.7/apache-maven-3.8.7-bin.zip
wrapperUrl=https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.1/maven-wrapper-3.1.1.jar
mvnw 0 → 100644
View file @ ba13cc5e
This diff is collapsed.
mvnw.cmd 0 → 100644
View file @ ba13cc5e
@REM ----------------------------------------------------------------------------
@REM Licensed to the Apache Software Foundation (ASF) under one
@REM or more contributor license agreements. See the NOTICE file
@REM distributed with this work for additional information
@REM regarding copyright ownership. The ASF licenses this file
@REM to you under the Apache License, Version 2.0 (the
@REM "License"); you may not use this file except in compliance
@REM with the License. You may obtain a copy of the License at
@REM
@REM https://www.apache.org/licenses/LICENSE-2.0
@REM
@REM Unless required by applicable law or agreed to in writing,
@REM software distributed under the License is distributed on an
@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@REM KIND, either express or implied. See the License for the
@REM specific language governing permissions and limitations
@REM under the License.
@REM ----------------------------------------------------------------------------
@REM ----------------------------------------------------------------------------
@REM Maven Start Up Batch script
@REM
@REM Required ENV vars:
@REM JAVA_HOME - location of a JDK home dir
@REM
@REM Optional ENV vars
@REM M2_HOME - location of maven2's installed home dir
@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a keystroke before ending
@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
@REM e.g. to debug Maven itself, use
@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
@REM ----------------------------------------------------------------------------
@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
@echo off
@REM set title of command window
title %0
@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on'
@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO%
@REM set %HOME% to equivalent of $HOME
if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
@REM Execute a user defined script before this one
if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
@REM check for pre script, once with legacy .bat ending and once with .cmd ending
if exist "%USERPROFILE%\mavenrc_pre.bat" call "%USERPROFILE%\mavenrc_pre.bat" %*
if exist "%USERPROFILE%\mavenrc_pre.cmd" call "%USERPROFILE%\mavenrc_pre.cmd" %*
:skipRcPre
@setlocal
set ERROR_CODE=0
@REM To isolate internal variables from possible post scripts, we use another setlocal
@setlocal
@REM ==== START VALIDATION ====
if not "%JAVA_HOME%" == "" goto OkJHome
echo.
echo Error: JAVA_HOME not found in your environment. >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
:OkJHome
if exist "%JAVA_HOME%\bin\java.exe" goto init
echo.
echo Error: JAVA_HOME is set to an invalid directory. >&2
echo JAVA_HOME = "%JAVA_HOME%" >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
@REM ==== END VALIDATION ====
:init
@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
@REM Fallback to current working directory if not found.
set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
set EXEC_DIR=%CD%
set WDIR=%EXEC_DIR%
:findBaseDir
IF EXIST "%WDIR%"\.mvn goto baseDirFound
cd ..
IF "%WDIR%"=="%CD%" goto baseDirNotFound
set WDIR=%CD%
goto findBaseDir
:baseDirFound
set MAVEN_PROJECTBASEDIR=%WDIR%
cd "%EXEC_DIR%"
goto endDetectBaseDir
:baseDirNotFound
set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
cd "%EXEC_DIR%"
:endDetectBaseDir
IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
@setlocal EnableExtensions EnableDelayedExpansion
for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
:endReadAdditionalConfig
SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
FOR /F "usebackq tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO (
IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
)
@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
if exist %WRAPPER_JAR% (
if "%MVNW_VERBOSE%" == "true" (
echo Found %WRAPPER_JAR%
)
) else (
if not "%MVNW_REPOURL%" == "" (
SET DOWNLOAD_URL="%MVNW_REPOURL%/org/apache/maven/wrapper/maven-wrapper/3.1.0/maven-wrapper-3.1.0.jar"
)
if "%MVNW_VERBOSE%" == "true" (
echo Couldn't find %WRAPPER_JAR%, downloading it ...
echo Downloading from: %DOWNLOAD_URL%
)
powershell -Command "&{"^
"$webclient = new-object System.Net.WebClient;"^
"if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^
"$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^
"}"^
"[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"^
"}"
if "%MVNW_VERBOSE%" == "true" (
echo Finished downloading %WRAPPER_JAR%
)
)
@REM End of extension
@REM Provide a "standardized" way to retrieve the CLI args that will
@REM work with both Windows and non-Windows executions.
set MAVEN_CMD_LINE_ARGS=%*
%MAVEN_JAVA_EXE% ^
%JVM_CONFIG_MAVEN_PROPS% ^
%MAVEN_OPTS% ^
%MAVEN_DEBUG_OPTS% ^
-classpath %WRAPPER_JAR% ^
"-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" ^
%WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
if ERRORLEVEL 1 goto error
goto end
:error
set ERROR_CODE=1
:end
@endlocal & set ERROR_CODE=%ERROR_CODE%
if not "%MAVEN_SKIP_RC%"=="" goto skipRcPost
@REM check for post script, once with legacy .bat ending and once with .cmd ending
if exist "%USERPROFILE%\mavenrc_post.bat" call "%USERPROFILE%\mavenrc_post.bat"
if exist "%USERPROFILE%\mavenrc_post.cmd" call "%USERPROFILE%\mavenrc_post.cmd"
:skipRcPost
@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
if "%MAVEN_BATCH_PAUSE%"=="on" pause
if "%MAVEN_TERMINATE_CMD%"=="on" exit %ERROR_CODE%
cmd /C exit /B %ERROR_CODE%
pom.xml 0 → 100644
View file @ ba13cc5e
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.11-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<releases>
<enabled>false</enabled>
</releases>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
</pluginRepositories>
</project>
src/main/java/com/example/demo/DemoApplication.java 0 → 100644
View file @ ba13cc5e
package com.example.demo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
src/main/java/com/example/demo/FluxErrorAndOtherMethods.java 0 → 100644
View file @ ba13cc5e
package com.example.demo;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import java.time.Duration;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
public class FluxErrorAndOtherMethods {
public Flux<String> stringFlux() {
return Flux.just("These","Strings","will","create","flux").log();
}
public Flux<String> stringFluxWithconcat() {
return Flux.just("These","Strings","will","create","flux")
.concatWith(Flux.just("new message")).log();
}
public Flux<String> stringFluxSorting() {
return Flux.just("These","Strings","will","create","flux")
.sort(Comparator.reverseOrder()).log();
}
public Flux<String> stringFluxDistinct() {
return Flux.just("These","These","Strings","Strings","will","will","create","flux")
.distinct()
.log();
}
public Flux<String> stringFluxFilter() {
return Flux.just("Ravi", "Ravendra", "Manoj", "Manish", "Mounika", "Devi", "Sandhya", "Prakash", "Aravind", "Basha", "Chaitu" )
.filter( s -> s.length() <= 5 ).log();
}
public Flux<String> stringFluxMap() {
return Flux.just("These","Strings","will","create","flux")
.map( s -> s.toUpperCase() ).log();
}
public Flux<String> stringFluxFlatMap() {
return Flux.just("These","Strings","will","create","flux")
.flatMap( s -> Flux.just(s.concat("-@1")) ).log();
}
public Flux<Integer> fluxOnErrorReturn() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.log()
.onErrorReturn( 100 ).log();
}
public Flux<Integer> fluxOnErrorReturnWithClassType() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.log()
.onErrorReturn( RuntimeException.class, 100 ).log();
}
public Flux<Integer> fluxOnErrorResume() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.onErrorResume( RuntimeException.class, ex -> Flux.range( 1, 3 ) ).log();
}
public Flux<Integer> fluxOnErrorMap() {
/*
Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.onErrorMap( RuntimeException.class, ex -> new RuntimeException("we messed up somewhere") ).log();
*/
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.onErrorMap( ex -> new RuntimeException("we messed up somewhere") ).log();
}
public Flux<Integer> fluxDoOnError() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.doOnError( ex -> new RuntimeException("we messed up somewhere") )
.log();
}
public Flux<Integer> fluxRetry() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.retry(3)
.log();
}
public Flux<Integer> fluxOnRetryWhen() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.retryWhen(Retry.max(3))
.log();
}
public Flux<Integer> fluxOnRetryWhenBackOff() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.retryWhen(Retry.backoff(2, Duration.ofSeconds(2)))
.log();
}
public Flux<Integer> fluxOnRetryWhenFixedDelay() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.retryWhen(Retry.fixedDelay(2, Duration.ofSeconds(5)))
.log();
}
public Mono<List<Integer>> fluxCollectList() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.collectList().log();
}
public Mono<List<Integer>> fluxCollectSortedList() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.collectSortedList( Comparator.reverseOrder() ).log();
}
public Mono<Map<String, String>> fluxCollectMap() {
return Flux.just( "krishna:8125486447", "ramu:8125719324", "balla:9908263541" )
.collectMap( s -> s.split(":")[0],
s -> s.split(":")[1] ).log();
}
public static void fluxBlockFirst() {
Flux.just( 1, 4, 5, 2, 0, 10 )
.delayElements( Duration.ofSeconds(5)).log()
.blockFirst( Duration.ofSeconds(10) );
}
public static void fluxBlockLast() {
Flux.just( 1, 4, 5, 2, 0, 10 )
.delayElements( Duration.ofSeconds(5)).log()
.blockLast( Duration.ofSeconds(300) );
}
public Flux<Integer> fluxOnErrorComplete() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.onErrorComplete( RuntimeException.class ).log();
}
public Flux<Integer> fluxOnErrorContinue() {
return Flux.just( 1, 4, 5, 2, 0, 10 )
.map( x -> 100/x )
.onErrorContinue( RuntimeException.class, (throwable, o) -> System.out.println( "Exception: "+ throwable.toString() ) ).log();
}
public static void main(String[] args) {
// fluxBlockFirst();
fluxBlockLast();
}
}
src/main/java/com/example/demo/FluxToMonoOperators.java 0 → 100644
View file @ ba13cc5e
package com.example.demo;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import java.time.Duration;
import java.util.stream.Collectors;
public class FluxToMonoOperators {
public static void main(String[] args) {
Flux<Integer> flux = Flux.just( 1, 4, 5, 2, 0, 10 );
Flux<Integer> flux1 = Flux.just( 100 );
flux.next().log().subscribe(System.out::println);
flux.last().log().subscribe(System.out::println);
Mono<Integer> m = Mono.from(flux).log();
m.subscribe();
flux.single().log().subscribe(System.out::println);
flux1.single().log().subscribe(System.out::println);
flux1.single(15).log().subscribe(System.out::println);
Flux.just().singleOrEmpty().log().subscribe(System.out::println);
flux.elementAt(3).log().subscribe(System.out::println);
flux.elementAt(8, 15 ).log().subscribe(System.out::println);
flux.any( integer -> integer <= 0 ).log().subscribe(System.out::println);
flux.all( integer -> integer > 0 ).log().subscribe(System.out::println);
flux.shareNext().log().subscribe(System.out::println);
flux.collect(Collectors.reducing( (i1, i2) -> i1 )).log().subscribe(System.out::println);
flux.collect(Collectors.reducing( (i1, i2) -> i1 )).map( o -> o.get() ).log().subscribe(System.out::println);
}
}
src/main/java/com/example/demo/MonoErrorMethods.java 0 → 100644
View file @ ba13cc5e
package com.example.demo;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
public class MonoErrorMethods {
public static void main(String[] args) {
// block
Mono.delay(Duration.ofSeconds(2))
.then(Mono.just("Krishna"))
.log()
.block(Duration.ofSeconds(3));
Mono.just( 100 )
.map( x -> x/0 )
.log()
.onErrorReturn( 100 ).log();
Mono.just( 100 )
.map( x -> x/0 )
.log()
.onErrorResume( ex -> Mono.empty() ).log();
Mono.just( 100 )
.map( x -> x/0 )
.log()
.onErrorComplete( RuntimeException.class ).log();
Mono.just( 100 )
.map( x -> x/0 )
.log()
.onErrorContinue( RuntimeException.class, (throwable, o) -> System.out.println( throwable.toString() ) ).log();
Mono.just( 100 )
.map( x -> x/0 )
.log()
.onErrorMap( ex -> new RuntimeException("we messed up somewhere") ).log();
Mono.just( 100 )
.map( x -> x/0 )
.log()
.doOnError( ex -> new RuntimeException("we messed up somewhere") ).log();
Mono.just( 100 )
.map( x -> x/0 )
.log()
.retry(3)
.log();
}
}
src/test/java/com/example/demo/DemoApplicationTests.java 0 → 100644
View file @ ba13cc5e
package com.example.demo;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
@SpringBootTest
class DemoApplicationTests {
FluxErrorAndOtherMethods sample;
@BeforeEach
void createInstance() {
System.out.println("createInstance() >> ");
sample = new FluxErrorAndOtherMethods();
}
@Test
void method_stringFlux() {
StepVerifier.create(sample.stringFlux())
.expectNext("These")
.expectNext("Strings")
.expectNext("will")
.expectNext("create")
.expectNext("flux")
.verifyComplete();
}
@Test
void method_stringFluxWithconcat() {
StepVerifier.create(sample.stringFluxWithconcat())
.expectNext("These")
.expectNext("Strings")
.expectNext("will")
.expectNext("create")
.expectNext("flux")
.expectNext("new message")
.verifyComplete();
}
@Test
void method_stringFluxSorting() {
StepVerifier.create(sample.stringFluxSorting())
.expectNext("will")
.expectNext("flux")
.expectNext("create")
.expectNext("These")
.expectNext("Strings")
.verifyComplete();
}
@Test
void method_stringFluxDistinct() {
StepVerifier.create(sample.stringFluxDistinct())
.expectNext("These")
.expectNext("Strings")
.expectNext("will")
.expectNext("create")
.expectNext("flux")
.verifyComplete();
}
@Test
void method_ConcatFlatMapImmediate() {
Flux<String> strFlux = Flux.just("These","Strings","will","create","flux" )
.concatWith(Flux.just("Hello", "World", "This", "is", "concat", "method"))
.delayElements(Duration.ZERO)
.flatMap( s -> Flux.just(s) )
.log();
StepVerifier.create(strFlux)
.expectNext("These","Strings","will","create","flux", "Hello", "World", "This", "is", "concat", "method" )
.verifyComplete();
}
@Test
void method_stringFluxFilter() {
StepVerifier.create(sample.stringFluxFilter())
.expectNext( "Ravi", "Manoj", "Devi", "Basha" )
.verifyComplete();
}
@Test
void method_stringFluxMap() {
StepVerifier.create(sample.stringFluxMap())
.expectNext( "THESE","STRINGS","WILL","CREATE","FLUX" )
.verifyComplete();
}
@Test
void method_stringFluxFlatMap() {
StepVerifier.create(sample.stringFluxFlatMap())
.expectNext( "These-@1","Strings-@1","will-@1","create-@1","flux-@1" )
.verifyComplete();
}
void method_ConcatArraySubscriberException() {
/*
.expectNext("THESE","STRINGS","CREATE","HELLO", "WORLD", "CONCAT", "METHOD", "SERVER", "MESSAGE" )
.thenAwait(Duration.ofMillis(100))
.expectError(RuntimeException.class)
.verify();
*/ }
@Test
void method_onErrorReturn() {
StepVerifier.create(sample.fluxOnErrorReturn())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.verifyComplete();
}
@Test
void method_onErrorReturnwithClass() {
StepVerifier.create(sample.fluxOnErrorReturnWithClassType())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.verifyComplete();
}
@Test
void method_onErrorComplete() {
StepVerifier.create(sample.fluxOnErrorComplete())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.verifyComplete();
}
@Test
void method_onErrorContinue() {
StepVerifier.create(sample.fluxOnErrorContinue())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 10 )
.verifyComplete();
}
@Test
void method_fluxDataToMap() {
Map<String, String> result = new HashMap<>();
result.put( "krishna", "8125486447");
result.put( "ramu", "8125719324");
result.put( "balla", "9908263541");
StepVerifier.create(sample.fluxCollectMap())
.expectNext(result)
.verifyComplete();
}
@Test
void method_onErrorResume() {
StepVerifier.create(sample.fluxOnErrorResume())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 1 )
.expectNext( 2 )
.expectNext( 3 )
.verifyComplete();
}
@Test
void method_onErrorMap() {
StepVerifier.create(sample.fluxOnErrorMap())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectError( RuntimeException.class )
.verify();
}
@Test
void method_doOnError() {
StepVerifier.create(sample.fluxDoOnError())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectError( RuntimeException.class )
.verify();
}
@Test
void method_retry() {
StepVerifier.create(sample.fluxRetry())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectError( RuntimeException.class )
.verify();
}
@Test
void method_retryWhen() {
StepVerifier.create(sample.fluxOnRetryWhen())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectError( RuntimeException.class )
.verify();
}
@Test
void method_retryBackOff() {
StepVerifier.create(sample.fluxOnRetryWhenBackOff())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectError( RuntimeException.class )
.verify();
}
@Test
void method_retryWhenWithFixedDuration() {
StepVerifier.create(sample.fluxOnRetryWhenFixedDelay())
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectNext( 100 )
.expectNext( 25 )
.expectNext( 20 )
.expectNext( 50 )
.expectError( RuntimeException.class )
.verify();
}
@Test
void method_collectList() {
StepVerifier.create(sample.fluxCollectList())
.expectNext(Arrays.asList(1, 4, 5, 2, 0, 10))
.verifyComplete();
}
@Test
void method_collectSortedList() {
StepVerifier.create(sample.fluxCollectSortedList())
.expectNext(Arrays.asList(10, 5, 4, 2, 1, 0))
.verifyComplete();
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment