Observable 클래스의 팩토리 함수 (1)

안녕하세요, 오늘은 저번 시간에 처음 배웠던 Observable 클래스를 조금 더 확장하여
이를 구현하는 다양한 종류의 팩토리 함수 중 가장 기본적인 몇 가지를 알아보도록 하겠습니다.

🏖  Observable 클래스의 팩토리 함수란?

Observable 클래스에는 자기 자신을 생성하는 함수, 중간 결과를 처리하는 함수, 디버그 및 예외 처리를 하는 함수 등 수많은 종류의 함수가 존재합니다. 이러한 함수들을 Observable 클래스에서는 팩토리 함수라고 부르며 실질적으로 이를 사용할 때는 직접 인스턴스를 만들지 않고 정적 팩토리 함수를 호출해 기능을 사용할 수도 있습니다. 아래의 표는 RxJava의 대표적인 팩토리 함수들입니다.

 

RxJava 버전	팩토리 함수
버전 1의 기본 팩토리 함수	create( ), just( ), from( )
버전 2의 추가 팩토리 함수	fromArray( ), fromIterable( ), fromCallable( ), fromFuture( )
기타 팩토리 함수	interval( ), range( ), timer( ), defer( )

기존의 RxJava 버전 1에서는 팩토리 함수의 종류가 매우 적었으나 버전 2로 업데이트되면서 Observable 클래스의 팩토리 함수 종류가 상당히 많아진 것을 볼 수 있습니다. 앞으로 저희는 Observable 클래스의 팩토리 함수들에 대해서 천천히 공부해볼 것이며 그중 오늘 배울 팩토리 함수는 Observable 클래스에서 가장 기본적인 just( ), subscribe( ), create( ) 3가지 함수를 알아보겠습니다.

🚘 just( ) 함수란?

just( ) 함수는 Observable 클래스에게 데이터를 주입하기 위해 사용되는 팩토리 함수입니다. just( ) 함수의 인자로 Observable 클래스에게 주입하고 싶은 데이터를 순차적으로 전달하면 됩니다. 이 과정에서 함수의 인자로 전달하는 데이터의 수는 1개에서 10개 사이의 데이터만 전달할 수 있습니다. 또한 정적 팩토리 함수를 사용할 때는 인자로 전달하는 데이터의 타입이 모두 같을 필요는 없으나 직접 인스턴스를 만들어 Observable 클래스를 생성할 때는 인자로 전달하는 데이터의 타입이 모두 같아야 하는 걸 기억하세요!

 

위와 같이 같이 실제 에디터 상에서 Observable 클래스의 just( ) 함수를 정적 팩토리 함수로 호출하여 사용할 때 함수의 인자로 제네릭 형식인 데이터를 1개에서 10개 사이로 전달할 수 있도록 제공하는 다양한 종류의 just( ) 함수가 존재하는 것을 볼 수 있습니다.

 

마블 다이어그램 중앙에 존재하는 원은 Observable 클래스에게 인자로 전달하는 데이터로 추후 just( ) 함수를 거치면 인자로 전달한 데이터가 Observable 클래스에게 주입됩니다. 또한 아래의 파이프 표시는 모든 데이터의 주입이 완료되었다는 것을 의미합니다. just( ) 함수는 앞으로 배울 다른 팩토리 함수들과는 다르게 데이터 내용의 변경되지 않고 인자로 전달했던 형태 그대로 주입됩니다.

 

위의 마블 다이어그램 방식으로 실제 코드를 작성해본다면 아래와 같이 Observable 클래스에 존재하는 just( ) 함수를 호출하여 인자로는 Observable 클래스에 주입할 데이터 1을 전달하였습니다. 그러나 실제 프로그램을 실행해본다면 콘솔에 아무 결과도 나오지 않는 것을 볼 수 있습니다. 그 이유는 바로 Observable 클래스에 데이터 주입만 해주었지 발행은 따로 하지 않았기 때문입니다.

public class FirstExample {
    private void emit() {
         Observable.just(1);
    }

    public static void main(String[] args) {
        FirstExample demo = new FirstExample();
        demo.emit();
    }
}

🚖 subscribe( ) 함수란?

