Upload examples

reactor-operators/.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src/main/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
+		<attributes>
+			<attribute name="test" value="true"/>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="output" path="target/classes"/>
+</classpath>

reactor-operators/.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>reactor-operators</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.springframework.ide.eclipse.boot.validation.springbootbuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+	</natures>
+</projectDescription>

reactor-operators/.settings/org.eclipse.core.resources.prefs

+eclipse.preferences.version=1
+encoding//src/main/java=UTF-8
+encoding//src/test/java=UTF-8
+encoding/<project>=UTF-8

reactor-operators/.settings/org.eclipse.jdt.core.prefs

+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
+org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
+org.eclipse.jdt.core.compiler.compliance=1.8
+org.eclipse.jdt.core.compiler.debug.lineNumber=generate
+org.eclipse.jdt.core.compiler.debug.localVariable=generate
+org.eclipse.jdt.core.compiler.debug.sourceFile=generate
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
+org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.release=disabled
+org.eclipse.jdt.core.compiler.source=1.8

reactor-operators/.settings/org.eclipse.m2e.core.prefs

+activeProfiles=
+eclipse.preferences.version=1
+resolveWorkspaceProjects=true
+version=1

reactor-operators/.settings/org.springframework.ide.eclipse.prefs

+boot.validation.initialized=true
+eclipse.preferences.version=1

reactor-operators/pom.xml

+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.nisum</groupId>
+	<artifactId>reactor-operators</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>reactor-operators</name>
+	<url>http://maven.apache.org</url>
+
+	<properties>
+		<java.version>1.8</java.version>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+	</properties>
+
+
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<version>2.1.5.RELEASE</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<version>2.1.5.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<version>3.2.9.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+</project>

reactor-operators/src/main/java/com/nisum/App.java

+package com.nisum;
+
+/**
+ * Hello world!
+ *
+ */
+public class App 
+{
+    public static void main( String[] args )
+    {
+        System.out.println( "Hello World!" );
+    }
+}

reactor-operators/src/main/java/com/nisum/domain/Address.java

+package com.nisum.domain;
+
+public class Address {
+
+	private String address;
+
+	public Address() {
+		super();
+	}
+
+	public Address(String address) {
+		super();
+		this.address = address;
+	}
+
+	public String getAddress() {
+		return address;
+	}
+
+	public void setAddress(String address) {
+		this.address = address;
+	}
+
+	@Override
+	public int hashCode() {
+		return address.hashCode();
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		Address other = (Address)obj;
+		return address.equals(other.getAddress());
+	}
+	
+	
+
+
+	
+	
+
+}

reactor-operators/src/main/java/com/nisum/domain/OnErrorTest.java

+package com.nisum.domain;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+
+public class OnErrorTest {
+	public static void main(String[] args) throws Exception{
+		Flux<Integer> f = Flux.range(1, 10);
+		f = f.map(i -> {
+			if(i==9) {
+				throw new RuntimeException();
+			}
+			return i;
+		});
+		f.subscribe(SubscriberUtil.getSubscriber());
+		
+		Thread.sleep(1000*50);
+	}
+}

reactor-operators/src/main/java/com/nisum/domain/Order.java

+package com.nisum.domain;
+
+import java.util.List;
+
+public class Order {
+
+	private String orderId;
+	private List<Address> addressList;
+	
+	
+
+	public Order() {
+		super();
+	}
+
+	public Order(String orderId, List<Address> addressList) {
+		super();
+		this.orderId = orderId;
+		this.addressList = addressList;
+	}
+
+	public String getOrderId() {
+		return orderId;
+	}
+
+	public void setOrderId(String orderId) {
+		this.orderId = orderId;
+	}
+
+	public List<Address> getAddressList() {
+		return addressList;
+	}
+
+	public void setAddressList(List<Address> addressList) {
+		this.addressList = addressList;
+	}
+
+	@Override
+	public String toString() {
+		return "Order [orderId=" + orderId + ", addressList=" + addressList + "]";
+	}
+
+	
+}

reactor-operators/src/main/java/com/nisum/flow/FlowTest.java

+package com.nisum.flow;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+
+public class FlowTest {
+
+	public static void main(String[] args) {
+		System.out.println("-------------- start-main");
+		Flux<String> f = m1().map(m -> m + "modified by main");
+		System.out.println("-------------- end-main");
+		f.subscribe(SubscriberUtil.getStrSubscriber());
+	}
+
+	public static Flux<String> m1() {
+		System.out.println("-------------- start-m1");
+		Flux<String> f = m2().map(m -> m + "modified by m1");
+		System.out.println("-------------- end-m1");
+		return f;
+	}
+
+	public static Flux<String> m2() {
+		System.out.println("-------------- start-m2");
+		Flux<String> f = Flux.just("1", "2", "3").map(m -> m + "modified by m2");
+		System.out.println("-------------- end-m2");
+		return f;
+	}
+
+}

reactor-operators/src/main/java/com/nisum/operators/create/flux/CreatingFlux.java

+package com.nisum.operators.create.flux;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+
+public class CreatingFlux {
+	
+	public static void main(String[] args) throws InterruptedException {
+
+		Flux<Integer> flux = null;
+		
+		flux = justWithSomeValues();
+//		flux = justByCallingAMethod();
+//		flux = fromArray();
+//		flux = fromIterable();
+//		flux = range();
+//		flux = fromStream();
+
+		System.out.println("==== Before subscribe ====");
+		flux.subscribe(SubscriberUtil.getSubscriber());
+
+	}
+
+	public static Flux<Integer> justWithSomeValues() {
+		return Flux.just(1, 2, 3, 4, 5);
+	}
+
+	public static Flux<Integer> justByCallingAMethod() {
+		return Flux.just(getData());
+	}
+
+	public static Flux<Integer> fromArray() {
+		Integer[] data = new Integer[] { 11, 12, 13, 14, 15 };
+		return Flux.fromArray(data);
+	}
+
+	public static Flux<Integer> fromIterable() {
+		List<Integer> data = Arrays.asList(21, 22, 23, 24, 25);
+		return Flux.fromIterable(data);
+	}
+
+	public static Flux<Integer> range() {
+		return Flux.range(1, 20);
+	}
+
+	public static Flux<Integer> fromStream() {
+		List<Integer> data = Arrays.asList(21, 22, 23, 24, 25);
+		return Flux.fromStream(data.stream());
+	}
+
+	public static Integer[] getData() {
+		System.out.println("===== Into Method that prepares data ====");
+		return new Integer[] { 11, 12, 13, 14, 15 };
+	}
+}

