Here we have two overload methods delayedExecutor(), the first method has the following syntax:

static Executor delayedExecutor(long delay, TimeUnit unit)

This method returns an Executor object from the default Executor object that the CompletableFuture object uses to execute the task, after the delay. And this new Executor object will do task execution.

The second is:

static Executor delayedExecutor(long delay, TimeUnit unit, Executor executor)

This method also returns an Executor object but is an Executor object that we pass into this method, after the delay. And this new Executor object will also do task execution.

For example, I want to calculate the sum of two numbers and for some conditions, I want it to happen after 2 seconds, my code will look like this:

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) throws InterruptedException {
        int a = 2;
        int b = 5;

        CompletableFuture.supplyAsync(() -> a + b, CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS))
            .thenAccept(result -> System.out.println(result));

        TimeUnit.SECONDS.sleep(10);
    }
}

When running the above example, you will see after 2s, the results will appear:

The post delayedExecutor() method of CompletableFuture object in Java appeared first on Huong Dan Java.

CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit)

This method is used to: if after a timeout period our task is still unfinished, instead of throwing out the TimeoutException exception like the orTimeout() method, our code will return the value that we passed in this method.

The following example I want to calculate the sum of two numbers and I want it to complete in 1s, but because in the process of processing, I give sleep 3s:

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) throws InterruptedException {
        int a = 2;
        int b = 5;

        CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return a + b;
        })
        .completeOnTimeout(0, 1, TimeUnit.SECONDS)
        .thenAccept(result -> System.out.println(result));

        TimeUnit.SECONDS.sleep(10);
    }
}

The result will return the value you passed into the completeOnTimeout() method, which is 0:

If I now increase the timeout time to 4s:

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) throws InterruptedException {
        int a = 2;
        int b = 5;

        CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return a + b;
        })
        .completeOnTimeout(0, 4, TimeUnit.SECONDS)
        .thenAccept(result -> System.out.println(result));

        TimeUnit.SECONDS.sleep(10);
    }
}

The result would be:

The post completeOnTimeout() method of CompletableFuture in Java appeared first on Huong Dan Java.

CompletableFuture<T> orTimeout(long timeout, TimeUnit unit)

This method is used to specify that if our task does not complete within a certain period of time, the program will stop and be TimeoutException.

In the following example, I want to calculate the sum of two numbers that will be completed in 1 second, but in the process of summing, I set up it sleep 3s.

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) throws InterruptedException {
        int a = 2;
        int b = 5;

        CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return a + b;
        })
        .orTimeout(1, TimeUnit.SECONDS)
        .whenComplete((result, exception) -> {
            System.out.println(result);
            if (exception != null)
                exception.printStackTrace();
        });

        TimeUnit.SECONDS.sleep(10);
    }
}

Result:

If I now increase the timeout time:

package com.huongdanjava.javaexample;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) throws InterruptedException {
        int a = 2;
        int b = 5;

        CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return a + b;
        })
        .orTimeout(4, TimeUnit.SECONDS)
        .whenComplete((result, exception) -> {
            System.out.println(result);
            if (exception != null)
                exception.printStackTrace();
        });

        TimeUnit.SECONDS.sleep(10);
    }
}

the program will not fail anymore.

The post orTimeout() method of CompletableFuture object in Java appeared first on Huong Dan Java.

This method is similar to the ifPresent() method which I introduced in this tutorial, but here is a difference since we have one more Runnable parameter, so that in-case if the Optional object is empty, the object of the Runnable interface will be executed.

For example:

package com.huongdanjava.javaexample;

import java.util.Optional;

public class Example {

	public static void main(String[] args) {
		String s = null;

		Optional<String> opt = Optional.ofNullable(s);
		opt.ifPresentOrElse(
			x -> System.out.println(x),
			() -> System.out.println("No value"));
	}

}

Result:

The post ifPresentOrElse() method of Optional object in Java appeared first on Huong Dan Java.

This method is used to create a new Stream object from an Optional object in Java.

If the Optional object contains a value, this method will return the Stream object containing that value, otherwise, it returns an empty Stream object.

For example:

package com.huongdanjava.javaexample;

import java.io.IOException;
import java.util.Optional;
import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) throws IOException {
		String s = "Khanh";

		Optional<String> opt = Optional.ofNullable(s);
		Stream<String> stream = opt.stream();
		stream.forEach(System.out::println);
	}
}