RxJava에서는 내가 동작시키기 원하는 것을 사전에 정의해둔 다음 실제 그것이 실행되는 시점을 임의로 조절할 수 있습니다. 이것을 가능하게 해주는 것이 바로 subscribe( ) 함수입니다. 보통 위에서 배운 Observable 클래스에서 just( ) 함수를 통한 데이터 주입을 마친 후 subscribe( ) 함수를 호출하여 이전에 주입한 데이터를 실질적으로 발행하는 역할로 현재 활발히 사용되고 있는 중입니다.

 

앞에서 보았던 just( ) 함수와 같이 subscribe( ) 함수 또한 실제 에디터 상에서 정적 팩토리 함수로 호출하였을 때 수많은 종류의 subscribe( ) 함수가 존재하는 것을 볼 수 있습니다. 그중 저희는 대표적인 4개의 subscribe( ) 함수만 살펴보도록 하겠습니다.

 

참고사항

1) subscribe 함수의 반환형은 Disposable 인터페이스 객체입니다.

2) onNext 이벤트의 자료형은 Consumer<? super T> 인터페이스 객체입니다.

3) onError 이벤트의 자료형은 Consumer<? super Throwable> 인터페이스 객체입니다.

4) onComplete 이벤트의 자료형은 Action 인터페이스 객체입니다.

1) subscribe( )

인자가 없는 subscribe( ) 함수는 onNext, onComplete 이벤트를 무시하고 onError 이벤트가 발생했을 때만 에러를 던집니다. 보통 Observable 클래스로 작성한 코드를 테스트할 때 자주 활용합니다.

2) subscribe(onNext)

인자가 1개인 오버로딩은 onNext 이벤트를 처리합니다. onError 이벤트 발생 시 에러를 던집니다.

3) subscribe(onNext, onError)

인자가 2개인 오버로딩은 onNext, onError 이벤트를 처리합니다.

4) subscribe (onNext, onError, OnComplete)

인자가 3개인 오버로딩은 onNext, onError, onComplete 이벤트를 모두 처리할 수 있습니다.

subscribe 함수의 반환형은 Disposable 인터페이스 객체이며 이는 RxJava의 구독 객체에 해당합니다. Disposable 인터페이스 객체에는 2가지의 함수가 존재하며 첫 번째 dispose( ) 함수는 Observable 클래스에게 더 이상 데이터를 발행하지 않도록 구독을 해제시키는 함수이며 두 번째 isDisposed( ) 함수는 Observable 클래스가 현재 데이터를 발행하지 않는지 확인하는 함수입니다.

 

Disposable 인터페이스 객체

1) void dispose( )

2) boolean isDisposed( )

subscribe( ) 함수는 데이터 주입이 완료된 Observable 클래스의 데이터 발행을 진행하며 만약 Observable 클래스의 데이터 발행이 모두 끝나 onComplete 이벤트를 보낼 시 자동으로 dispose( ) 함수를 호출해 Observable 클래스와의 구독 관계를 해지합니다.

public class FirstExample {
    private void emit() {
        Observable<String> observable = Observable.just("RED", "GREEN", "BLUE");
        Disposable disposable = observable.subscribe(
                value -> System.out.println("[onNext] value : " + value),
                error -> System.err.println("[onError] err : " + error.getMessage()),
                () -> System.out.println("[onComplete]")
        );
        System.out.println("isDisposed() : " + disposable.isDisposed());
    }

    public static void main(String[] args) {
        FirstExample demo = new FirstExample();
        demo.emit();
    }
}

위의 코드는 Observable 클래스의 알림과 Disposable 인터페이스 객체에 관한 간단한 예제입니다. 예제에서 사용된 subscribe( ) 함수는 인자가 3개인 오버로딩 함수를 사용했으며 각 이벤트 별로 로그를 출력해 상태를 관찰했습니다. 또한 위에서도 언급했듯이 Observable 클래스의 데이터 발행이 모두 끝난 onComplete 이벤트 시점에서 내부 시스템이 자동으로 dispose( ) 함수를 호출해 Observable 클래스와의 구독 관계를 해지하여 isDisposed( ) 함수를 호출했을 시 True의 값이 나오는 것을 확인할 수 있습니다.

 

