[RxSwift] 02.Observables / raw

What is an observable?

*Observable* 은 Rx의 심장과도 같다.

"observable", "observable sequence", "sequence"라고 Rx에서 서로 다른이름으로 불리는 것을 볼 수 있는데, 사실은 모두 같은 것이다. 때로는 다른 Rx 언어를 개발하던 사람이 "stream"이라고 불르는 경우가 있다. 사실 이것 또한 같은 것이다.

RxSwift에서는 단 하나의 단어로 칭하는데, 바로 *Sequence* 이다.

결국 *Observable* 이란 sequence인데, 특별한 능력을 가진 sequence이다.

그 능력은 바로 *asynchronous* 이다.

Observable은 시간의 흐름에 따라서 event를 생산하는데, 이것을 emtting이라고 부른다.

Observable을 개념화 하는 가장 좋은 방법 중에 하나는 marble diagram으로 나타내는 것이다.

왼쪽에서 오른쪽으로 향하는 화살표는 시간의 흐름을 나타내고, 숫자가 쓰여진 구슬들은 element of a sequnce를 나타낸다.

얼마의 시간을 나타내는지 확정할 수 있는 것이 아니고, 그저 lifetime of observable 동안을 나타낸 것이다.

Lifecycle of an observable

observable이 emit하는 것은 *next event* 를 통해서 가능하다.

새로운 diagram을 보면, 맨 오른쪽에 수직선이 하나 그어져있다. 이것은 end of the road for this observable을 나타낸다. 즉 끝을 나타낸다.

이것은 *completed*event 라고 부르며, 그렇게 observable terminate가 발생한다.

그 이후에는 더이상 아무것도 emit할 수 없다.

때로는 뭔가 잘못 될 수도 있다.

빨간색으로 된 X가 오류를 뜻하는 marble이다. Observable은 error를 갖고있는 *error event* 를 emit하기도 한다. 이 때, *completed event* 와 마찬가지로 observable terminate가 발생한다. 그래서 마찬가지로 그 이후에는 더 이상 아무것도 emit할 수 없다.

코드로 보면 아래와 같다.

/// Represents a sequence event.
///
/// Sequence grammar:
/// **next\* (error | completed)**
public enum Event<Element> {
    /// Next element is produced.
    case next(Element)

    /// Sequence terminated with an error.
    case error(Swift.Error)

    /// Sequence completed successfully.
    case completed
}

Creating observables

    let observable = Observable<Int>.just(one)
  // Observable<Int>(one)

  let observable2 = Observable.of(one, two, three)
  // Observable<Int>(one), Observable<Int>(two), Observable<Int>(three)  

  let observable3 = Observable.of([one, two, three])
  // Observable<[Int}>([one, two, three])

  let observable4 = Observable.from([one, two, three])
    // Observable<Int>(one), Observable<Int>(two), Observable<Int>(three)

• .just() : 단 하나의 element만을 갖는 sequence를 생성한다.
• .of() : 포함하는 parameter의 개수만큼의 sequence를 생성한다.
• .from() : array만을 parameter로 받으며, array의 요소를 각각의 sequence로 만든다.

Subscribing to observables

observable은 sequence definition이고,

observable을 subscribing 하는 것은 Iterator 에서 next() 함수를 실행하는 것과 매우 비슷하다.

let sequence = 0..<3

var iterator = sequence.makeIterator()

while let n = iterator.next() {
  print(n)
}

/* Prints:
 0
 1
 2
 */

Subscribing to observables is more streamlined.

각각의 event type에 대해서 handler을 추가할 수 있다. (next, error, completed)

example(of: "subscribe") {
  let one = 1
  let two = 2
  let three = 3

  let observable = Observable.of(one, two, three)

  observable.subscribe { event in
    print(event)
  }
}

위와 같이 observable을 subscribe할 수 있고, subscribe의 return값은 Disposable이다.

위 코드의 결과를 보면 아래와 같다.

--- Example of: subscribe ---
next(1)
next(2)
next(3)
completed

3개의 Observable에 대해서 next event가 emit되었고, completed event 가 emit된 후 terminate되었다.

element에 직접 접근하고 싶다면, 아래와 같은 코드로 작성하면 된다.

observable.subscribe { event in
  if let element = event.element {
    print(element)
  }
}

앞에서 공부했듯이, next event 만이 element를 갖고있기 때문에 next event 의 emit만이 콘솔로 출력된다.

1
2
3

위와 같이 작성해서 element를 얻어내는 방법 말고도 더 멋진 방법이 있다.

observable.subscribe(onNext: { element in
  print(element)
})

이렇게 코드를 작성하면, 명시적으로 next event 에 대해서만 실행되는 closure을 작성할 수 있다.

empty

지금까지 봐왔던 observable은 하나 또는 여러개의 element가 있었다.

다른 객체들과 마찬가지로 observable또한 zero element의 객체를 만들 수 있다.

example(of: "empty") {
  let observable = Observable<Void>.empty()
}

