Subscribe 를 한 번만 할 수 있도록 하는 Operator 입니다.

LimitSubscribe.swift

import RxSwift
import class Foundation.NSRecursiveLock

extension ObservableType {
  func limitSubscribe() -> Observable<Element> {
    return LimitSubscribe<Element>(source: self.asObservable()).asObservable()
  }
}

final class LimitSubscribe<Element>: ObservableType {

  private let lock = NSRecursiveLock()
  private var isSubscribed: Bool = false
  private let source: Observable<Element>

  init(source: Observable<Element>) {
    self.source = source
  }

  func subscribe<Observer>(_ observer: Observer) -> Disposable where Observer : ObserverType, LimittedSubscribeObservable.Element == Observer.Element {
    self.lock.lock(); defer { self.lock.unlock() }
    
    if self.isSubscribed {
      #if DEBUG
      fatalError("This Observable can subscribe only once!")
      #else
      observer.onCompleted()
      #endif
      return Disposables.create()
    } else {
      self.isSubscribed = true
      return self.source.subscribe(observer)
    }
  }
}

Playground.swift

let stream = Observable.of(["A", "B"])
  .limitSubscribe()

stream.subscribe()
stream.subscribe() // crash!