RxSwift I：开始学习 RxSwift

Posted on Edited on In

📒RxSwift I：开始学习 RxSwift

Rx_Logo_M.png

第一章：RxSwift 介绍

RxSwift是一个组合异步和事件驱动编程的库，通过使用可观察序列和功能样式运算符来，从而允许通过调度程序进行参数化执行。

RxSwift 在本质上简化了开发异步程序，允许代码对新数据作出反应，并以顺序和孤立的方式处理它。

介绍异步编程

51526546882_.pic.jpg

使用 Cocoa 和 UIKit 异步的API的问题在于：复杂的异步代码变得非常难写，部分原因是苹果SDK提供的API种类繁多。如NotificationCenter、Delegate、GCD、闭包、Combine。

异步编程词汇表：

  1. 状态（State），具体地说：共享可变状态
  2. 命令式编程（Imperative programming）
  3. 副作用（Side effects）
  4. 声明式编程（Declarative code）：声明式代码让您可以定义行为片段。只要有相关事件发生，RxSwift 就会运行这些行为，并提供一个不可变的、孤立的数据片段来处理。这样，您可以使用异步代码，但做出与简单 for 循环相同的假设：您正在使用不可变数据并且可以以顺序、确定的方式执行代码。
  5. 响应式系统（Reactive systems）

响应式系统（Reactive systems）是一个相当抽象的术语，它涵盖了Web或iOS应用程序，它们显示了大多数或全部以下特性：

  • 响应式设计（Responsive）：始终保持UI更新，代表了最新的应用程序状态。
  • 能复原的（Resilient）：每个行为都是独立定义的，并提供灵活的错误恢复。
  • 灵活的（Elastic）: 该代码处理不同的工作负载，通常实现诸如懒惰驱动数据收集、事件节流和资源共享等特性。
  • 消息驱动（Message driven）：组件使用基于消息的通信来提高可重用性和隔离性，解耦类的生命周期和实现。

RxSwift 基础

61526549641_.pic_hd.jpg

