Getting Started

（RxSwift）的用法会尽量与ReactiveX.io官方文档保持一致。其他常见的跨平台文档和教程对RxSwift也应当有效。

1. Observables aka Sequences
2. Disposing
3. 创建你自己的 Observable
4. 创建一个执行工作的Observable
5. 分享订阅和shareReplay操作符
6. Operators（操作符）
7. Playgrounds
8. 错误处理
9. 调试
10. Variables
11. KVO
12. UI layer tips
13. Making HTTP requests
14. RxDataSources

Observables aka Sequences

Basics

你可以同观察者模式(Observable<Element> sequence) 和普通序列(Sequence)的对比来理解Rx，这也是理解Rx最重要的事情。

每一个Observable 序列都是一个序列，Observable对比Swift的Sequence最大的优势是它可以异步的接受元素。这是RxSwift的核心。这里的文档是我们展开来介绍这个理念的。

1. Observable(ObservableType)等价于 Sequence（序列）。
2. ObservableType.subscribe 方法等价于 Sequence.makeIterator method.
3. Observable回掉要被传递到ObservableType.subscribe来接受元素，类似于next() 返回的 iterator。

序列是一个简单、熟悉并且容易理解概念。
人类是一个拥有强大视觉皮层的动物，当我们要理解一个概念时，可视化概念能帮我们更加容易的理解。
通过尝试模拟每一个Rx操作符内部的事件状态机，我们能够更容易理解Rx。
在一个异步模型的系统中，如果我们不用Rx，很可能意味着我们的代码充满了状态机和暂时的状态需要我们模拟。
list和序列可能是我们学习数学和程序最先了解的概念。
这是一个数字序列

--1--2--3--4--5--6--| // terminates normally

另一个字母序列

--a--b--a--a--a---d---X // terminates with error

一些序列序列时有限的一些是无限的，比如按钮的tap序列。

---tap-tap-------tap--->

上述表示法方法叫做marble diagrams。这里可以看到更多的marble diagrams rxmarbles.com.
如果我们指定序列语法作为一个有规律的表达，可以写成这样：
next (error | completed)?*
这个表达了下面的意思：

• 一个序列可以有0个或或者多个元素
• 序列一旦接受到error或者completed事件，那么这个序列将不会产生新的元素。

序列（Observable）定义如下。

enum Event<Element>  {
    case next(Element)      // next element of a sequence
    case error(Swift.Error) // sequence failed with error
    case completed          // sequence terminated successfully
}

class Observable<Element> {
    func subscribe(_ observer: Observer<Element>) -> Disposable
}

protocol ObserverType {
    func on(_ event: Event<Element>)
}

序列发送completed或者error方法后这个序列内的所有元素将会被释放。
调用返回的subscription中的dispose方法将会取消产生新的元素，并且立即释放元素。

如果一个序列在有限的时间内中断，没有调用dispose或者disposed(by: disposeBag)方法并不会导致资源泄漏。并且，这些资源能够被使用直到这个序列完成（不在产生新的元素，或者调用error）。
如果一个序列自身并没有中断，比如按钮的tap事件，资源就会被永久地分配，除非使用takeUntil操作符或者其他方式来手工调用调用dispose，或者disposeBag内部自动调用。

Disposing

当我们正在序列上做某些操作时想要释放所有已经分配或者即将被分配的资源。有一种额外的方法可以中断序列。调用dispose方法。

例如

let scheduler = SerialDispatchQueueScheduler(qos: .default)
let subscription = Observable<Int>.interval(0.3, scheduler: scheduler)
    .subscribe { event in
        print(event)
    }

Thread.sleep(forTimeInterval: 2.0)

subscription.dispose()

打印

next(0)
next(1)
next(2)
next(3)
next(4)
next(5)

注意：手动调用dispose方法并不是一个好的编码习惯，有更好的方式来dispose subscription 比如使用DisposeBag，takeUntil操作符或者其他方式。

那么上述代码在调用dispose()方法后还会输出东西么，答案是，不一定。
如果这个scheduler是一个 serial scheduler(例如 MainScheduler) 并且dispose是在同一个serial scheduler当中调用，答案是。不会再输出。否则，答案是，会在输出。