reactor-operators/src/main/java/com/nisum/operators/create/mono/CreatingMono.java

+package com.nisum.operators.create.mono;
+
+import java.util.Optional;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Mono;
+
+public class CreatingMono {
+
+	public static void main(String[] args) throws InterruptedException {
+
+		Mono<Integer> mono = null;
+
+//		mono = justWithSomeValues();
+//		mono = justByCallingAMethod();
+//		mono = fromRunnable();
+//		mono = justOrEmptyWithOptional(true);
+//		mono = justOrEmptyWithOptional(false);
+//		mono = justOrEmptyWithValue(10);
+//		mono = justOrEmptyWithValue(null);
+		mono = fromCallable();
+//		mono = Mono.empty();
+
+		System.out.println("==== Before subscribe ====");
+		
+		mono.subscribe(SubscriberUtil.getSubscriber());
+		
+	}
+
+	public static Mono<Integer> justWithSomeValues() {
+		return Mono.just(1);
+	}
+
+	public static Mono<Integer> justByCallingAMethod() {
+		return Mono.just(getData());
+	}
+
+	public static Mono<Integer> fromRunnable() {
+		
+		return Mono.fromRunnable(() -> {
+			try {
+				Thread.sleep(1000 * 5L);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+			System.out.println("==== End Of run method ====");
+		});
+	}
+
+	public static Mono<Integer> justOrEmptyWithOptional(boolean flag) {
+		return flag ? Mono.justOrEmpty(Optional.of(1)) : Mono.justOrEmpty(Optional.empty());
+	}
+
+	public static Mono<Integer> justOrEmptyWithValue(Integer value) {
+		return Mono.justOrEmpty(value);
+	}
+
+	public static Mono<Integer> fromCallable() {
+		return Mono.fromCallable(() -> {			
+			Thread.sleep(1000*10);
+			System.out.println("==== Getting data from Callable ====");
+			return 20;
+		});
+	}
+
+	public static int getData() {
+		System.out.println("===== Into Method that prepares data ====");
+		return 1;
+	}
+
+}

reactor-operators/src/main/java/com/nisum/operators/error/ErrorOperators.java

+package com.nisum.operators.error;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class ErrorOperators {
+
+	public static void main(String[] args) {
+		
+//		errorEvent();
+
+//		Fallback value example 
+//		onErrorReturnWithTryCatch();
+		onErrorReturn();
+//		onErrorReturnWithPredicate();
+
+//		onErrorResumeTryCatch();
+//		onErrorResume();
+//		onErrorResumeWithPredicate();
+
+		throwBusinessExceptionWithTryCatch();
+		throwBusinessExceptionUsingError();
+
+		onErrorMapForBusinessException();
+
+	}
+
+	public static void errorEvent() {
+
+		Flux.just(1, 2, 3, 4).map(n -> {
+				if (n == 2)
+					throw new RuntimeException("Sample Message");
+				return n;
+			})
+			.onErrorMap(ex -> new RuntimeException("Modified"))
+
+			.onErrorReturn(10)
+
+			.onErrorResume(ex -> {
+				return Flux.just(50, 60, 70);
+			})
+
+			.subscribe(SubscriberUtil.getSubscriber());
+
+	}
+
+	private static void onErrorMapForBusinessException() {
+		Mono<String> myMono = Mono.just("myMessage").map(myMessage -> getFromExternalService(myMessage, true))
+//				We are converting the existing runtime exception to MyBusinessException 	
+				.onErrorMap(runtimeEx -> new MyBusinessException());
+		myMono.subscribe(null,
+				exception -> System.out.println("throwBusinessExceptionUsingError -- " + exception.getMessage()));
+	}
+
+	private static void throwBusinessExceptionWithTryCatch() {
+		try {
+			getBusinessException("myMessage");
+		} catch (MyBusinessException exception) {
+			System.out.println("throwBusinessExceptionWithTryCatch -- " + exception.getMessage());
+		}
+	}
+
+	private static void throwBusinessExceptionUsingError() {
+		Mono<String> myMono = Mono.just("myMessage").map(myMessage -> getFromExternalService(myMessage, true))
+//			We are converting the existing runtime exception to MyBusinessException 	
+				.onErrorResume(runtimeEx -> Mono.error(new MyBusinessException()));
+		myMono.subscribe(null,
+				exception -> System.out.println("throwBusinessExceptionUsingError -- " + exception.getMessage()));
+	}
+
+	private static String getBusinessException(String myMessage) {
+
+		String myString = "empty";
+		try {
+			myString = getFromExternalService("My message", true);
+		} catch (Exception e) {
+			throw new MyBusinessException();
+		}
+		return myString;
+
+	}
+
+// Return the publisher..
+	private static void onErrorResume() {
+		String key = "on-resume-invocation";
+		Mono.just(key).map(message -> getFromExternalService(message, true))
+				.onErrorResume(exception -> Mono.just(getFromLocalCache(key))).subscribe(System.out::println);
+	}
+
+	private static void onErrorResumeWithPredicate() {
+		String key = "on-resume-invocation-with-predicate";
+		Mono.just(key).map(message -> getFromExternalService(message, true))
+				.onErrorResume(ArithmeticException.class, exception -> Mono.just(getFromLocalCache(key)))
+				.onErrorResume(RuntimeException.class, exception -> Mono.just(getFromLocalCache(key)))
+				.subscribe(System.out::println);
+	}
+
+	private static void onErrorResumeTryCatch() {
+		String message = "try-catch-invocation";
+		try {
+			message = getFromExternalService(message, true);
+		} catch (Exception e) {
+			message = getFromLocalCache(message);
+		}
+		System.out.println(message);
+
+	}
+
+	private static String getFromLocalCache(String message) {
+		return message + " -- The Data is coming from local cache..";
+	}
+
+	private static String getFromExternalService(String message, boolean isThrowException) {
+		if (isThrowException) {
+			throw new RuntimeException("Error in retrieving the data from external service");
+		}
+		return "The message is from external service";
+	}
+
+	private static String onErrorReturnWithTryCatch() {
+		String myFallbackValue = "";
+		try {
+			throwErrorOnSix(6);
+		} catch (Throwable error) {
+			myFallbackValue = "onErrorReturnWithTryCatch -- Oops! some thing happen...";
+		}
+		return myFallbackValue;
+	}
+
+	// Return the Object
+	private static void onErrorReturn() {
+		Flux<String> fluxMessage = Flux.range(0, 7).map(ErrorOperators::throwErrorOnSix)
+				.onErrorReturn("onErrorReturn -- Oops! some thing happen...");
+		fluxMessage.subscribe(System.out::println);
+	}
+
+	private static void onErrorReturnWithPredicate() {
+		Flux<String> fluxMessage = Flux.range(0, 7).map(ErrorOperators::throwErrorOnSix)
+				.onErrorReturn(ArithmeticException.class,
+						"onErrorReturnWithPredicate Arthematic-- Oops! some thing happen...")
+				.onErrorReturn(RuntimeException.class,
+						"onErrorReturnWithPredicate Runtime-- Oops! some thing happen...");
+		fluxMessage.subscribe(System.out::println);
+	}
+
+	private static String throwErrorOnSix(int count) {
+		if (count == 6) {
+			throw new RuntimeException("Some random exception");
+		}
+		return "Everything-Working-Fine";
+	}
+
+}

reactor-operators/src/main/java/com/nisum/operators/error/HandlingError.java

+package com.nisum.operators.error;
+
+import java.time.Duration;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class HandlingError {
+
+	public static void main(String[] args) {
+//		tryCatch();
+		error();
+//		onErrorMap();
+//		onErrorReturn();
+//		onErrorResume();
+	}
+
+	public static void tryCatch() {
+
+		Flux<Integer> flux = null;
+
+		try {
+			flux = Flux.range(1, 20)
+					.delayElements(Duration.ofMillis(1000 * 5))
+					.map(n -> {
+				if (n == 7)
+					throw new RuntimeException("ErrorMessage");
+				return n;
+			});
+			
+			flux.subscribe(SubscriberUtil.getSubscriber());
+			
+			
+		} catch (Throwable t) {
+			System.out.println("=== Into catch block");
+		}
+		
+		
+		
+		
+		System.out.println("------------------- out of try-catch");
+		
+		
+		try {
+			Thread.sleep(1000 * 300);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+	}
+
+	/*
+	 * onSubscribe 1,2,3,4,5,6, ------------------- out of try-catch
+	 */
+
+	public static void error() {
+		
+		
+		Mono<Integer> mono = Mono.error(new RuntimeException("ErrorMessage"));
+		
+		mono.subscribe(SubscriberUtil.getSubscriber());
+	}
+
+	public static void onErrorMap() {
+
+		Flux.just(1, 2, 3, 4)
+		.map(n -> {
+			if (n == 2)
+				throw new RuntimeException("ErrorMessage");
+			return n;
+		})
+		.onErrorMap(ex -> new RuntimeException("Modified-ErrorMessage"))
+		.subscribe(SubscriberUtil.getSubscriber());
+
+	}
+
+	public static void onErrorReturn() {
+
+		Flux.just(1, 2, 3, 4).map(n -> {
+			if (n == 2)
+				throw new RuntimeException("ErrorMessage");
+			return n;
+		}).onErrorReturn(9)
+		.subscribe(SubscriberUtil.getSubscriber());
+
+	}
+
+	public static void onErrorResume() {
+
+		Flux.just(1, 2, 3, 4).map(n -> {
+
+			if (n == 2)
+				throw new RuntimeException("ErrorMessage");
+			return n;
+
+		}).onErrorResume(ex -> {
+			return Flux.just(50, 60, 70);
+		}).subscribe(SubscriberUtil.getSubscriber());
+
+	}
+}

reactor-operators/src/main/java/com/nisum/operators/error/MyBusinessException.java

+package com.nisum.operators.error;
+
+public class MyBusinessException extends RuntimeException {
+
+	private static final long serialVersionUID = 1L;
+	
+	private static final String CUSTOM_MESSAGE = "MyBusinessException Message";
+	
+	public MyBusinessException() {
+		super(CUSTOM_MESSAGE);
+	}
+
+	public MyBusinessException(String message, Throwable cause, boolean enableSuppression,
+			boolean writableStackTrace) {
+		super(message, cause, enableSuppression, writableStackTrace);
+	}
+
+	public MyBusinessException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public MyBusinessException(String message) {
+		super(message);
+	}
+
+	public MyBusinessException(Throwable cause) {
+		super(CUSTOM_MESSAGE, cause);
+	}
+	
+	
+
+}

reactor-operators/src/main/java/com/nisum/operators/error/OnNextTest.java

+package com.nisum.operators.error;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+
+public class OnNextTest {
+	public static void main(String[] args) {}
+}

reactor-operators/src/main/java/com/nisum/operators/filter/FilteringSequence.java

+package com.nisum.operators.filter;
+
+import com.nisum.domain.Address;
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class FilteringSequence {
+
+	public static void main(String[] args) {
+
+		
+		//filter();
+		distinct();
+		
+	}
+
+	public static void filter() {
+		Flux.range(1, 26)
+		
+		.map(i-> {
+			System.out.println(i);
+			return i;
+		})
+		.filter(n -> n % 2 == 0)
+		.subscribe(SubscriberUtil.getSubscriber());
+	}
+
+	public static void distinct() {
+		Flux.just(new Address("Kothaguda"), 
+				new Address("Madhapur"), 
+				new Address("Madhapur"),
+				new Address("Ameerpet"),
+				new Address("Ameerpet"))		
+		.distinct()
+		.subscribe(SubscriberUtil.getAddrSubscriber());
+	}
+
+}

reactor-operators/src/main/java/com/nisum/operators/time/WorkingWithTime.java

+package com.nisum.operators.time;
+
+import java.time.Duration;
+
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class WorkingWithTime {
+
+	public static void main(String[] args) {		
+		delay();
+		//timeout();
+	}
+
+	public static void delay() {
+		Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6, 7)
+				
+				.delayElements(Duration.ofMillis(1000 * 3));
+		
+		flux.subscribe(SubscriberUtil.getSubscriber());
+		System.out.println("====================");
+		sleep();
+	}
+
+	public static void timeout() {
+		Flux.just(1, 2, 3, 4, 5, 6, 7).map(n -> {
+			
+			try {
+				Thread.sleep(1000*2);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+			
+			return n * 2;
+		}).timeout(Duration.ofMillis(1000*4)).subscribe(SubscriberUtil.getSubscriber());
+		
+		sleep();
+	}
+
+	public static void sleep() {
+		try {
+			Thread.sleep(1000 * 60);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+	}
+}

reactor-operators/src/main/java/com/nisum/operators/transform/TransformingPublisher.java

+package com.nisum.operators.transform;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.nisum.domain.Address;
+import com.nisum.domain.Order;
+import com.nisum.util.SubscriberUtil;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+
+public class TransformingPublisher {
+
+	protected static List<Order> orders = Arrays.asList(
+			new Order("order1", Arrays.asList(new Address("Kothaguda"), new Address("Madhapur"))),
+			new Order("order2", Arrays.asList(new Address("Ameerpet"), new Address("Secunderabad"))));
+	
+	
+
+	public static void main(String[] args) {
+		
+		
+		
+	
+		
+		
+		
+		
+//		map();
+//		flatMap();
+//		concatWith();
+//		concat();
+		zip();
+//		zipWith();
+		
+//		zipMono();
+//		zipMonoList();	
+		
+		
+				
+		
+	}
+
+	public static void map() {
+		
+		Flux<Order> strFlux = Flux.fromIterable(orders);
+		
+		Flux<String> intFlux = strFlux.map(order -> order.getOrderId());
+		
+		intFlux.subscribe(SubscriberUtil.getStrSubscriber());
+	}
+
+	public static void flatMap() {
+		
+		Flux<Order> strFlux = Flux.fromIterable(orders);
+		
+		
+		//Flux<List<Address>> a = strFlux.map(order-> order.getAddressList());
+		
+		Flux<Address> intFlux = strFlux.flatMap(order -> Flux.fromIterable(order.getAddressList()));
+	
+		intFlux.subscribe(SubscriberUtil.getAddrSubscriber());
+	}
+
+	public static void concatWith() {
+		Flux<Integer> flux1 = Flux.just(1, 2, 3, 4, 5);
+		Flux<Integer> flux2 = Flux.just(6, 7, 8, 9, 10);
+		
+		//Flux.concat(sources)
+		
+		//Flux<Integer> f3 = flux1.concatWith(flux2);
+		
+		Flux<Integer> f3 = flux2.mergeWith(flux1);
+		
+		f3.subscribe(SubscriberUtil.getSubscriber());
+		//flux1.concatWith(flux2).subscribe(SubscriberUtil.getSubscriber());
+	}
+
+	public static void concat() {
+		Flux<Integer> flux1 = Flux.just(1, 2, 3, 4, 5);
+		Flux<Integer> flux2 = Flux.just(6, 7, 8, 9, 10);
+		Flux.concat(flux1, flux2).subscribe(SubscriberUtil.getSubscriber());
+	}
+
+	public static void zip() {
+		
+		Flux<Integer> flux1 = Flux.just(1, 2, 3);
+		
+		Flux<String> flux2 = Flux.just("One", "Two", "Three", "Four");
+		
+		Flux<Tuple2<Integer, String>> tuple = Flux.zip(flux1, flux2);
+		
+				
+		Flux.zip(flux1, flux2).doOnNext(t -> {
+			System.out.println(t.getT1() + "--------" + t.getT2());
+		}).subscribe();
+	}
+	
+	public static void zipMono() {
+		Mono<Integer> Mono1 = Mono.just(1).map(n -> n * 2);
+		Mono<String> Mono2 = Mono.just("One");
+		//Mono<String> Mono2 = Mono.empty();
+		
+		//Mono<Tuple2<Integer, String>> tuple = Mono.zip(Mono1,Mono2);=		
+		
+		//Flux.zip(combinator, sources)
+		
+		Mono.zip(Mono1,Mono2).doOnNext(t -> {
+			System.out.println(t.getT1() + "--------" + t.getT2());
+		}).subscribe();
+	}
+	
+	public static void zipMonoList() {
+		
+		
+		Mono<List<Integer>> Mono1= Flux.just(1, 2, 3).map(n -> n * 2).collectList();
+		//Mono<List<String>> Mono2 = Flux.just("One", "Two", "Three", "Four").collectList();
+		Mono<List<Object>> Mono2 = Flux.empty().collectList();
+		
+
+		
+		//Mono<Integer> Mono1 = Mono.just(1).map(n -> n * 2);
+		//Mono<String> Mono2 = Mono.just("One");
+		//Mono<String> Mono2 = Mono.empty();
+		
+		//Mono<Tuple2<Integer, String>> tuple = Mono.zip(Mono1,Mono2);=		
+		
+		Mono.zip(Mono1,Mono2).doOnNext(t -> {
+			System.out.println(t.getT1() + "--------" + t.getT2());
+		}).subscribe();
+	}
+
+	public static void zipWith() {
+		Flux<Integer> flux1 = Flux.just(1, 2, 3).map(n -> n * 2);
+		Flux<String> flux2 = Flux.just("One", "Two", "Three");
+		flux2.zipWith(flux1).doOnNext(t -> {
+			System.out.println(t.getT1() + "--------" + t.getT2());
+		}).subscribe();
+	}
+
+	public static void error() {
+		
+		Flux.just(1, 2, 3, 4).map(n -> {
+			if (n == 2)
+				throw new RuntimeException("Sample Message");
+			return n;
+		})
+		
+		.onErrorMap(ex-> new RuntimeException("Modified"))
+		
+		.onErrorReturn(10)
+		
+		.onErrorResume(ex -> {
+			return Flux.just(50, 60, 70);
+		})
+		
+		.subscribe(SubscriberUtil.getSubscriber());
+		
+	}
+	
+	
+
+}

reactor-operators/src/main/java/com/nisum/synchronus/BlockTest.java

+package com.nisum.synchronus;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class BlockTest {
+	public static void main(String[] args) {
+		String j = Mono.just("abc").block();
+		System.out.println(j);
+		System.out.println("----------mono");
+		int i = Flux.range(1,5).blockLast();
+		System.out.println("----------flux");
+		System.out.println(i);
+
+	}
+}

reactor-operators/src/main/java/com/nisum/util/SubscriberUtil.java

+package com.nisum.util;
+
+import org.reactivestreams.Subscriber;
+
+
+import org.reactivestreams.Subscription;
+
+import com.nisum.domain.Address;
+public class SubscriberUtil {
+
+	public static Subscriber<Integer> getSubscriber() {
+
+		return new Subscriber<Integer>() {
+
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(1000);
+				System.out.println("==== Received onSubscribe event/notification ====");
+			}
+
+			@Override
+			public void onNext(Integer t) {
+				System.out.println(String.format("==== Received onNext event/notification with value %d ====", t));
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				System.out.println("==== Received Error event/notification ===="+t.getMessage());
+
+			}
+
+			@Override
+			public void onComplete() {
+				System.out.println("==== Received onComplete event/notification ====");
+			}
+
+		};
+	}
+
+	public static Subscriber<String> getStrSubscriber() {
+
+		return new Subscriber<String>() {
+
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(1000);
+				System.out.println("**** Received onSubscribe event/notification ****");
+			}
+
+			@Override
+			public void onNext(String t) {
+				System.out.println(String.format("**** Received onNext event/notification with value %s ****", t));
+
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				System.out.println("**** Received Error event/notification ****");
+
+			}
+
+			@Override
+			public void onComplete() {
+				System.out.println("**** Received onComplete event/notification ****");
+			}
+
+		};
+	}
+	
+	public static Subscriber<Address> getAddrSubscriber() {
+		return new Subscriber<Address>() {
+
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(1000);
+				System.out.println("$$$$ Received onSubscribe event/notification $$$$");				
+			}
+
+			@Override
+			public void onNext(Address t) {
+				System.out.println(String.format("$$$$ Received onNext event/notification with value %s $$$$", t.getAddress()));
+				
+			}
+
+			@Override
+			public void onError(Throwable t) {
+				System.out.println("$$$$ Received Error event/notification $$$$");
+			}
+
+			@Override
+			public void onComplete() {
+				System.out.println("$$$$ Received onComplete event/notification $$$$");
+			}
+		};
+	}
+}

