Creating your own Observable (aka observable sequence) && Creating an Observable that performs work

RxSwift/Documentation/GettingStarted.md at main · ReactiveX/RxSwift

• 해당 내용은 RxSwift의 공식 깃허브의 내용을 바탕으로 작성한 글입니다.
• 번역이나 이해가 부족하여 틀린 내용이 있을 수 있습니다.

Creating your own Observable

• observables를 이해하는데 가장 중요한 것은 단순히 생성만으로는 어떠한 수행도 하지 않는다는 것이다.
• Observable은 요소들을 생성할 수 있는 방법이지만, 이들 중 일부분은 부작용을 일으키는것뿐만 아니라, 현재 존재하여 실행되는 프로세스한테도 영향을 끼친다.

func searchWikipedia(searchTerm: String) -> Observable<Results> {}

let searchForMe = searchWikipedia("me")

// no requests are performed, no work is being done, no URL requests were fired

let cancel = searchForMe
  // sequence generation starts now, URL requests are fired
  .subscribe(onNext: { results in
      print(results)
  })

• subscribe메서드를 실행할 경우 시퀀스 생성이 시작된다.

func myJust<E>(_ element: E) -> Observable<E> {
    return Observable.create { observer in
        observer.on(.next(element))
        observer.on(.completed)
        return Disposables.create()
    }
}

myJust(0)
    .subscribe(onNext: { n in
      print(n)
    })

• Observable시퀀스를 생성하는 많은 방법중 가장 쉬운 방법은 create메서드를 사용하는 것이다.
• RxSwift는 subscribe시 하나의 요소만을 반환하는 방법을 제공하는데 이것이 위 코드의 Just이다.

예상 출력값

0

• 위의 코드들이 클로저와 subscribe 메서드를 사용하여 구현하는 가장 쉬운 방법이며, 하나의 인자, 하나의 값을 넘기고 disposable을 반환한다.

• create메서드는 어떨까?

• Sequence는 동기로 구현된다. 그러므로 subscribe메서드를 호출하기 전 요소들을 생성하고 종료된다.

• 그리고 동기 시퀀스를 반환할 경우 NopDisposable이 반환된다.

• 간단한 배열 Observable을 만들어보자.

func myFrom<E>(_ sequence: [E]) -> Observable<E> {
    return Observable.create { observer in
        for element in sequence {
            observer.on(.next(element))
        }

        observer.on(.completed)
        return Disposables.create()
    }
}

let stringCounter = myFrom(["first", "second"])

print("Started ----")

// first time
stringCounter
    .subscribe(onNext: { n in
        print(n)
    })

print("----")

// again
stringCounter
    .subscribe(onNext: { n in
        print(n)
    })

print("Ended ----")

• 출력값

Started ----
first
second
----
first
second
Ended ----

Creating an Observable that performs work

• 위 예제를 토대로 interval연산자를 사용하여 예제를 업그레이드 시켜보

func myInterval(_ interval: DispatchTimeInterval) -> Observable<Int> {
    return Observable.create { observer in
        print("Subscribed")
        let timer = DispatchSource.makeTimerSource(queue: DispatchQueue.global())
        timer.schedule(deadline: DispatchTime.now() + interval, repeating: interval)

        let cancel = Disposables.create {
            print("Disposed")
            timer.cancel()
        }

        var next = 0
        timer.setEventHandler {
            if cancel.isDisposed {
                return
            }
            observer.on(.next(next))
            next += 1
        }
        timer.resume()

        return cancel
    }
}

let counter = myInterval(.milliseconds(100))

print("Started ----")

let subscription = counter
    .subscribe(onNext: { n in
        print(n)
    })


Thread.sleep(forTimeInterval: 0.5)

subscription.dispose()

print("Ended ----")

출력값

Started ----
Subscribed
0
1
2
3
4
Disposed
Ended ----

• 이전과 같이 동기로 작동하는 코드에서 타이머를 global()쓰레드에서 실행시키고 비동기적으로 작동하도록 코드가 구성되었다.
• 이렇게 코드를 구성할 경우 과연 비동기적으로 코드가 작동할지 확인해보자

let counter = myInterval(.milliseconds(100))

print("Started ----")

let subscription1 = counter
    .subscribe(onNext: { n in
        print("First \(n)")
    })

let subscription2 = counter
    .subscribe(onNext: { n in
        print("Second \(n)")
    })

Thread.sleep(forTimeInterval: 0.5)

subscription1.dispose()

print("Disposed")

Thread.sleep(forTimeInterval: 0.5)

subscription2.dispose()

print("Disposed")

print("Ended ----")

출력값

Started ----
Subscribed
Subscribed
First 0
Second 0
First 1
Second 1
First 2
Second 2
First 3
Second 3
First 4
Second 4
Disposed
Second 5
Second 6
Second 7
Second 8
Second 9
Disposed
Ended ----

• 위 예제들처럼 subscribe이 될 경우 별도의 시퀀스를 따로따로 생성되며, 비동기로 작동시킬 수 있다.