你有两个平行进程

1. 一个正在产生元素。
2. 另一个正在dispose（清除）

上述关于调用是否会打印的问题仍然会存在，如果两个进程并不再同一个进程。
下面代码会调用dispose后不会在输出。

let subscription = Observable<Int>.interval(0.3, scheduler: scheduler)
            .observeOn(MainScheduler.instance)
            .subscribe { event in
                print(event)
            }

// ....

subscription.dispose() // called from main thread

let subscription = Observable<Int>.interval(0.3, scheduler: scheduler)
            .observeOn(serialScheduler)
            .subscribe { event in
                print(event)
            }

// ...

subscription.dispose() // executing on same `serialScheduler`

Dispose Bags

Dispose Bags在RX中如同ARC。
当申明一个DisposeBag后，每次添加disposables 时dispose会被调用。DisposeBag没有dispose方法，所以不能够明确的dispose方法。如果需要立即cleanup 只需要重新生成一个新的。

self.disposeBag = DisposeBag()

如果仍然有手动的需要，使用CompositeDisposable，他能满足这个需求，但是一旦调用dispose方法，他就会立即清理新加入的disposable。

Take until

另外的自动清理subscription的方法是takeUntil操作符。

sequence
    .takeUntil(self.rx.deallocated)
    .subscribe {
        print($0)
    }

Implicit Observable guarantees（Observable默许的保证）

有一对额外的保证，所有的序列必须遵守。

1. 无论Observable在哪一个进程产生了元素，如果Observable创建一个元素并且发送到observer（观察者）的observer.on(.next(nextElement))，那么直到observer.on 执行完毕，不能再发送下一个元素。
2. 也不能发送中断.completed 或者 .error。

考虑下面代码

someObservable
  .subscribe { (e: Event<Element>) in
      print("Event processing started")
      // processing
      print("Event processing ended")
  }

他会输出

Event processing started
Event processing ended
Event processing started
Event processing ended
Event processing started
Event processing ended

不会输出

Event processing started
Event processing started
Event processing ended
Event processing ended

创建你自己的 Observable

理解Observable至关重要的一点：
当一个Observable被创建以后，他不会做任何事。

Observable有很多种方式创建元素。一些是因为边界影响，一些是因为点击正在运行的进程例如点击鼠标等等。

然而，如果你只是调用一个方法返回一个Observable，序列不会执行生产，并且没有边界影响。Observable只是声明序列怎样生产，元素用什么参数生产。subscribe方法调用的时候开始生产元素。

例如你有一个类似下面的方法：

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)
  })

有很多方法能够创建Observable，最简单的方法可能是使用create方法。

让我们写一个方法创建一个返回一个元素序列的方法,这个方法叫做'just'这是其实现

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)
    })

结果

0

那么到底什么是create方法呢？

他只是一个方便的方法，能够让你用swift闭包轻松的实现subscribe。subscribe方法携带一个参数observer，返回一个disposable。
这样实现的序列是同步的，他会在subscribe调用返回 disposable之前产生元素或者中断。因为他实际上根returns的什么无关，所以生产元素的过程不能被打断。

当产生同步序列时，通常返回的disposable是NopDisposable的单例

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 ----

创建一个执行工作的Observable

事情变的越来约有趣了，现在让我们创建一个之前我们用到的interval操作符。
下面是