reactor-operators/src/main/java/com/nisum/webclient/OrderSubscriber.java

+package com.nisum.webclient;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import com.nisum.domain.Order;
+
+public class OrderSubscriber implements Subscriber<Order> {
+
+	@Override
+	public void onSubscribe(Subscription s) {
+		s.request(25);
+		System.out.println("Subscribed ==== ");
+	}
+
+	@Override
+	public void onNext(Order t) {
+		System.out.println("============== Order Details =============");
+		System.out.println(t);
+	}
+
+	@Override
+	public void onError(Throwable t) {
+		System.out.println("Received onError event");
+		
+	}
+
+	@Override
+	public void onComplete() {
+		System.out.println("Received onComplete event");
+	}
+
+}

reactor-operators/src/main/java/com/nisum/webclient/WebClientTest.java

+package com.nisum.webclient;
+
+import java.time.Duration;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Flux;
+
+public class WebClientTest {
+	public static void main(String[] args) throws InterruptedException {
+		Flux<Order> orders = WebClient.create("http://localhost:8883/order-list/reactive").get().retrieve().bodyToFlux(Order.class).timeout(Duration.ofMillis(1000*100));
+		orders.subscribe(new OrderSubscriber());
+		Thread.sleep(1000*100);
+	}
+}

reactor-operators/src/test/java/com/nisum/reactor_operators/AppTest.java