这个标志是一只电鳗。(Rx 项目曾经被称为Volta）

RxSwift 是微软开源的 ReactiveX 的Swift语言的实现。

RxSwift 在传统的Cocoa编程和纯函数编程之间找到了最佳位置。它允许您对事件作出反应, 方法是使用不可变的代码定义以确定性的、可组合的方式处理异步输入部分。

Rx代码的三个组成部分是 Observable, Operator和 Scheduler。

Observable<Element>类提供了Rx代码的基础：异步产生一系列事件的能力，它可以“携带”数据的不可变快照。简单来说，它允许类在一段时间内订阅其他类发出的值。

ObservableType 协议 (Observable需要遵循的) 非常简单。Observable可能发出 (并且Observer能接受) 仅三类型事件:

- next 下一个事件: “携带” 最新 (或 “下一个 “) 数据值的事件。这是Observer “接收” 值的方式。

- completed 已完成的事件: 此事件以成功终止事件序列。这意味着Observable完成其生命周期成功, 不会发出任何其他事件。

- error 错误事件: Observable终止带有错误, 不会发出其他事件.

两种不同的可观测序列: 有限和无限的。

由于它们是高度解耦和可组合的, 所以这些方法通常称为Operator。比如filter。

Operator也是高度可组合的，它们总是把数据作为输入并输出它们的结果，所以你可以用许多不同的方式轻松地将它们连接起来，实现比单个Operator自己能做的更多的事情。

Scheduler是GCD和OperationQueue的Rx等价物。

RxSwift将充当你的订阅（在左边）和Scheduler（在右边）之间的调度器，将工件发送到正确的上下文，并无缝地允许它们与彼此的输出一起工作。

71526551674_.pic_hd.jpg

要读取此关系图, 请在不同的计划程序中按预定的顺序 (1、2、3、…) 来执行彩色作品。例如:

·蓝色网络订阅以在基于自定义 NSOperation 的计划程序上运行的一段代码 (1) 开始。

·数据输出块作为下一个块 (2) 的输入, 它运行在一个不同的调度程序上, 它位于并发后台 GCD 队列中。

·最后, 在主线程调度程序上计划最后一块蓝色代码 (3), 以便用新数据更新 UI。

App architecture 应用的架构

值得一提的是，RxSwift并没有以任何方式改变应用程序的架构；它主要处理事件、异步数据序列和通用通信协议。

通过在苹果开发文档中实现MVC体系结构，可以创建具有Rx的应用程序。如果你喜欢的话，你也可以选择实现MVP架构或MVVM。RxSwift也可以帮你实现自己的单向数据架构。

微软的MVVM架构是专门针对在平台上创建的事件驱动软件开发的，该平台提供数据绑定。RxSwift和MVVM很好地结合在一起，在这本书的末尾，你会看到这个模式以及如何用RxSwift来实现它。

MVVM和RxSwift结合在一起的原因是，ViewModel允许您公开Observable属性，这些属性可以直接绑定到View Controller 代码中的UIKit控件。这使得绑定模型数据到UI非常简单地展示和编码：

81526552550_.pic.jpg

本书中的所有其他示例都使用MVC架构来保持示例代码简单易懂。

RxCocoa

RxSwift是通用Rx API的实现。因此，它不涉及任何Cocoa或UIKit类。

RxCocoa是RxSwift的配套库，所有的类都有助于UIKit和Cocoa的开发。除了具有一些高级类之外，RxCocoa还为许多UI组件添加了响应式扩展，以便您可以订阅不同的UI事件。

例如，使用RxCocoa订阅UISwitch的状态变化是非常容易的，例如：

1
2
3
4
toggleSwitch.rx.isOn
	.subcribe(onNext: {enabled in
    	print(enabled ? "it's ON" : "it's OFF")
	})

RxCocoa adds the rx.isOn property (among others) to the UISwitch class so you can subscribe to generally useful event sequences.

RxCocoa将rx.isOn属性（其中之一）添加到UISwitch类，这样您就可以订阅通常有用的Observable序列。

101526552877_.pic.jpg

安装

官方git：https://github.com/ReactiveX/RxSwift

使用 CocoaPods 、 Carthage 和 Swift Package Manager 均很方便集成RxSwift。

RxSwift 和 Combine

RxSwift 和 Combine（以及 Swift 中的其他相应式编程框架）共享许多通用语言和非常相似的概念。

RxSwift是一个较旧的，完善的框架，具有一些自己的原始概念，运算符名称和类型多样性，这主要是由于其多平台跨语言标准，该标准也适用于Linux，这对于Server-Side Swift非常有用。它也是开源的，所以如果你愿意，你可以直接为其核心做出贡献，并确切地看到它的特定部分是如何工作的。它与所有支持 Swift 的 Apple 平台版本兼容，一直支持 iOS 8。

Combine 是 Apple 新的、闪亮的框架，涵盖了类似的概念，但专门针对 Swift 和 Apple 自己的平台量身定制。它与 Swift 标准库共享许多通用语言，因此即使新手也觉得 API 非常熟悉。它仅支持从iOS 13，macOS 10.15等开始的较新的Apple平台。不幸的是，截至今天，它还没有开源，并且不支持Linux。

幸运的是，由于 RxSwift 和 Combine 非常相似，因此您的 RxSwift 知识可以轻松转移到 Combine，反之亦然。RxCombine（https://github.com/CombineCommunity/RxCombine）等项目允许您根据需要混合搭配 RxSwift Observables 和 Combine Publishers。

RxSwift 6.5.0 也迎来了 Swift Concurrency 的支持。提供了互操作性：

  1. await 调用Observable的values
  2. 封装 async Task 为 Observable

社区

RxSwift社区非常友好，思想开放，并且热衷于讨论模式，常用技巧或互相帮助。

更多的Rx库和实验，像雨后春笋一样的涌现，可以在这里找到：https://github.com/RxSwiftCommunity

可能最好的方式来满足许多对RxSwift感兴趣的人，这是Slack的频道：http://rxswift-slack.herokuapp.com。

Slack频道拥有约5000名成员！ 日常的主题包括：互相帮助，讨论RxSwift或其同伴库的潜在新功能，以及共享RxSwift博客文章和会议讲座。

第二章：Observables

Observable 是什么

Observable、observable sequence 和 sequence 在 Rx 都是一个意思。或者在其他Rx实现中称之为stream。

最好称之为 Observable，不过翻译过来还是序列顺口些。

Observable 的生命周期

  • observable发出包含元素的next事件。 它可以继续这样做，直到它：
  • …发出error事件并终止，或
  • …发出completed事件并终止。
  • 一旦observable被终止，它不能再发出事件。

将这种概念化的最佳方法之一是使用弹珠图（Marble Diagrams 基于时间轴上绘制的值）。

141526623515_.pic.jpg

151526623523_.pic.jpg

创建 Observable

类型擦除的ObservableType，也就是 Swift 中的泛型。

它代表了一种推式风格队列。

let observable: Observable = Observable.just(one)

订阅 Observable

订阅observable sequence的事件处理程序。

func subscribe(_ on: @escaping (RxSwift.Event<Self.E>) -> Swift.Void) -> Disposable

1
2
3
4
5
6
7
8
9
let one = 1
let two = 2
let three = 3

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

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

在指定的范围内生成一个整数的observable sequence，使用指定的scheduler生成和发送Observer消息。

1
static func range(start: Self.E, count: Self.E, scheduler: ImmediateSchedulerType = default) -> RxSwift.Observable<Self.E>
1
2
3
4
5
6
7
8
let observable = Observable<Int>.range(start: 1, count: 10)
observable
    .subscribe(onNext: { i in
        let n = Double(i)
        let fibonacci = Int(((pow(1.61803, n) - pow(0.61803, n)) /
            2.23606).rounded())
        print(fibonacci)
    })

Disposing 和 terminating

请记住, 在收到订阅之前, Observable的内容不会执行任何事情。它是触发一个Observable的开始发出事件的订阅, 直到它发出. error 或.completed完成事件并终止。您可以通过取消对它的订阅来手动终止Observable。

subscription.dispose()

单独管理每个订阅将是单调乏味的, 因此 RxSwift 引入 DisposeBag 类型。DisposeBag 持有 disposable协议的对象，通常是使用.disposed(by:) 方法, 并将调用dispose(), 当DisposeBag即将deallocated。

1
2
3
4
5
6
let observable = Observable.of("A", "B", "C")

let subscription = observable.subscribe { event in
    print(event)
}
subscription.dispose()

create操作符接受一个名为subscribe的参数。 它的工作是提供对可观察对象进行调用订阅的实现。 换句话说，它定义了将发送给订阅者的所有事件。

1
static func create(_ subscribe: @escaping (AnyObserver<String>) -> Disposable) -> Observable<String>

创建 observable 工厂类

可以创建一个可观察的工厂，向每个订阅者发布一个新的Observable，而不是创建一个等待订阅者的Observable。

1
static func deferred(_ observableFactory: @escaping () throws -> Observable<Int>) -> Observable<Int>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
let disposeBag = DisposeBag()

var flip = false

let factory: Observable<Int> = Observable.deferred {
    flip = !flip

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

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

    print()
}
1
2
3
4
123 
456 
123 
456

使用Traits

Traits是具有比常规Observable更窄的行为集合的一种Observable。它们的使用是可选的；您可以在任何可能使用trait的地方使用常规Observable。他们的目的是提供一种给你的API或者代码的读者更清楚地表达意图的方式。

RxSwift中有三种Traits: Single, Maybe 和 Completable.

Singles将发出.success(value)或.error事件。 .success(value)实际上是.next和.completed事件的组合。 这对于一次性进程非常有用，它可以成功并产生一个值或失败，例如下载数据或从磁盘加载数据。

下面的例子是读取 Copyright.txt 的文件内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
let disposeBag = DisposeBag()

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

func loadText(from filename: String) -> Single<String> {
    return Single.create { single in
        let disposable = Disposables.create()

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

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

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

        single(.success(contents))

        return disposable
    }
}

loadText(from: "Copyright")
    .subscribe {
        switch $0 {
        case .success(let string):
            print(string)
        case .error(let error):
            print(error)
        }
    }
    .disposed(by: disposeBag)

副作用

do Operator允许插入副作用，处理程序执行过程中并不会以任何方式更改发出的事件的操作。

打印debug 信息

debug Operator, 它将打印observable的每个事件的信息。

第三章：Subjects

Subject 是什么

既可以作为Observable，也可以作为Observer，这就是所谓的 Subjects。

也是最方便互操作的，BehaviorRelay比较常用。

RxSwift 中有四个subject类型:

  1. PublishSubject: 开始为空, 只向订阅者发出新元素.
  2. ReplaySubject: 用缓冲区大小, 并将保持元素的缓冲区大小, 并将其重播到新订阅者.
  3. BehaviorSubject: 从初始值开始, 将其重播或将最新的元素给新订阅者.
  4. AsyncSubject：仅发出序列中的最后一个next事件，并且仅当subject收到completed事件时才发出。这是一个很少使用的主题。

PublishSubject

1
2
3
/// Represents an object that is both an observable sequence as well as an observer.
/// Each notification is broadcasted to all subscribed observers.
public final class PublishSubject<Element>

当你只是想让订阅者只接受在订阅的时候以后发生的新的事件，直到他们unsubscribe，或者subject已经terminated以.completed或.error事件的方式，PublishSubject就可以派上用场。

RxSwiftPublishSubject

第一个订阅者在将 1 添加到subject后进行订阅，因此它不会收到该事件。不过，它确实得到了2和3。由于第二个订阅者在添加2之前不会加入，因此它只能获得3。

1
2
3
4
5
6
extension ObservableType {
    /// 添加带有`id`的观察者并打印每个发出的事件。
    func addObserver(_ id: String) -> Disposable {
        subscribe { print("观察者:", id, "事件:", $0) }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 从订阅开始向所有观察者广播新事件。
example("PublishSubject") {
    let disposeBag = DisposeBag()
    let subject = PublishSubject<String>()

    subject.addObserver("1").disposed(by: disposeBag)
    subject.onNext("🐶")
    subject.onNext("🐱")

    subject.addObserver("2").disposed(by: disposeBag)
    subject.onNext("🅰️")
    subject.onNext("🅱️")
    
    subject.onCompleted()
}
1
2
3
4
5
6
7
8
9
--- PublishSubject example ---
观察者: 1 事件: next(🐶)
观察者: 1 事件: next(🐱)
观察者: 1 事件: next(🅰️)
观察者: 2 事件: next(🅰️)
观察者: 1 事件: next(🅱️)
观察者: 2 事件: next(🅱️)
观察者: 1 事件: completed
观察者: 2 事件: completed

ReplaySubject

1
2
3
4
/// Represents an object that is both an observable sequence as well as an observer.
///
/// Each notification is broadcasted to all subscribed and future observers, subject to buffer trimming policies.
public class ReplaySubject<Element>

ReplaySubject将临时缓存, 或缓冲区, 它们发出的最新元素, 由您选择的 specified 大小决定。然后, 他们会将该缓冲区重播到新订阅者。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 向所有观察者广播新事件，并向新观察者广播之前指定的bufferSize大小的事件数。
example("ReplaySubject") {
    let disposeBag = DisposeBag()
    let subject = ReplaySubject<String>.create(bufferSize: 1)

    subject.addObserver("1").disposed(by: disposeBag)
    subject.onNext("🐶")
    subject.onNext("🐱")

    subject.addObserver("2").disposed(by: disposeBag)
    subject.onNext("🅰️")
    subject.onNext("🅱️")
    
    subject.onCompleted()
}
1
2
3
4
5
6
7
8
9
10
--- ReplaySubject example ---
观察者: 1 事件: next(🐶)
观察者: 1 事件: next(🐱)
观察者: 2 事件: next(🐱) // 相比PublishSubject多了订阅前一个事件
观察者: 1 事件: next(🅰️)
观察者: 2 事件: next(🅰️)
观察者: 1 事件: next(🅱️)
观察者: 2 事件: next(🅱️)
观察者: 1 事件: completed
观察者: 2 事件: completed

BehaviorSubject

1
2
3
/// Represents a value that changes over time.
/// Observers can subscribe to the subject to receive the last (or initial) value and all subsequent notifications.
public final class BehaviorSubject<Element>

当您希望使用最新数据预先填充View时, BehaviorSubject非常有用。例如, 可以将用户详情页中的控件绑定到BehaviorSubject, 以便在应用程序获取新数据时, 可以使用最新值来预先填充显示。

RxSwiftBehaviorSubject 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
example("BehaviorSubject") {
    let disposeBag = DisposeBag()
    let subject = BehaviorSubject(value: "🔴")

    subject.addObserver("1").disposed(by: disposeBag)
    subject.onNext("🐶")
    subject.onNext("🐱")

    subject.addObserver("2").disposed(by: disposeBag)
    subject.onNext("🅰️")
    subject.onNext("🅱️")
  
    subject.onCompleted()
}
1
2
3
4
5
6
7
8
9
10
11
--- BehaviorSubject example ---
观察者: 1 事件: next(🔴) // 相比ReplaySubject多了订阅前一个事件
观察者: 1 事件: next(🐶)
观察者: 1 事件: next(🐱)
观察者: 2 事件: next(🐱)
观察者: 1 事件: next(🅰️)
观察者: 2 事件: next(🅰️)
观察者: 1 事件: next(🅱️)
观察者: 2 事件: next(🅱️)
观察者: 1 事件: completed
观察者: 2 事件: completed

Relay

Relay在保持其replay行为的同时包装了subject。与其他subject不同，可以使用 accept(_:) 添加值，而非 onNext(_:)。这是因为Relay只能接受值，即不能向它们添加错误或已完成的事件。

PublishRelay将包装PublishSubject ，而BehaviorRelay将包装BehaviorSubject。中继与包装主体的区别在于，它们可以保证永远不会终止。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import RxRelay

example("PublishRelay") {
    let disposeBag = DisposeBag()
    let subject = PublishRelay<String>()

    subject.addObserver("1").disposed(by: disposeBag)
    subject.accept("🐶")
    subject.accept("🐱")

    subject.addObserver("2").disposed(by: disposeBag)
    subject.accept("🅰️")
    subject.accept("🅱️")
}
1
2
3
4
5
6
7
--- PublishRelay example ---
观察者: 1 事件: next(🐶) //next事件和PublishSubject一样，少了completed事件
观察者: 1 事件: next(🐱)
观察者: 1 事件: next(🅰️)
观察者: 2 事件: next(🅰️)
观察者: 1 事件: next(🅱️)
观察者: 2 事件: next(🅱️)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import RxRelay

example("BehaviorRelay") {
    let disposeBag = DisposeBag()
    let subject = BehaviorRelay<String>(value: "🔴")

    subject.addObserver("1").disposed(by: disposeBag)
    subject.accept("🐶")
    subject.accept("🐱")

    subject.addObserver("2").disposed(by: disposeBag)
    subject.accept("🅰️")
    subject.accept("🅱️")
}
1
2
3
4
5
6
7
8
9
--- BehaviorRelay example ---
观察者: 1 事件: next(🔴)  //next事件和BehaviorSubject，少了completed事件
观察者: 1 事件: next(🐶)
观察者: 1 事件: next(🐱)
观察者: 2 事件: next(🐱)
观察者: 1 事件: next(🅰️)
观察者: 2 事件: next(🅰️)
观察者: 1 事件: next(🅱️)
观察者: 2 事件: next(🅱️)

第四章：Observables 和 Subjects 实践

本章重点是在一个完整的应用开发中使用RxSwift，一步一步的学会如何把概念应用到实际项目中。

在 view controller 中使用 subject/relay

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
override func viewDidLoad() {
  super.viewDidLoad()

  images
    .subscribe(onNext: { [weak imagePreview] photos in
      guard let preview = imagePreview else { return }
      preview.image = photos.collage(size: preview.frame.size)
    })
    .disposed(by: bag)

  images
    .subscribe(onNext: { [weak self] photos in
      self?.updateUI(photos: photos)
    })
    .disposed(by: bag)
}

使用 subject 在 view controller 之间传值

view controller 之间传值可以通过 delegate，但是用 subject 更好。

Talkingtootherviewcontrollersviadelegate

Talkingtootherviewcontrollersviasubjects

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@IBAction func actionAdd() {
    let photosViewController = storyboard!.instantiateViewController(
        withIdentifier: "PhotosViewController") as! PhotosViewController

    navigationController!.pushViewController(photosViewController, animated: true)

    photosViewController.selectedPhotos
        .subscribe(
            onNext: { [weak self] newImage in
                guard let images = self?.images else { return }
                images.accept(images.value + [newImage])
            },
            onDisposed: {
                print("completed photo selection")
            }
        )
        .disposed(by: bag)
}

封装已有API为Observable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import Foundation
import Photos
import RxSwift
import UIKit

class PhotoWriter {
  enum Errors: Error {
    case couldNotSavePhoto
  }

  static func save(_ image: UIImage) -> Observable<String> {
    return Observable.create { observer in
      var savedAssetId: String?
      PHPhotoLibrary.shared().performChanges({
        let request = PHAssetChangeRequest.creationRequestForAsset(from: image)
        savedAssetId = request.placeholderForCreatedAsset?.localIdentifier
      }, completionHandler: { success, error in
        DispatchQueue.main.async {
          if success, let id = savedAssetId {
            observer.onNext(id)
            observer.onCompleted()
          } else {
            observer.onError(error ?? Errors.couldNotSavePhoto)
          }
        }
      })

      return Disposables.create()
    }
  }
}

RxSwift traits 实践

Single

Single 只有.success 或 .error 两种事件 。适合作为封装网络接口的返回值，要么成功，要么失败。single

1
2
3
4
5
6
7
8
9
PhotoWriter.save(image)
      .asSingle()
      .subscribe(onSuccess: { [weak self] id in
        self?.showMessage("Saved with id: \(id)")
        self?.actionClear()
        }, onError: { [weak self] error in
          self?.showMessage("Error", description: error.localizedDescription)
      })
      .disposed(by: bag)

Maybe

Maybe 有三种事件：.success、.completed 和 .error 。

和 Single 一样，你既可以通过 Maybe.create({ ... }) 直接创建，也可以使用 .asMaybe()

maybe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
PhotoWriter.save(image)
    .asMaybe()
    .subscribe(
        onSuccess: { [weak self] id in
            self?.showMessage("Saved with id: \(id)")
            self?.actionClear()
        },
        onError: { [weak self] error in
            self?.showMessage("Error", description: error.localizedDescription)
        },
        onCompleted: { [weak self] in
            self?.showMessage("Saved")
        }
    )
    .disposed(by: bag)

Completable

Completable 只有.completed 和 .error 两种事件。适合只关心是否完成，而不需要传递value的情况。使用方法：Completable.create({ ... })

completable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class PhotoWriter {
  enum Errors: Error {
    case couldNotSavePhoto
  }

  static func save(_ image: UIImage) -> Completable {
    return Completable.create { completable in
      PHPhotoLibrary.shared().performChanges({
        PHAssetChangeRequest.creationRequestForAsset(from: image)
      }, completionHandler: { success, error in
        DispatchQueue.main.async {
          if success {
            completable(.completed)
          } else {
            completable(.error(error ?? Errors.couldNotSavePhoto))
          }
        }
      })
      return Disposables.create()
    }
  }
}

PhotoWriter.save(image)
  .subscribe(
    onCompleted: { [weak self] in
      self?.showMessage("Saved")
    },
    onError: { [weak self] error in
      self?.showMessage("Error", description: error.localizedDescription)
    }
  )
  .disposed(by: bag)