func myInterval(_ interval: TimeInterval) -> Observable<Int> {
    return Observable.create { observer in
        print("Subscribed")
        let timer = DispatchSource.makeTimerSource(queue: DispatchQueue.global())
        timer.scheduleRepeating(deadline: DispatchTime.now() + interval, interval: 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(0.1)

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 ----

如果你这样写

let counter = myInterval(0.1)

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()

Thread.sleep(forTimeInterval: 0.5)

subscription2.dispose()

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 ----

订阅的每一个订阅者产生他自己的分离的元素序列，操作符默认是没有状态的。极大部分操作符都是没有状态的。

分享订阅和shareReplay操作符

如果你想几个observers分享同一个订阅的事件（元素）怎么做呢？
有两点需要声明

1. 新订阅者开始观察之前怎样处理过去已经受到的元素。（重放最近一个，重放全部，重放最后n个）。
2. 怎样决定何时开始共享的订阅。（重新计数，手动或者其他算法）

通常的选择是一个混合体replay(1).refCount()又称作shareReplay()。

let counter = myInterval(0.1)
    .shareReplay(1)

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()

Thread.sleep(forTimeInterval: 0.5)

subscription2.dispose()

print("Ended ----")

输出

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

注意以上代码只有一个Subscribed和Disposed被调用。

观察URL也一样。下面是怎样用rx包装HTTP请求，同interval操作符有很多类似的地方。

extension Reactive where Base: URLSession {
    public func response(_ request: URLRequest) -> Observable<(Data, HTTPURLResponse)> {
        return Observable.create { observer in
            let task = self.dataTaskWithRequest(request) { (data, response, error) in
                guard let response = response, let data = data else {
                    observer.on(.error(error ?? RxCocoaURLError.Unknown))
                    return
                }

                guard let httpResponse = response as? HTTPURLResponse else {
                    observer.on(.error(RxCocoaURLError.nonHTTPResponse(response: response)))
                    return
                }

                observer.on(.next(data, httpResponse))
                observer.on(.completed)
            }

            task.resume()

            return Disposables.create {
                task.cancel()
            }
        }
    }
}

Operators（操作符）

RxSwift中实现了很多操作符。
操作符示意图可以在ReactiveX.io中查看
几乎所有的操作符示范可以在Playgrounds查看
使用这个Playgrounds请打开Rx.xcworkspace，编译RxSwift-macOS，然后在Rx.xcworkspace 树形图中打开Playgrounds。
如果你需要一个操作符，并且你不知道怎样找到他可以通过这里decision tree of operators

自定义操作符

有两种方式 可以自定义操作符

简单方式

所有的内部代码都是使用操作符高度优化的版本。所以他们不是最好的工具，这就是他为什么鼓励使用标准的操作符。
幸运的，有一种简单的方式来创建操作符。创建新的操作符的确是创建observables的一切。前面已经介绍过怎样创建操作符。
让我们看看map操作符是怎样实现的

extension ObservableType {
    func myMap<R>(transform: @escaping (E) -> R) -> Observable<R> {
        return Observable.create { observer in
            let subscription = self.subscribe { e in
                    switch e {
                    case .next(let value):
                        let result = transform(value)
                        observer.on(.next(result))
                    case .error(let error):
                        observer.on(.error(error))
                    case .completed:
                        observer.on(.completed)
                    }
                }

            return subscription
        }
    }
}

你可以这样使用它

let subscription = myInterval(0.1)
    .myMap { e in
        return "This is simply \(e)"
    }
    .subscribe(onNext: { n in
        print(n)
    })

输出

Subscribed
This is simply 0
This is simply 1
This is simply 2
This is simply 3
This is simply 4
This is simply 5
This is simply 6
This is simply 7
This is simply 8
...

Life happens 实际发生的

如果遇到使用自定义操作符也很难解决的情况怎么办呢？你可以先推出Rx，执行行动，然后再将结果传递给Rx用Subject s。

这种情况通常不应该被实践，这也不是一个好的编码习惯，但是你可以这样做。

let magicBeings: Observable<MagicBeing> = summonFromMiddleEarth()

  magicBeings
    .subscribe(onNext: { being in     // exit the Rx monad
        self.doSomeStateMagic(being)
    })
    .disposed(by: disposeBag)

  //
  //  Mess
  //
  let kitten = globalParty(   // calculate something in messy world
    being,
    UIApplication.delegate.dataSomething.attendees
  )
  kittens.on(.next(kitten))   // send result back to rx
  //
  // Another mess
  //

  let kittens = Variable(firstKitten) // again back in Rx monad

  kittens.asObservable()
    .map { kitten in
      return kitten.purr()
    }
    // ....

每次这样做，一些人会这样写

 kittens
    .subscribe(onNext: { kitten in
      // do something with kitten
    })
    .disposed(by: disposeBag)

所以尽量不要这样做。

Playgrounds

如果你不确定一个操作符的确切使用方法。playgrounds几乎包含了所有的用法。并且有实例有插图。
使用这个Playgrounds请打开Rx.xcworkspace，编译RxSwift-macOS，然后在Rx.xcworkspace 树形图中打开Playgrounds。
如果要在Playgrounds中查看例子的视图。打开Assistant Editor，你可以点击View > Assistant Editor > Show Assistant Editor打开Assistant Editor。

错误处理

有两种错误机制。

observables异步错误处理机制

错误处理比较简单。如果一个序列因为错误中断。那么所有相关的序列就会中断。断路逻辑。
你可以通过catch操作符修复错误的observables。有很多重载能让你在细节上具体修复。

调试编译错误

当写优雅的RxSwift/RxCocoa代码时。你将很依赖编译器来推断observables类型，这也是为什么swift让人惊叹。但是有时候这也会成为阻挠。

images = word
    .filter { $0.containsString("important") }
    .flatMap { word in
        return self.api.loadFlickrFeed("karate")
            .catchError { error in
                return just(JSON(1))
            }
      }

如果编译器报错，首先建议声明返回类型

images = word
    .filter { s -> Bool in s.containsString("important") }
    .flatMap { word -> Observable<JSON> in
        return self.api.loadFlickrFeed("karate")
            .catchError { error -> Observable<JSON> in
                return just(JSON(1))
            }
      }

如果还不能通过。你可以添加更多的声明

images = word
    .filter { (s: String) -> Bool in s.containsString("important") }
    .flatMap { (word: String) -> Observable<JSON> in
        return self.api.loadFlickrFeed("karate")
            .catchError { (error: Error) -> Observable<JSON> in
                return just(JSON(1))
            }
      }

建议首先声明闭包的返回类型和参数。
通常情况下如果解决了这个问题，你可以重新去掉声明来让你的代码清洁。

调试

单独使用debugger很有用，但是通常使用debug操作符更高效。debug操作符将会输出所有的事件，你也可以为这些事件添加标签。

debug就像一个调查，下面是一个示例

let subscription = myInterval(0.1)
    .debug("my probe")
    .map { e in
        return "This is simply \(e)"
    }
    .subscribe(onNext: { n in
        print(n)
    })

Thread.sleepForTimeInterval(0.5)

subscription.dispose()

输出

[my probe] subscribed
Subscribed
[my probe] -> Event next(Box(0))
This is simply 0
[my probe] -> Event next(Box(1))
This is simply 1
[my probe] -> Event next(Box(2))
This is simply 2
[my probe] -> Event next(Box(3))
This is simply 3
[my probe] -> Event next(Box(4))
This is simply 4
[my probe] dispose
Disposed

你也可以轻松的创建你自己版本的debug擦作符

extension ObservableType {
    public func myDebug(identifier: String) -> Observable<Self.E> {
        return Observable.create { observer in
            print("subscribed \(identifier)")
            let subscription = self.subscribe { e in
                print("event \(identifier)  \(e)")
                switch e {
                case .next(let value):
                    observer.on(.next(value))

                case .error(let error):
                    observer.on(.error(error))

                case .completed:
                    observer.on(.completed)
                }
            }
            return Disposables.create {
                   print("disposing \(identifier)")
                   subscription.dispose()
            }
        }
    }
 }

Enabling Debug Mode

为了用RxSwift.Resources调试内存泄漏，或者自动log所有的HTTP请求。你必须Enabling Debug Mode。
为了Enabling Debug Mode， 一个TRACE_RESOURCES标志必须被添加到RxSwift的编译设置里面。在其他swift标记的下面。
深入的描述和介绍怎样为Cocoapods & Carthage添加TRACE_RESOURCES可以参见这里
378

调试内存泄漏

在debug mode中Rx在全局变量Resources.total中记录所有的资源分配足迹。

如果你想要一个倒置资源泄漏的逻辑，最简单的方法就是printing RxSwift.Resources.total

/* add somewhere in
    func application(_ application: UIApplication, didFinishLaunchingWithOptions launchOptions: [UIApplicationLaunchOptionsKey : Any]? = nil)
    */
    _ = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
        .subscribe(onNext: { _ in
            print("Resource count \(RxSwift.Resources.total)")
        })

最高效的测试内存管理泄漏

• 导航到你要测试屏幕并且使用它
• 返回
• 观察初始化的资源数
• 第二次进入你要测试的屏幕并且使用它
• 返回
• 查看最后的资源数

如果初始化的资源数和最后的资源数不同，那么就可能存在内存泄漏。
两次导航的是因为第一次导航时强制加载了懒加载的资源。

Variables

Variables 呈现了一些observable的状态。没有包含值的Variable是不存在的，因为Variable初始化时需要一个值。
Variable包装了一个 Subject
. 跟确切地说是一个BehaviorSubject。跟BehaviorSubject不同的是，它只暴露value的接口，所以Variables不能用error中断。

他也会在subscription是广播它现在的值。

在variable 被 deallocated之后,他会通过.asObservable()来返回observable序列。

let variable = Variable(0)

print("Before first subscription ---")

_ = variable.asObservable()
    .subscribe(onNext: { n in
        print("First \(n)")
    }, onCompleted: {
        print("Completed 1")
    })

print("Before send 1")

variable.value = 1

print("Before second subscription ---")

_ = variable.asObservable()
    .subscribe(onNext: { n in
        print("Second \(n)")
    }, onCompleted: {
        print("Completed 2")
    })

print("Before send 2")

variable.value = 2

print("End ---")

输出

Before first subscription ---
First 0
Before send 1
First 1
Before second subscription ---
Second 1
Before send 2
First 2
Second 2
End ---
Completed 1
Completed 2

KVO

KVO是一个Objective-C的机制。这就意味值它并不是类型安全的。这个工程尝试去解决这个问题。
有两种方式支持KVO。

// KVO
extension Reactive where Base: NSObject {
    public func observe<E>(type: E.Type, _ keyPath: String, options: KeyValueObservingOptions, retainSelf: Bool = true) -> Observable<E?> {}
}

#if !DISABLE_SWIZZLING
// KVO
extension Reactive where Base: NSObject {
    public func observeWeakly<E>(type: E.Type, _ keyPath: String, options: KeyValueObservingOptions) -> Observable<E?> {}
}
#endif

怎么监听 UIView的frame
WARNING: UIKit isn't KVO compliant, but this will work.

view
  .rx.observe(CGRect.self, "frame")
  .subscribe(onNext: { frame in
    ...
  })

or

view
  .rx.observeWeakly(CGRect.self, "frame")
  .subscribe(onNext: { frame in
    ...
  })

rx.observe

rx.observe 是KVO机制的一个简单包装，但是它有更多的使用场景。

• 它能够用来观察以self开头的路径，或者在继承图中以祖先开头的路径(retainSelf = false)
• 它能够用来观察在继承图中以后代开头的路径(retainSelf = true)
• 路径必须是strong的，另外如果没有注销KVO监听，会有crashing的风险。

例如：

self.rx.observe(CGRect.self, "view.frame", retainSelf: false)

rx.observeWeakly

rx.observeWeakly相对rx.observe要更加耗时一些，因为它会在弱引用时处理对象的存储单元分配。
rx.observe能够使用的场景rx.observeWeakly都能够使用。

• 因为它不会retain被观察的目标，所以它能够观察继承图不明确的对象。
• 它能够观察weak属性
  例如：

someSuspiciousViewController.rx.observeWeakly(Bool.self, "behavingOk")

Observing structs

KVO是Objective-C 的机制，所以它严重依赖NSValue。
RxCocoa 能够实现对CGRect, CGSize and CGPoint structs 的KVO监听。
当需要对其他struct进行监听的时候，需要手动的从NSValue提取structure。
这里 是一个通过实现KVORepresentable协议为其他structs拓展KVO机制和rx.observe*方法的示范，

UI layer tips

当绑定到UIKit controls时，在UI层中，有明确的几点你的Observables需要满足。

Threading

Observables要在MainScheduler（UIThread）中发送值。这也是UIKit/Cocoa中要求的。
你的APIs在MainScheduler中返回结果是一个很好的方式，如果你在后台线程中绑定控件，Debug编译RxCocoa时会抛出一个异常提示你。
为了解决这个问题你要添加 observeOn(MainScheduler.instance).
URLSession extensions 默认不是在MainScheduler返回结果

Errors

你不能为UIKit controls绑定失败，因为那是没有定义的行为。
如果你不知道是否Observable会失败，你可以使用catchErrorJustReturn(valueThatIsReturnedWhenErrorHappens)来确保它不会失败，但是如果一个错误发生之后的序列仍然完整。
如果你想要后续序列的行为继续产生元素，需要使用一些retry的版本。

Sharing subscription

有时候你需要在你的UI层分享订阅，但是你不想让分离的HTTP请求来为成倍的UI元素绑定同样的数据。
你可以这样做

let searchResults = searchText
    .throttle(0.3, $.mainScheduler)
    .distinctUntilChanged
    .flatMapLatest { query in
        API.getSearchResults(query)
            .retry(3)
            .startWith([]) // clears results on new search term
            .catchErrorJustReturn([])
    }
    .shareReplay(1)          // <- notice the `shareReplay` operator

你通常想要的是一旦有结果就分享搜索的结果。这也是shareReplay意义所在。
UI层中在变化链末尾添加shareReplay是一个shareReplay是一个很好的经验法则，因为你真的想要分享计算结果。而不想在为成倍的界面元素绑定searchResults时开启分离的请求链接。
你也可以看看Driver单元，他是为包装这种shareReply请求，确保元素在主线程被观察并且没有错误倍绑定到UI而专门设计的。

Making HTTP requests

网络请求是首先要做的事情之一。
首先要绑定URLRequest对象，这个对象体现了你要做的工作。
Request决定了它是一个get或者post请求，请求body是什么，参数是什么。
这是一个怎样创建get请求的代码：

let req = URLRequest(url: URL(string: "http://en.wikipedia.org/w/api.php?action=parse&page=Pizza&format=json"))

如果只是想在其他observables之外执行这个请求：

let responseJSON = URLSession.shared.rx.json(request: req)

// no requests will be performed up to this point
// `responseJSON` is just a description how to fetch the response


let cancelRequest = responseJSON
    // this will fire the request
    .subscribe(onNext: { json in
        print(json)
    })

Thread.sleep(forTimeInterval: 3.0)

// if you want to cancel request after 3 seconds have passed just call
cancelRequest.dispose()

默认情况下URLSession extensions不在主线程返回结果。
如果你想要看更低层一些的响应入口：

URLSession.shared.rx.response(myURLRequest)
    .debug("my request") // this will print out information to console
    .flatMap { (data: NSData, response: URLResponse) -> Observable<String> in
        if let response = response as? HTTPURLResponse {
            if 200 ..< 300 ~= response.statusCode {
                return just(transform(data))
            }
            else {
                return Observable.error(yourNSError)
            }
        }
        else {
            rxFatalError("response = nil")
            return Observable.error(yourNSError)
        }
    }
    .subscribe { event in
        print(event) // if error happened, this will also print out error to console
    }

Logging HTTP traffic

debug模式RxCocoa默认会打印所有的网络请求，如果不想这样，请设置Logging.URLRequests过滤器。

// read your own configuration
public struct Logging {
    public typealias LogURLRequest = (URLRequest) -> Bool

    public static var URLRequests: LogURLRequest =  { _ in
    #if DEBUG
        return true
    #else
        return false
    #endif
    }
}

RxDataSources

RxDataSources 是一个类的集合，完全的实现了UITableViews 和UICollectionView 的data sources方法。
RxDataSources 在 这里.
RxExample 包含了方法示例和怎样使用它

原文链接
the basics, getting started with RxSwift