+package com.nisum.reactor_operators;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+/**
+ * Unit test for simple App.
+ */
+public class AppTest 
+    extends TestCase
+{
+    /**
+     * Create the test case
+     *
+     * @param testName name of the test case
+     */
+    public AppTest( String testName )
+    {
+        super( testName );
+    }
+
+    /**
+     * @return the suite of tests being tested
+     */
+    public static Test suite()
+    {
+        return new TestSuite( AppTest.class );
+    }
+
+    /**
+     * Rigourous Test :-)
+     */
+    public void testApp()
+    {
+        assertTrue( true );
+    }
+}

reactor-operators/target/classes/META-INF/MANIFEST.MF

+Manifest-Version: 1.0
+Built-By: sgali
+Build-Jdk: 1.8.0_151
+Created-By: Maven Integration for Eclipse
+

reactor-operators/target/classes/META-INF/maven/com.nisum/reactor-operators/pom.properties

+#Generated by Maven Integration for Eclipse
+#Fri Apr 23 22:08:06 IST 2021
+version=0.0.1-SNAPSHOT
+groupId=com.nisum
+m2e.projectName=reactor-operators
+m2e.projectLocation=C\:\\Users\\sivaji\\Documents\\WebFluxDemo\\WebFlux\\reactor-operators
+artifactId=reactor-operators

