Programming Tutorials

concurrent.Flow instead of Observable class in Java

By: Norman Chap in Java Tutorials on 2023-05-01  

The Observer and Observable classes were deprecated in Java 9, and it is recommended to use the java.util.concurrent.Flow classes instead. Here is an example of using the Flow API to achieve the same functionality:

import java.util.concurrent.Flow.*;
import java.util.List;
import java.util.ArrayList;

// This is the observing class.
class Watcher implements Subscriber<Integer> {
    private Subscription subscription;

    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        if (subscription != null) {
            subscription.request(1);
        }
    }

    public void onNext(Integer arg) {
        System.out.println("update() called, count is " + arg);
        if (subscription == null) {
            return;
        }
        subscription.request(1);
    }
    

    public void onComplete() {}

    public void onError(Throwable throwable) {}
}

// This is the class being observed.
class BeingWatched implements Publisher<Integer> {
    private final List<Subscriber<? super Integer>> subscribers = new ArrayList<>();

    public void subscribe(Subscriber<? super Integer> subscriber) {
        subscribers.add(subscriber);
    }

    void counter(int period) {
        for (; period >= 0; period--) {
            for (Subscriber<? super Integer> subscriber : subscribers) {
                subscriber.onNext(period);
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                System.out.println("Sleep interrupted");
            }
        }
        for (Subscriber<? super Integer> subscriber : subscribers) {
            subscriber.onComplete();
        }
    }
}

class ObserverDemo {
    public static void main(String args[]) {
        BeingWatched observed = new BeingWatched();
        Watcher observing = new Watcher();
        /*
         * Add the observing to the list of observers for
         * observed object.
         */
        observed.subscribe(observing);
        observed.counter(10);
    }
}

The output will be like this:

update() called, count is 10
update() called, count is 9
update() called, count is 8
update() called, count is 7
update() called, count is 6
update() called, count is 5
update() called, count is 4
update() called, count is 3
update() called, count is 2
update() called, count is 1
update() called, count is 0

This implementation uses the Publisher interface to represent the observed object, and the Subscriber interface to represent the observing object. The subscribe() method is used to add a Subscriber to the list of subscribers for the Publisher, and the onNext() method is called by the Publisher to notify the Subscriber of a new value. The onComplete() method is called by the Publisher when there are no more values to be emitted.