Result:

In the case where we have a List of Optional objects, one of them is empty, the stream() method of the Optional object can help us remove these Optional empty objects.

For example:

package com.huongdanjava.javaexample;

import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) throws IOException {
		List<Optional<String>> names = List.of(
			Optional.of("Khanh"), 
			Optional.empty(), 
			Optional.of("Quan"));

		Stream<String> stream = names.stream().flatMap(Optional::stream);
		stream.forEach(System.out::println);
	}
}

You can refer to the flatMap() method of the Stream object here.

Result:

The post stream() method of Optional object in Java appeared first on Huong Dan Java.

Content:

public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f)

Example:

package com.huongdanjava.javastream;

import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) {
		Stream.iterate(1, i -> i + 1).forEach(System.out::println);
	}
}

When running the above example, you will see that a Stream object contains endless numbers beginning at 1.

With Java 9, the iterate() method has improved by adding another parameter, the Predicate interface, which allows us to stop these endless numbers based on the conditions we define with the Predicate interface.

Content:

public static<T> Stream<T> iterate(T seed, Predicate<? super T> hasNext, UnaryOperator<T> next)

Example:

package com.huongdanjava.javastream;

import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) {
		Stream.iterate(1, i -> i < 10, i -> i + 1).forEach(System.out::println);
	}
}

Result:

The post Method iterate() of Stream object in Java appeared first on Huong Dan Java.

static<T> Stream<T> ofNullable(T t)

This method will return the Stream object of an element in case this element is not null; if it is null, it returns an empty Stream.

Example:

package com.huongdanjava.javaexample;

import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) {
		String s = null;
		Stream.ofNullable(s).forEach(System.out::println);
	}
}

When you run the above code, it will print nothing in the console. But if you edit the variable s, not null:

package com.huongdanjava.javaexample;

import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) {
		String s = "Khanh";
		Stream.ofNullable(s).forEach(System.out::println);
	}
}

then the result will be:

The post Method ofNullable() of Stream object in Java appeared first on Huong Dan Java.

default Stream<T> dropWhile(Predicate<? super T> predicate)

This method also has the parameter Predicate interface and its function is the opposite of the takeWhile() method. This method also passes each element in your Stream object from left to right and ignores all elements that satisfy the condition. Once the condition is no longer fulfilled, it will take all the remaining elements to return.

Example:

package com.huongdanjava.javastream;

import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) {
		Stream.of(3,2,4,1,4,6,7,8,9,10)
			.dropWhile(i -> i < 5)
			.forEach(System.out::println);
	}
}

In the above example, the dropWhile() method will ignore all elements smaller than 5 and when an element larger than 5, it will return all the remaining elements including this element.

Result:

The post Method dropWhile() of Stream object in Java appeared first on Huong Dan Java.

default Stream<T> takeWhile(Predicate<? super T> predicate)

The parameter of this method is the Predicate interface and this method takes the elements in your Stream object from left to right, until the condition of the Predicate object is no longer fulfilled.

Example:

package com.huongdanjava.javastream;

import java.util.stream.Stream;

public class Example {

	public static void main(String[] args) {
		Stream.of(1,2,3,4,5,6,7,8,9,10)
			.takeWhile(i -> i < 5)
			.forEach(System.out::println);
	}
}

In the above example, the condition for getting the elements from left to right is that they must be less than 5. When the element is greater than 5, the method will stop and returns the result.

Result:

The post Method takeWhile() of Stream object in Java appeared first on Huong Dan Java.

The main idea of ​​Reactive Streams is:

• We have a Publisher to broadcast the information.
• We have one or more Subscribers to receive the information that Publisher publishes.
• And a Subscription bridges the gap between Publisher and Subscribers.

Publisher connects with Subscribers in principle:

• The Publisher will add all Subscribers that it needs to notify.
• Subscribers will receive all the information added to the Publisher.
• Subscribers will request and process one or more information from the Publisher asynchronously through the Subscription object.
• When a Publisher has information to publish, this information is sent to all Subscribers who are requesting it.

From version 9, Java supports us to create Reactive Streams applications by introducing three interfaces Flow.Publisher, Flow.Subscriber, Flow.Subscription, and a class named SubmissionPublisher that implements the Flow.Publisher interface. Each interface will play a different role, corresponding to the principles of Reactive Streams.