reactor-operators/target/classes/META-INF/maven/com.nisum/reactor-operators/pom.xml

+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.nisum</groupId>
+	<artifactId>reactor-operators</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>reactor-operators</name>
+	<url>http://maven.apache.org</url>
+
+	<properties>
+		<java.version>1.8</java.version>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+	</properties>
+
+
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<version>2.1.5.RELEASE</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<version>2.1.5.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<version>3.2.9.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+</project>

webflux-webclient/.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src/main/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
+		<attributes>
+			<attribute name="test" value="true"/>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jdk1.8.0_151">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="output" path="target/classes"/>
+</classpath>

webflux-webclient/.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>webflux-webclient</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.springframework.ide.eclipse.boot.validation.springbootbuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+	</natures>
+</projectDescription>

webflux-webclient/pom.xml

+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.nisum</groupId>
+	<artifactId>webflux-webclient</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>webflux-webclient</name>
+	<url>http://maven.apache.org</url>
+
+	<properties>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<version>2.4.0</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<version>2.1.5.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+	
+		<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+	
+</project>

webflux-webclient/src/main/java/com/nisum/App.java

+package com.nisum;
+
+/**
+ * Hello world!
+ *
+ */
+public class App 
+{
+    public static void main( String[] args )
+    {
+        System.out.println( "Hello World!" );
+    }
+}

webflux-webclient/src/main/java/com/nisum/body/BodyTest.java

+package com.nisum.body;
+
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+import com.nisum.domain.OrderDetails;
+
+import reactor.core.publisher.Mono;
+
+public class BodyTest {
+	public static void main(String[] args) {
+		
+		String baseUrl = "http://localhost:8883";
+		String uri = "order/body";
+		
+		OrderDetails orderDetails = new OrderDetails("order-1-body", "completed");
+		
+		Mono<Order> mono = WebClient.create(baseUrl).post().uri(uri)
+				
+				.body(BodyInserters.fromObject(orderDetails))
+				
+				.retrieve().bodyToMono(Order.class);
+
+		Order order = mono.block();
+		
+		System.out.println(order);
+	}
+}

webflux-webclient/src/main/java/com/nisum/domain/Address.java