하지만 element가 없다면, 자료형이 추론될 수 없기 때문에 선언할 때 직접 명시해주어야 한다.

observable.subscribe(
  // 1
  onNext: { element in
    print(element)
  },

  // 2
  onCompleted: {
    print("Completed")
  }
)

*empty* observable은 element가 없기 때문에 next event emit이 발생하지 않는다.

--- Example of: empty ---
Completed

따라서 위와 같이 즉시 completed event emit이 발생한다.

never

*empty* operator과 마찬가지로 *never* operator은 아무 것도 갖지 않은 observable을 생성한다.

아무것도 emit하지 않는것은 같지만, never은 terminate되지도 않는다는 차이점이 있다.

즉, infinite duration동안 존재할 수 있다.

example(of: "never") {
  let observable = Observable<Void>.never()

  observable.subscribe(
    onNext: { element in
      print(element)
    },
    onCompleted: {
      print("Completed")
    }
  )
}

위 코드의 결과로는 아무것도 나타나지 않는다.

--- Example of: never ---

range

지금까지는 특정한 element나 값을 observable에 할당 해 주었지만, 범위를 통해서도 생성할 수 있다.

example(of: "range") {
  // 1
  let observable = Observable<Int>.range(start: 1, count: 10)

  observable
    .subscribe(onNext: { i in
      // 2
      let n = Double(i)

      let fibonacci = Int(
        ((pow(1.61803, n) - pow(0.61803, n)) /
          2.23606).rounded()
      )

      print(fibonacci)
  })
}

위 코드는 아래와 같이 해석된다.

1. start 의 값을 가지는 Observable부터, count 만큼의 Observable이 생성된다.
2. 각각의 emitted element에 대한 nth Fibonacci가 계산되어 출력된다.

*never* 의 경우를 제외하고는, 지금까지 봐온 observable은 자연적으로 completed event emit이 발생한다는 것을 기억하자.

Disposing and terminating

observable은 subscription이 발생하기 전 까지는 아무것도 하지 않는다는 것을 알아두자

