Portfolio


import java.util.Observable;
import java.util.Observer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ObserverExample {

    static class IntegerObservable extends Observable implements Runnable{

        @Override
        public void run() {
            for (int i = 1; i <= 10; i++) {
                setChanged();
                notifyObservers(i);   //push 역할
                // int i = it.next(); // pull 역할
            }
        }
    }

    @SuppressWarnings("Deprecation")
    public static void main(String[] args) {
        // Source -> (Event/Data Source) -> Observer
        Observer ob = new Observer() {
            @Override
            public void update(Observable o, Object arg) {
                System.out.println(Thread.currentThread().getName()+"  arg = " + arg);
            }
        };

        IntegerObservable io = new IntegerObservable();
        io.addObserver(ob);

//        io.run();

        ExecutorService es = Executors.newSingleThreadExecutor();
        es.execute(io);
        System.out.println(Thread.currentThread().getName()+" EXIT");
        es.shutdown();
    }
}

ObserverExample.java


onSubscribe onNext* (onError | onComplete)?

onSubscribe - 한번 필수 → onNext (0~N회) → (onError | onComplete) 배타적으로 0~1회 호출


package demo.tobyspringtv.episode_5.pubsub;

import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;

public class PubSubMain {
        // Publisher  <- Observable
        // Subscriber <- Observer
    public static void main(String[] args) {

        /* onSubscribe onNext* (onError | onComplete)? */
        Publisher<Integer> p = new CustomPublisher();
        Subscriber<Integer> s = new CustomSubscriber<>();

        p.subscribe(s);
    }
}

PubSub.java


package demo.tobyspringtv.episode_5.pubsub;

import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.Flow;

public class CustomPublisher implements Flow.Publisher<Integer> {

    Iterable<Integer> iter = Arrays.asList(1, 2, 3, 4, 5);
    final Iterator<Integer> it =iter.iterator();

    @Override
    public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
        subscriber.onSubscribe(new Flow.Subscription() {
            @Override
            public void request(long n) {
                try{
                    while(n -- > 0){
                        if(it.hasNext()){
                            subscriber.onNext(it.next());
                        }else{
                            subscriber.onComplete();
                            break;
                        }
                    }
                }catch (RuntimeException e){
                    subscriber.onError(e);
                }
            }

            @Override
            public void cancel() {
            }
        });
    }
}

CustomPublisher.java


package demo.tobyspringtv.episode_5.pubsub;

import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Subscription;

public class CustomSubscriber<T> implements Flow.Subscriber<T> {

    Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        System.out.println("onSubscribe");
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(T item) {
        System.out.println("onNext " + item);
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("onError");
    }

    @Override
    public void onComplete() {
        System.out.println("onComplete");
    }
}

CustomSubscriber.java


onSubscribe
onNext 1
onNext 2
onNext 3
onNext 4
onNext 5
onComplete

Process finished with exit code 0

실행 결과

Reactive Programming