+package com.nisum.domain;
+
+public class Address {
+
+	private String line1;
+	private String line2;
+	private String state;
+	private String country;
+	private String zip;
+
+	public Address() {
+		super();
+	}
+
+	public Address(String line1, String line2, String state, String country, String zip) {
+		super();
+		this.line1 = line1;
+		this.line2 = line2;
+		this.state = state;
+		this.country = country;
+		this.zip = zip;
+	}
+
+	public String getLine1() {
+		return line1;
+	}
+
+	public void setLine1(String line1) {
+		this.line1 = line1;
+	}
+
+	public String getLine2() {
+		return line2;
+	}
+
+	public void setLine2(String line2) {
+		this.line2 = line2;
+	}
+
+	public String getState() {
+		return state;
+	}
+
+	public void setState(String state) {
+		this.state = state;
+	}
+
+	public String getCountry() {
+		return country;
+	}
+
+	public void setCountry(String country) {
+		this.country = country;
+	}
+
+	public String getZip() {
+		return zip;
+	}
+
+	public void setZip(String zip) {
+		this.zip = zip;
+	}
+
+	@Override
+	public String toString() {
+		return "Address [line1=" + line1 + ", line2=" + line2 + ", state=" + state + ", country=" + country + ", zip="
+				+ zip + "]";
+	}
+
+	
+}

webflux-webclient/src/main/java/com/nisum/domain/Item.java

+package com.nisum.domain;
+
+public class Item {
+
+	private String itemId;
+	private int quantity;
+	private int price;
+
+	public Item() {
+		super();
+	}
+
+	public Item(String itemId, int quantity, int price) {
+		super();
+		this.itemId = itemId;
+		this.quantity = quantity;
+		this.price = price;
+	}
+
+	public String getItemId() {
+		return itemId;
+	}
+
+	public void setItemId(String itemId) {
+		this.itemId = itemId;
+	}
+
+	public int getQuantity() {
+		return quantity;
+	}
+
+	public void setQuantity(int quantity) {
+		this.quantity = quantity;
+	}
+
+	public int getPrice() {
+		return price;
+	}
+
+	public void setPrice(int price) {
+		this.price = price;
+	}
+
+	@Override
+	public String toString() {
+		return "Item [itemId=" + itemId + ", quantity=" + quantity + ", price=" + price + "]";
+	}
+	
+	
+
+}

webflux-webclient/src/main/java/com/nisum/domain/Order.java

+package com.nisum.domain;
+
+import java.util.List;
+
+public class Order {
+
+	private String orderId;
+	private List<Item> items;
+	private Address address;
+
+	public Order() {
+		super();
+	}
+
+	public Order(String orderId, List<Item> items, Address address) {
+		super();
+		this.orderId = orderId;
+		this.items = items;
+		this.address = address;
+	}
+
+	public String getOrderId() {
+		return orderId;
+	}
+
+	public void setOrderId(String orderId) {
+		this.orderId = orderId;
+	}
+
+	public List<Item> getItems() {
+		return items;
+	}
+
+	public void setItems(List<Item> items) {
+		this.items = items;
+	}
+
+	public Address getAddress() {
+		return address;
+	}
+
+	public void setAddress(Address address) {
+		this.address = address;
+	}
+
+	@Override
+	public String toString() {
+		return "Order [orderId=" + orderId + ", items=" + items + ", address=" + address + "]";
+	}
+	
+	
+
+}

webflux-webclient/src/main/java/com/nisum/domain/OrderDetails.java

+package com.nisum.domain;
+
+public class OrderDetails {
+	
+	private String orderId;
+	private String status;
+
+	public OrderDetails() {
+		super();
+	}
+
+	public OrderDetails(String orderId, String status) {
+		super();
+		this.orderId = orderId;
+		this.status = status;
+	}
+
+	public String getOrderId() {
+		return orderId;
+	}
+
+	public void setOrderId(String orderId) {
+		this.orderId = orderId;
+	}
+
+	public String getStatus() {
+		return status;
+	}
+
+	public void setStatus(String status) {
+		this.status = status;
+	}
+
+	@Override
+	public String toString() {
+		return "OrderDetails [orderId=" + orderId + ", status=" + status + "]";
+	}
+	
+	
+
+}

webflux-webclient/src/main/java/com/nisum/domain/OrderStatus.java

+package com.nisum.domain;
+
+public class OrderStatus {
+	
+	private String orderId;
+	private String status;
+	
+	public OrderStatus() {
+		super();
+	}
+	
+	public OrderStatus(String orderId, String status) {
+		super();
+		this.orderId = orderId;
+		this.status = status;
+	}
+	
+	public String getOrderId() {
+		return orderId;
+	}
+	public void setOrderId(String orderId) {
+		this.orderId = orderId;
+	}
+	public String getStatus() {
+		return status;
+	}
+	public void setStatus(String status) {
+		this.status = status;
+	}
+
+	@Override
+	public String toString() {
+		return "OrderStatus [orderId=" + orderId + ", status=" + status + "]";
+	}
+	
+	
+	
+}

webflux-webclient/src/main/java/com/nisum/domain/Person.java

+package com.nisum.domain;
+public class Person {
+	
+	private int id;
+	private String name;
+	
+	
+	public Person(int id, String name) {
+		super();
+		this.id = id;
+		this.name = name;
+	}
+	
+	public Person() {		
+	}
+
+	
+	public int getId() {
+		return id;
+	}
+	public void setId(int id) {
+		this.id = id;
+	}
+	public String getName() {
+		return name;
+	}
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public String toString() {
+		return "Person [id=" + id + ", name=" + name + "]";
+	}
+	
+}

webflux-webclient/src/main/java/com/nisum/exchange/ExchangeTest.java

+package com.nisum.exchange;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class ExchangeTest {
+
+	public static void main(String[] args) {
+
+		String baseUrl = "http://localhost:8883";
+		String uri = "order-details";
+		
+		Mono<Order> mono = WebClient.create(baseUrl).get().uri(uri).exchange().flatMap(clientResponse -> {
+			
+			System.out.println("==== Status Code " + clientResponse.rawStatusCode());
+			
+			return clientResponse.bodyToMono(Order.class);
+		});
+		
+		Order order = mono.block();
+		System.out.println(order);
+		
+	}
+
+}

webflux-webclient/src/main/java/com/nisum/headers/HeadersBuilderTest.java