subscription은 *observables work에 대한 트리거이다. 그렇게 시작돼서error,completed` event가 발생하기 전 까지 새로운 *event emit이 발생된다.

하지만, cancelling a subscription을 통해서 직접 observable의 terminate를 발생시킬 수 있다.

아래는 그 예시 코드이다.

example(of: "dispose") {
  // 1
  let observable = Observable.of("A", "B", "C")

  // 2
  let subscription = observable.subscribe { event in
    // 3
    print(event)
  }
}

명시적으로 cancel a subscription 하기 위해서는, dispose() 를 호출하면 된다. cancel the subscription 하거나 dispose하면, observable은 더이상 emitting event를 하지 않을것이다.

각각의 subscription을 관리하는 일은 굉장히 힘들다면, DisposeBag 을 사용할 수 있다. dispose bag은 disposable을 보유하고 있다가, deallocated될 때 각각의 disposable에 dispose()를 호출한다.

example(of: "DisposeBag") {
  // 1
  let disposeBag = DisposeBag()

  // 2
  Observable.of("A", "B", "C")
    .subscribe { // 3
      print($0)
    }
    .disposed(by: disposeBag) // 4
}

1. dispose bag을 만든다.
2. observable을 만든다.
3. 각각의 observable을 subscribe한 뒤, emitted event를 출력한다.
4. dispose bag에 담는다.

위 코드는 가장 많이 사용되는 패턴이다. observable을 만들고, subscribing하고, 즉시 subscription을 dispose bag에 담는 패턴이다.

이렇게 귀찮을정도로 dispose에 신경쓰는 이유는 아래와 같다.

만약 dispose bag에 추가하지 않거나, dispose를 하지 않은 채로 terminate했을 때, 메모리 누수가 발생할 것이다.

create

create operator은 observable이 subscriber에게 emit할 모든 event를 명시할 수 있는 또다른 방법이다.

subscribe 매개변수는 AnyObserver 을 받아서 Disposable 을 리턴하는 escaping closure이다. AnyObserver 이란 observable sequence에 값을 추가하는 것을 쉽게 도와주는 제네릭 타입인데, emit to subscribers된다.

Observable<String>.create { observer in
  // 1
  observer.onNext("1")

  // 2
  observer.onCompleted()

  // 3
  observer.onNext("?")

  // 4
  return Disposables.create()
}

1. 2. 3. 각각 event를 추가한다.
   3. subscribe operator는 반드시 Disposable 을 리턴해야하기 때문에, Disposables.create() 를 통해서 disposable을 생성해 리턴한다.

위 명령이 잘 되는지 확인하기 위해서, 아래 코드를 활용하면 된다.

.subscribe(
  onNext: { print($0) },
  onError: { print($0) },
  onCompleted: { print("Completed") },
  onDisposed: { print("Disposed") }
)
.disposed(by: disposeBag)

결과는 아래와 같다.

--- Example of: create ---
1
Completed
Disposed

아래 명령어를 이용해서 Error을 정의할 수 있다.

enum MyError: Error {
  case anError
}

그런 뒤, observer에 error을 발생시키는 코드를 추가한다.

observer.onError(MyError.anError)

결과는 아래와 같다.

--- Example of: create ---
1
anError
Disposed

예상대로, Error가 발생한 뒤에는 observable이 terminate된다.

위 코드에서 disposed(by: disposeBag) 을 없앤다면, observable이 영원히 끝나지 않는것을 볼 수 있다.

example(of: "create") {
  enum MyError: Error {
    case anError
  }

  let disposeBag = DisposeBag()

  Observable<String>.create { observer in
    // 1
    observer.onNext("1")

//    observer.onError(MyError.anError)

    // 2
//    observer.onCompleted()

    // 3
    observer.onNext("?")

    // 4
    return Disposables.create()
  }
  .subscribe(
    onNext: { print($0) },
    onError: { print($0) },
    onCompleted: { print("Completed") },
    onDisposed: { print("Disposed") }
  )
//  .disposed(by: disposeBag)
}

그 결과는 아래와 같이, 영원히 끝나지도 사라지지도 않는다.

--- Example of: create ---
1
?

Creating observable factories

subscriber을 기다리는(?) observable을 만드는 것 말고, 각각의 subscriber에게 새로운 observable을 제공해주는 observable factories를 만드는 것이 가능하다.

example(of: "deferred") {
  let disposeBag = DisposeBag()

  // 1
  var flip = false

  // 2
  let factory: Observable<Int> = Observable.deferred {

    // 3
    flip.toggle()

    // 4
    if flip {
      return Observable.of(1, 2, 3)
    } else {
      return Observable.of(4, 5, 6)
    }
  }

  // 5
    for _ in 0...3 {
      factory.subscribe(onNext: {
        print($0, terminator: "")
      })
      .disposed(by: disposeBag)

      print()
    }
}

이제, factory를 subscribe 할 때 마다, 반대의 observable을 얻을 수 있다. 즉, 123, 456 을 얻는 패턴이 반복 될 것이다.

--- Example of: deferred ---
123
456
123
456

Using Traits

Traits란 일반적인 observable보다 좀 더 좁은 행동범위를 갖고 있는 observable이다. 사용은 선택적이고, trait을 사용할 수 있는 곳에서는 모두 일반적인 observable을 사용할 수 있다. 이 trait의 목적은 사용자에게 자신의 목적을 떠 명확하게 전달함에 있다.

Single

• emit success(value) or error(error) event
• success 후 value를 전달하거나 fail을 하는 단 한 번 동작하는 작업을 할 때 유용하다.
  • 예) data를 다운로드 할 때, disk에서 로딩할 때

Completable

• emit completed or error(error) event
• 어떤 value도 emit하지 않아서, 작업이 성공적으로 완료됐는지, 실패했는지를 알고싶을 때 유용하다.
  • 예) file write

Maybe

• emit succeess(value), or completed , error(error)

Single 예시

example(of: "Single") {
  // 1
  let disposeBag = DisposeBag()

  // 2
  enum FileReadError: Error {
    case fileNotFound, unreadable, encodingFailed
  }

  // 3
  func loadText(from name: String) -> Single<String> {
    // 4
    return Single.create { single in

    }
  }
}

1. dispose bag 을 추가한다.
2. Error을 정의한다.
3. Single 을 리턴하는 loadText함수를 정의한다.
4. Single 을 생성하고 리턴한다.

// 1
let disposable = Disposables.create()

// 2
guard let path = Bundle.main.path(forResource: name, ofType: "txt") else {
  single(.error(FileReadError.fileNotFound))
  return disposable
}

// 3
guard let data = FileManager.default.contents(atPath: path) else {
  single(.error(FileReadError.unreadable))
  return disposable
}

// 4
guard let contents = String(data: data, encoding: .utf8) else {
  single(.error(FileReadError.encodingFailed))
  return disposable
}

// 5
single(.success(contents))
return disposable

1. Disposable 을 생성한다. 왜냐면, create 의 subscribe 클로져는 그걸 리턴해야하기 때문에
2. 3. 4. 각각의 단계를 수행하면서, error가 발생하면 Single 객체를 리턴한다.
   4. 여기까지 왔다면, 모든 과정이 성공적으로 수행된것이기 때문에 Single 에 success를 넣고 리턴한다.

Challenges

Do

never operator은 completed를 출력하지 않는다.

따라서 dispose되기 전까지는 제대로 subscribe되고있는지 확인할 수 없다.

on operator을 사용하면 더 다양한 케이스에 대해서 행동을 정의할 수 있다.

.do(
        onNext: ,
        afterNext: ,
        onError: ,
        afterError: ,
        onCompleted: ,
        afterCompleted: ,
        onSubscribe: ,
        onSubscribed: ,
        onDispose:
    )

on operator은 emiited event를 변경하지 않고 그대로 chain에 전달한다.

Debug

debug operator을 통해서 event를 debug할 수 있다.

출력 형식은 "<Date> <Time>: <identifier> -> <event>"