To better understand Reactive Streams in Java, I will make an example using these interfaces:

First, I will create an object that contains the information that the Publisher needs inform to Subscribers.

This object is called Student with simple content as follows:

package com.huongdanjava.reactivestreams;

public class Student {

  private String name;

  public String getName() {
    return name;
  }

  public void setName(String name) {
    this.name = name;
  }

  @Override
  public String toString() {
    return "Student{" +
        "name='" + name + '\'' +
        '}';
  }
}

The next object would be a class implement interface Subscriber.

Interface Subscriber has four methods:

• The onComplete() method: This method is called when the Publisher object completes its role.
• The onError() method: This method is called when something goes wrong in the Publisher and is notified to the Subscriber.
• The onNext() method: This method is called whenever the Publisher has new information to be notified to all Subscribers.
• The onSubscribe() method: This method is called when Publisher adds Subscriber.

The content of this object is as follows:

package com.huongdanjava.reactivestreams;

import java.util.concurrent.Flow;

public class Consumer implements Flow.Subscriber {

  public void onSubscribe(Flow.Subscription subscription) {
    System.out.println("Consumer: onSubscribe");
  }

  public void onNext(Object item) {
    System.out.println("Consumer: onNext" + item);
  }

  public void onError(Throwable throwable) {
    System.out.println("Consumer: onError");
  }

  public void onComplete() {
    System.out.println("Consumer: onComplete");
  }
}

Now, I will create the main() Example class to run the example.

This Example class will create a new Publisher using the SubmissionPublisher class, subscribe to the Consumer object, and submit 10 Student objects with a time interval of 1s.

Interface Publisher has a method as follows:

• The subscribe() method: This method takes the parameter as a Subscriber object. The Publisher will put this Subscriber object in the list of objects it needs to notify whenever new information is added.

The content of the Example class is as follows:

package com.huongdanjava.reactivestreams;

import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) {
        Consumer c = new Consumer();

        SubmissionPublisher<Student> sp = new SubmissionPublisher<>();
        sp.subscribe(c);

        for (int i = 0; i < 10; i++) {
            Student student = new Student();
            student.setName("Student " + i);

            sp.submit(student);

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        sp.close();
    }
}

At this time, when running the application you only see the results as follows;

This is because the Publisher only calls the Subscriber’s onSubscribe() method to add it to the list of objects to be notified. Subscriber not yet declared to request information from Publisher.

Now we will use the Subscription object in the onSubscribe() method parameter in the Subscriber object to request the Publisher to retrieve the information.

Interface Subscription has a method as follows:

• The request() method: This method is used by the Subscriber object to request information from the Publisher.

We will revise the onSubscribe() method of the Consumer object as follows:

package com.huongdanjava.reactivestreams;

import java.util.concurrent.Flow;

public class Consumer implements Flow.Subscriber {

  public void onSubscribe(Flow.Subscription subscription) {
    System.out.println("Consumer: onSubscribe");
    subscription.request(2);
  }

  public void onNext(Object item) {
    System.out.println("Consumer: onNext" + item);
  }

  public void onError(Throwable throwable) {
    System.out.println("Consumer: onError");
  }

  public void onComplete() {
    System.out.println("Consumer: onComplete");
  }
}

Here, I only request 2 Student objects added to Publisher.

Result:

To request all Student objects each time they are added to Publisher, you can modify the onNext () method using the Subscription object to request.

For example:

package com.huongdanjava.reactivestreams;

import java.util.concurrent.Flow;

public class Consumer implements Flow.Subscriber {

  private Flow.Subscription subscription;

  public void onSubscribe(Flow.Subscription subscription) {
    System.out.println("Consumer: onSubscribe");
    this.subscription = subscription;
    subscription.request(2);
  }

  public void onNext(Object item) {
    System.out.println("Consumer: onNext" + item);
    subscription.request(1);
  }

  public void onError(Throwable throwable) {
    System.out.println("Consumer: onError");
  }

  public void onComplete() {
    System.out.println("Consumer: onComplete");
  }
}

Then, you will see when the Publisher adds a new Student, the Subscriber can get that Student immediately:

As you can see, when the Publisher completes its work, the onComplete() method in the Subscriber object will be called.

Also, in the Subscription object, we also have a cancel() method so that Subscribers can discard the message from the Publisher if they want.

The post Reactive Streams in Java appeared first on Huong Dan Java.