+package com.nisum.headers;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.http.MediaType;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Mono;
+
+public class HeadersBuilderTest {
+
+	public static void main(String[] args) {
+
+		// http://localhost:8883/order/header
+
+		Map<String, List<String>> headersMap = new HashMap<>();
+		buildHeaderMap("orderId", "order-13-header-builder", headersMap);
+		buildHeaderMap("Accept", "application/x-www-form-urlencoded", headersMap);
+	
+
+		MultiValueMap<String, String> multiValueHeaderMap = new LinkedMultiValueMap<>(headersMap);
+
+		String endPoint = "order/header";
+
+		Mono<Order> mono = WebClient.create("http://localhost:8884")
+					.get()
+					.uri(endPoint)
+					
+					.headers(headersConsumer -> {
+							headersConsumer.addAll(multiValueHeaderMap);
+						})
+					.accept(MediaType.APPLICATION_JSON)
+					.retrieve().bodyToMono(Order.class);;
+
+
+		Order order = mono.block();
+		System.out.println(order);
+
+	}
+	
+
+	public static void buildHeaderMap(String headerName, String headerValue, Map<String, List<String>> headerMap) {
+		List<String> orderIds = new ArrayList<>();
+		orderIds.add(headerValue);
+		headerMap.put(headerName, orderIds);
+	}
+
+}

webflux-webclient/src/main/java/com/nisum/headers/HeadersTest.java

+package com.nisum.headers;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Mono;
+
+public class HeadersTest {
+
+	public static void main(String[] args) {
+
+		// http://localhost:8883/order/header
+		String endPoint = "order/header";
+
+		Mono<Order> mono = WebClient.create("http://localhost:8883")
+					.get()
+					.uri(endPoint)
+					.header("orderId", "order-12-header")
+					.retrieve().bodyToMono(Order.class);;
+
+		Order order = mono.block();
+		System.out.println(order);
+
+	}
+	
+}

webflux-webclient/src/main/java/com/nisum/nonblocking/NonBlockingTest.java

+package com.nisum.nonblocking;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Person;
+
+import reactor.core.publisher.Mono;
+
+public class NonBlockingTest {
+	public static void main(String[] args) {
+		Mono<Person> p = WebClient.create("http://localhost:8081/person").get().retrieve().bodyToMono(Person.class);
+		System.out.println("------------1-----------");
+		System.out.println("------------2-----------");
+		System.out.println("------------3-----------");
+		System.out.println(p.subscribe(pr -> {
+			System.out.println(pr.getName());
+		}));
+		
+		try {
+			Thread.sleep(1000*60);
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+	}
+}

webflux-webclient/src/main/java/com/nisum/person/client/Student.java

+package com.nisum.person.client;
+
+public class Student {
+	
+	int _id;
+	String name;
+	
+	public int get_id() {
+		return _id;
+	}
+	public void set_id(int _id) {
+		this._id = _id;
+	}
+	public String getName() {
+		return name;
+	}
+	public void setName(String name) {
+		this.name = name;
+	}
+	
+	
+}

webflux-webclient/src/main/java/com/nisum/person/client/StudentSubscriber.java

+package com.nisum.person.client;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public class StudentSubscriber implements Subscriber<Student>{
+
+	@Override
+	public void onSubscribe(Subscription s) {
+		System.out.println("subscribed");
+		
+	}
+
+	@Override
+	public void onNext(Student t) {
+		System.out.println(t._id);
+		
+	}
+
+	@Override
+	public void onError(Throwable t) {
+		System.out.println("onError");
+		
+	}
+
+	@Override
+	public void onComplete() {
+		System.out.println("onComplete");
+		
+	}
+
+}

webflux-webclient/src/main/java/com/nisum/person/client/WebClientTest.java

+package com.nisum.person.client;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class WebClientTest {
+	public static void main(String[] args) {
+		WebClient webClient = WebClient.create("http://localhost:8081/get-person-by-id?id=1");
+		Mono<Student> mono = webClient.get().retrieve().bodyToMono(Student.class);
+		mono.subscribe(new StudentSubscriber());
+	}
+}

webflux-webclient/src/main/java/com/nisum/queryparams/QueryParamBuilderTest.java

+package com.nisum.queryparams;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriBuilder;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Mono;
+
+public class QueryParamBuilderTest {
+
+	public static void main(String[] args) {
+
+		// http://localhost:8883/order?orderId=order-10
+		
+		Map<String, List<String>> queryParamsMap = new HashMap<>();
+		buildQueryParamMap("orderId", "order-10-queryparam-builder", queryParamsMap);
+		
+		String endPoint = "order";
+
+		Mono<Order> mono = WebClient.create("http://localhost:8883")
+					.get()
+					
+					.uri(uriBuilder -> buildUri(uriBuilder, endPoint, queryParamsMap))
+					
+					.retrieve()
+					.bodyToMono(Order.class);
+		
+		Order order = mono.block();
+		System.out.println(order);
+	}
+
+	public static URI buildUri(UriBuilder uriBuilder, String uri, Map<String, List<String>> queryParams) {
+		
+		UriBuilder builder = uriBuilder.path(uri);
+		MultiValueMap<String, String> queryParamsMap = new LinkedMultiValueMap<>(queryParams);
+		
+		if (queryParams != null) {
+			builder.queryParams(new LinkedMultiValueMap<>(queryParamsMap));
+		}
+		return builder.build();
+	}
+	
+	public static void buildQueryParamMap(String queryParam, String value, Map<String, List<String>> queryParamsMap) {
+		List<String> orderIds = new ArrayList<>();
+		orderIds.add(value);
+		queryParamsMap.put(queryParam, orderIds);
+	}
+}

webflux-webclient/src/main/java/com/nisum/queryparams/QueryParamTest.java

+package com.nisum.queryparams;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Mono;
+
+public class QueryParamTest {
+	public static void main(String[] args) {
+		
+		// http://localhost:8883/order?orderId=order-10
+
+		String queryParamName = "orderId";
+		String queryParamValue = "order-10-queryparam";
+
+		String endPoint = String.format("order?%s=%s", queryParamName, queryParamValue);
+
+		Mono<Order> mono = WebClient.create("http://localhost:8883").get().uri(endPoint).retrieve()
+				.bodyToMono(Order.class);
+
+		Order order = mono.block();
+		System.out.println(order);
+		
+	}
+}

webflux-webclient/src/main/java/com/nisum/subscribers/OrderSubscriber.java

+package com.nisum.subscribers;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import com.nisum.domain.Order;
+
+public class OrderSubscriber implements Subscriber<Order> {
+
+	@Override
+	public void onSubscribe(Subscription s) {
+		s.request(25);
+		System.out.println("Subscribed ==== ");
+	}
+
+	@Override
+	public void onNext(Order t) {
+		System.out.println("============== Order Details =============");
+		System.out.println(t);
+	}
+
+	@Override
+	public void onError(Throwable t) {
+		System.out.println("Received onError event");
+		
+	}
+
+	@Override
+	public void onComplete() {
+		System.out.println("Received onComplete event");
+	}
+
+}

webflux-webclient/src/main/java/com/nisum/urivariables/UriVariablesBuilderTest.java

+package com.nisum.urivariables;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriBuilder;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Mono;
+
+public class UriVariablesBuilderTest {
+	public static void main(String[] args) {
+		
+		// http://localhost:8883/order/order-11
+
+		Map<String, String> uriVariables = new HashMap<>();
+		uriVariables.put("orderId", "order-11-uribuilder");
+
+		String endPoint = "order/{orderId}";
+
+		Mono<Order> mono = WebClient.create("http://localhost:8883").get()
+				.uri(uriBuilder -> buildUri(uriBuilder, endPoint, uriVariables)).retrieve().bodyToMono(Order.class);
+
+		Order order = mono.block();
+		System.out.println(order);
+
+	}
+
+	public static URI buildUri(UriBuilder uriBuilder, String uri, Map<String, String> uriVariables) {
+		
+		UriBuilder builder = uriBuilder.path(uri);
+
+		if (uriVariables != null) {
+			return builder.build(uriVariables);
+		}
+		
+		return builder.build();
+	}
+}

webflux-webclient/src/main/java/com/nisum/urivariables/UriVariablesTest.java

+package com.nisum.urivariables;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+
+import reactor.core.publisher.Mono;
+
+public class UriVariablesTest {
+	
+	public static void main(String[] args) {
+		
+		// http://localhost:8883/order/order-9
+		
+		String uriVariableValue = "order-9-urivariable";
+
+		String endPoint = String.format("order/%s", uriVariableValue);
+		
+		System.out.println(endPoint);
+		
+		Mono<Order> mono = WebClient.create("http://localhost:8883").get().uri(endPoint).retrieve()
+				.bodyToMono(Order.class);
+
+		Order order = mono.block();
+		System.out.println(order);
+	}
+
+}

webflux-webclient/src/main/java/com/nisum/webclient/PersonSub.java

+package com.nisum.webclient;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import com.nisum.domain.Person;
+
+public class PersonSub implements Subscriber<Person> {
+
+	@Override
+	public void onSubscribe(Subscription s) {
+		System.out.println("Subscribed....");
+	}
+
+	@Override
+	public void onNext(Person t) {
+		System.out.println(t);
+	}
+
+	@Override
+	public void onError(Throwable t) {
+		System.out.println("Error----------");
+		System.out.println(t.getMessage());
+		t.printStackTrace();
+	}
+
+	@Override
+	public void onComplete() {
+		System.out.println("Completed.....");
+	}
+ 	
+}

webflux-webclient/src/main/java/com/nisum/webclient/ProcessorTest.java

+package com.nisum.webclient;
+
+public class ProcessorTest {
+	public static void main(String[] args) {
+		
+	}
+}

webflux-webclient/src/main/java/com/nisum/webclient/WebClientStreamTest.java

+package com.nisum.webclient;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+import com.nisum.domain.Person;
+import com.nisum.subscribers.OrderSubscriber;
+
+import reactor.core.publisher.Flux;
+
+public class WebClientStreamTest {
+
+	public static void main(String[] args) throws Exception {
+		
+		String baseUrl = "http://localhost:8081/";
+		String uri = "persons-stream";
+		WebClient webClient = WebClient.create(baseUrl);
+		Flux<Person> persons = webClient.get().uri(uri).header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_NDJSON_VALUE).retrieve().bodyToFlux(Person.class);
+		persons.subscribe(new PersonSub());
+		Thread.sleep(1000*100);
+	}
+		
+}

