[디자인패턴] 리액터 패턴 (Reactor Pattern)

리액터 패턴

// (서버측) 생산자
interface Publisher<T> {
	void subscribe(Subscriber<T> s);
}

class SimplePublisher implements Publisher<Integer> {
	
	private int max;
	
	public SimplePublisher(int max) {
		this.max = max;
	}
	
	@Override
	public void subscribe(Subscriber<Integer> s) {
		// 1. 구독자 관리
		// 2. 콜백(생산자에게 계약을 돌려줌)
		s.onSubscribe(new SimpleSubscription(s, max));
	}
	
}

// (서버측) 계약 (백프레셔)
interface Subscription {
	void request(int n);	// 양 조절
	void cancel();	// 그만 보내
}

class SimpleSubscription implements Subscription {
	private Subscriber<Integer> sub;
	private int max;
	private int next = 1;

	public SimpleSubscription(Subscriber<Integer> sub, int max) {
		this.sub = sub;
		this.max = max;
	}
	
	@Override
	public void request(int n) {
		if (n > max) {
			sub.onError(new RuntimeException("최대값(" + max + ") 초과"));
		} else {
			int loop = next + n;
			for (; next < loop; next++) {
				sub.onNext(next);
				if (next == max) {
					sub.onComplete();
					break;
				}
			}
		}
	}

	@Override
	public void cancel() {
		System.out.println("[cancel] 데이터 그만 보내");
	}
	
}

// (클라이언트측) 소비자
interface Subscriber<T> {
	void onSubscribe(Subscription s);
	void onNext(T item);
	void onError(Throwable t);
	void onComplete();
	Subscription upstream();
}

class SimpleSubscriber implements Subscriber<Integer> {
	
	private Subscription s;
	
	public Subscription upstream() {
		return s;
	}

	@Override
	public void onSubscribe(Subscription s) {
		this.s = s;
	}

	@Override
	public void onNext(Integer item) {
		System.out.println("onNext : " + item);
	}

	@Override
	public void onError(Throwable t) {
		
		System.out.println("[onError] 에러발생! " + t.getMessage());
	}

	@Override
	public void onComplete() {
		System.out.println("[onComplete] 데이터 수신 완료");
	}
	
}

public class App {
	public static void main(String[] args) {
		Publisher<Integer> pub = new SimplePublisher(10);
		Subscriber<Integer> sub = new SimpleSubscriber();
		pub.subscribe(sub);
		
		Subscription s = sub.upstream();
		
		s.request(3);
		s.request(3);
		s.request(3);
		s.request(3);
		s.request(3);
	}
}