결론적으로 Observable 클래스는 just( ) 등의 팩토리 함수로 데이터 흐름을 정의한 후 subscribe( ) 함수를 호출해야만 실제로 데이터가 발행된다는 것을 기억해야 합니다. 그만큼 subscribe( ) 함수는 Observable 클래스에서 중요한 역할을 하고 있습니다.

🚔 create( ) 함수란?

앞에서 배웠던 just( ) 함수는 Observable 클래스에 주입할 데이터를 인자로 넣으면 자동으로 해당하는 알림 이벤트가 발생하지만 create( ) 함수와 같은 경우는 onNext, onComplete, onError와 같은 알림을 개발자가 직접 호출해야만 해당 알림이 발생합니다. 즉 관찰대상을 관찰하고 있는 구독자에게 데이터를 발행하기 위해서는 onNext, onComplete 함수를 호출해야 한다는 것입니다.

 

create( ) 함수와 같은 경우는 위에서 보았던 just( ), subscribe( ) 함수와는 달리 오버로딩된 함수의 종류가 1개밖에 존재하지 않으며 인자로 ObservableOnSubscribe 인스턴스 객체에 해당하는 데이터를 넘겨주면 됩니다. 또한 아래는 create( ) 함수의 마블 다이어그램으로 위에서 말했던 것과 같이 개발자가 직접적으로 onNext, onComplete 함수를 호출하고 있는 것을 볼 수 있습니다.

 

위의 마블 다이어그램 방식으로 실제 코드를 작성해보다면 아래와 같이 Observable 클래스에 존재하는 create( ) 함수를 호출하여 인자로는 ObservableEmitter 인터페이스 객체를 람다식 형태로 선언한 후 onNext, onComplete 함수를 통해 데이터를 순차적으로 주입하고 있습니다. 그러나 기존의 create( ) 함수와 같은 경우 인자로 ObservableOnSubscribe 인터페이스 객체를 전달해야 하지만 현재는 ObservableEmitter 인터페이스 객체를 람다 표현식으로 변환해 인자로 전달하고 있는 모습을 볼 수 있습니다.

 

람다 표현식을 활용하면 Observable 클래스의 create( ) 함수를 호출할 때 발생하는 불필요한 익명 객체나 멤버 변수를 기재하지 않고 꼭 필요한 변수만 소스 코드에 작성하면 되므로 소스 코드의 가독성이 높아지기 때문에 람다 표현식을 자주 활용하고 있습니다.

public class FirstExample {
    private void emit() {
        Observable<Integer> observable = Observable.create(
                (ObservableEmitter<Integer> emitter) -> {
                    emitter.onNext(100);
                    emitter.onNext(200);
                    emitter.onNext(300);
                    emitter.onComplete();
                }
        );
        observable.subscribe(System.out::println);
    }

    public static void main(String[] args) {
        FirstExample demo = new FirstExample();
        demo.emit();
    }
}

공식 문서에 따르면 create( ) 함수는 RxJava에 익숙한 사용자만 활용하도록 권고하고 있습니다. 실제 개발 시 create( ) 함수를 사용하지 않고 다른 팩토리 함수를 활용해도 같은 효과를 낼 수 있기 때문입니다. 만약 사용해야 한다면 아래 사항을 꼭 확인하세요!

 

1) Observable 클래스가 구독 해지되었을 때 등록된 콜백을 모두 해제해야 합니다.

2) 관찰자가 관찰대상을 관찰하는 동안에만 onNext, onComplete 이벤트를 호출해야 합니다.

3) 에러가 발생했을 시 오직 onError 이벤트로만 에러를 전달해야 합니다.

4) 배압 현상을 직접 처리해야 합니다.

👍 글을 마치며

저번 시간에 배웠던 Observable 클래스를 확장하여 이를 실질적으로 구현하는 기본 팩토리 함수들에 대해서 알아보았습니다. 각각의 팩토리 함수들은 자신들의 역할이 구분되어 있으며 어떠한 상황에 각 함수를 사용해야 할지 명확이 나뉘어 있었습니다. 다음 시간에는 RxJava 버전 2에서 새롭게 추가된 Observable 클래스의 from 관련 여러 가지의 팩토리 함수들에 대해서 살펴보겠습니다.

 

참고 : RxJava 프로그래밍