webflux-webclient/src/main/java/com/nisum/webclient/WebClientTest.java

+package com.nisum.webclient;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.nisum.domain.Order;
+import com.nisum.subscribers.OrderSubscriber;
+
+import reactor.core.publisher.Flux;
+
+public class WebClientTest {
+	public static void main(String[] args) throws InterruptedException {
+			
+		String baseUrl = "https://catalog-ui-service-qa.apps.np.stratus.albertsons.com";
+		String uri = "catalog-ui-service/reports/ecom-products/download3";
+		WebClient webClient = WebClient.create(baseUrl);
+		Flux<Order> orders = webClient.get().uri(uri).retrieve().bodyToFlux(Order.class);
+		orders.subscribe(new OrderSubscriber());
+		
+		Thread.sleep(1000*100);
+	}
+}

webflux-webclient/src/test/java/com/nisum/webflux_webclient/AppTest.java

+package com.nisum.webflux_webclient;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+/**
+ * Unit test for simple App.
+ */
+public class AppTest 
+    extends TestCase
+{
+    /**
+     * Create the test case
+     *
+     * @param testName name of the test case
+     */
+    public AppTest( String testName )
+    {
+        super( testName );
+    }
+
+    /**
+     * @return the suite of tests being tested
+     */
+    public static Test suite()
+    {
+        return new TestSuite( AppTest.class );
+    }
+
+    /**
+     * Rigourous Test :-)
+     */
+    public void testApp()
+    {
+        assertTrue( true );
+    }
+}

webflux-webclient/target/classes/META-INF/MANIFEST.MF

+Manifest-Version: 1.0
+Built-By: sgali
+Build-Jdk: 1.8.0_151
+Created-By: Maven Integration for Eclipse
+

webflux-webclient/target/classes/META-INF/maven/com.nisum/webflux-webclient/pom.properties

+#Generated by Maven Integration for Eclipse
+#Fri Apr 23 22:08:18 IST 2021
+version=0.0.1-SNAPSHOT
+groupId=com.nisum
+m2e.projectName=webflux-webclient
+m2e.projectLocation=C\:\\Users\\sivaji\\Documents\\WebFluxDemo\\WebFlux\\webflux-webclient
+artifactId=webflux-webclient

webflux-webclient/target/classes/META-INF/maven/com.nisum/webflux-webclient/pom.xml

+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.nisum</groupId>
+	<artifactId>webflux-webclient</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>webflux-webclient</name>
+	<url>http://maven.apache.org</url>
+
+	<properties>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<version>2.4.0</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<version>2.1.5.RELEASE</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+	
+		<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+	
+</project>