Publisher<Int, Never>
    ┌────┐   ┌────┐      ┌────┐   ┌────┐
    │    │   │    │      │    │   │    │
 ───│ 13 │──▶│ 27 │─────▶│ 35 │──▶│ 56 │──│─▶
    │    │   │    │      │    │   │    │
    └────┘   └────┘      └────┘   └────┘
time 0:01     0:05        0:15     0:19  STOP

/// Declares that a type can transmit a sequence of values over time.
/// 声明一种可随时间传递值序列的类型。
///
/// There are four kinds of messages:
/// 共有四种消息：
///     subscription - A connection between `Publisher` and `Subscriber`.
///     订阅 - 发布者与订阅者之前的连接
///     value - An element in the sequence.
///     值 - 序列中的一个元素
///     error - The sequence ended with an error (`.failure(e)`).
///     错误 - 序列以错误结束（.failure(e)`）
///     complete - The sequence ended successfully (`.finished`).
///     完成 - 序列以成功结束（`.finished`）
/// Both `.failure` and `.finished` are terminal messages.
/// .failure` 和 `.finished` 均为终止消息。
///
/// You can summarize these possibilities with a regular expression:
/// 可使用正则表达式归纳其可能性：
///   value*(error|finished)?
///
/// Every `Publisher` must adhere to this contract.
/// 每个发布者都必须遵守该协议
@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public protocol Publisher {

    /// The kind of values published by this publisher.
    // 该发布者发射值的类型。
    associatedtype Output

    /// The kind of errors this publisher might publish.
    /// 该发布者可能发射错误的类型。
    ///
    /// Use `Never` if this `Publisher` does not publish errors.
    /// 若该发布者不会发射出错误可使用 `Never`
    associatedtype Failure : Error

    /// This function is called to attach the specified `Subscriber` to this `Publisher` by `subscribe(_:)`
    /// 该方法将在 `subscribe(_:)` 中被调用，把特定订阅者附加到该发布者上。
    ///
    /// - SeeAlso: `subscribe(_:)`
    /// - Parameters:
    ///     - subscriber: The subscriber to attach to this `Publisher`.
    ///                   once attached it can begin to receive values.
    ///                   附加到该发布者的订阅者，一旦附加上即开始接收值。
    func receive<S>(subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

/// 协议扩展
@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
extension Publisher {

    /// Attaches the specified subscriber to this publisher.
    /// 将特定订阅者附加到该发布者上。
    ///
    /// Always call this function instead of `receive(subscriber:)`.
    /// 应总是调用该方法而非 `receive(subscriber:)`。
    /// Adopters of `Publisher` must implement `receive(subscriber:)`. The implementation of `subscribe(_:)` in this extension calls through to `receive(subscriber:)`.
    /// 遵守 `Publisher` 的类型必须实现 `receive(subscriber:)`。该扩展中的 `subscribe(_:)` 实现将调用 `receive(subscriber:)`。
    /// - SeeAlso: `receive(subscriber:)`
    /// - Parameters:
    ///     - subscriber: The subscriber to attach to this `Publisher`. After attaching, the subscriber can start to receive values.
    ///                   被附加到发布者的订阅者。附加后，订阅者可以开始接收值。
    public func subscribe<S>(_ subscriber: S) where S : Subscriber, Self.Failure == S.Failure, Self.Output == S.Input
}

/// A protocol that declares a type that can receive input from a publisher.
/// 可以接收发布者作为输入的协议。
@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public protocol Subscriber : CustomCombineIdentifierConvertible {

    /// The kind of values this subscriber receives.
    /// 该订阅者接收的值的类型。
    associatedtype Input

    /// The kind of errors this subscriber might receive.
    /// 该订阅者接收可能接收到错误的类型。
    ///
    /// Use `Never` if this `Subscriber` cannot receive errors.
    /// 若该订阅者不会接收到错误可使用 `Never`
    associatedtype Failure : Error

    /// Tells the subscriber that it has successfully subscribed to the publisher and may request items.
    /// 告知订阅者已成功订阅发布者，且可以请求项目。
    ///
    /// Use the received `Subscription` to request items from the publisher.
    /// 使用接收到的 `Subscription` 向发布者请求项目。
    /// - Parameter subscription: A subscription that represents the connection between publisher and subscriber.
    ///                           代表发布者与订阅者联系的订阅。
    func receive(subscription: Subscription)

    /// Tells the subscriber that the publisher has produced an element.
    /// 告知订阅者，发布者已经产生了一个元素。
    ///
    /// - Parameter input: The published element.
    ///                    发布的元素。
    /// - Returns: A `Demand` instance indicating how many more elements the subcriber expects to receive.
    ///            一个 `Demand` 实例，指订阅者期望接收到的元素个数。
    func receive(_ input: Self.Input) -> Subscribers.Demand

    /// Tells the subscriber that the publisher has completed publishing, either normally or with an error.
    /// 告知订阅者，发布者已经完成发布，正常抑或发生错误。
    ///
    /// - Parameter completion: A `Completion` case indicating whether publishing completed normally or with an error.
    ///                         一个 `Completion` 情况，指发布正常完成，或者出错。
    func receive(completion: Subscribers.Completion<Self.Failure>)
}

/// A protocol representing the connection of a subscriber to a publisher.
/// 代表订阅者与发布者连接的协议。
///
/// Subcriptions are class constrained because a `Subscription` has identity -
/// defined by the moment in time a particular subscriber attached to a publisher.
/// 订阅受类型限制，因为 `Subscription` 的标示由连接到发布者的特定订阅者的时间定义。
/// Canceling a `Subscription` must be thread-safe.
/// 取消 `Subscription` 必须要线程安全。
///
/// You can only cancel a `Subscription` once.
/// `Subscription` 只能被取消一次。
///
/// Canceling a subscription frees up any resources previously allocated by attaching the `Subscriber`.
@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
public protocol Subscription : Cancellable, CustomCombineIdentifierConvertible {

    /// Tells a publisher that it may send more values to the subscriber.
    /// 告知发布者可以发送更多值到订阅者。
    func request(_ demand: Subscribers.Demand)
}

/// A publisher that eventually produces one value and then finishes or fails.
@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
final public class Future<Output, Failure> : Publisher where Failure : Error {

    public typealias Promise = (Result<Output, Failure>) -> Void

    public init(_ attemptToFulfill: @escaping (@escaping Future<Output, Failure>.Promise) -> Void)

    /// This function is called to attach the specified `Subscriber` to this `Publisher` by `subscribe(_:)`
    ///
    /// - SeeAlso: `subscribe(_:)`
    /// - Parameters:
    ///     - subscriber: The subscriber to attach to this `Publisher`.
    ///                   once attached it can begin to receive values.
    final public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
}

import Foundation
import Combine

public func example(of description: String, action: () -> Void) {
    print("\n——— Example of:", description, "———")
    action()
}


//example(of: "Notification") {
//    // 创建通知名称
//    let myNotification = Notification.Name("myNotification")
//    let center = NotificationCenter.default
//    let observer = center.addObserver(forName: myNotification,
//                                      object: nil,
//                                      queue: nil) { notification in
//                                        print("收到了通知")
//    }
//    center.post(name: myNotification, object: nil)
//    center.removeObserver(observer)
//}

example(of: "Publisher") {
    // 创建通知名称
    let myNotification = Notification.Name("myNotification")
    let center = NotificationCenter.default

    // 定义发布者，通知中心
    // func publisher(for name: Notification.Name, object: AnyObject? = nil) -> NotificationCenter.Publisher
    let publisher = NotificationCenter
        .default
        .publisher(for: myNotification,
                   object: nil)
    // 发布者 -> 订阅者
    // func sink(receiveValue: @escaping ((Notification) -> Void)) -> AnyCancellable
    let subscription = publisher.sink { _ in
        print("订阅者接收到了通知")
    }

    // 发送通知
    center.post(name: myNotification, object: nil)

    // 取消订阅，释放资源；若不手动调用，将直到发布者完成或存储订阅的对象销毁时才会取消订阅
    // _ = some_subscription：未存储的订阅将在创建时的程序流退出时被取消订阅
    subscription.cancel()

//    ——— Example of: Publisher ———
//    订阅者接收到了通知
}

example(of: "Just") {
    // 发布者，只发射一个值，然后完成
    let just = Just("你好，世界！")

    _ = just.sink(receiveCompletion: {
        // finished
        print("订阅者 1 接收到「完成」：", $0)
    }, receiveValue: {
        print("订阅者 1 接收到「值」：", $0)
    })

    _ = just.sink(receiveCompletion: {
        print("订阅者 2 接收到「完成」：", $0)
    }, receiveValue: {
        print("订阅者 2 接收到「值」：", $0)
    })

    // 当 Just 发布者每被订阅一次，就会发射一个值，并完成

//    订阅者 1 接收到「值」： 你好，世界！
//    订阅者 1 接收到「完成」： finished
//    订阅者 2 接收到「值」： 你好，世界！
//    订阅者 2 接收到「完成」： finished
}

example(of: "assign(to🔛)") {
    class SomeObject {
        var value: String = "" {
            didSet {
                // 被设置时打印
                print(value)
            }
        }
    }

    let obj = SomeObject()

    // 发射 A -> 发射 B
    let publisher = ["A", "B"].publisher
    // 分配到 obj 的 value（必须为 class 类型）
    _ = publisher.assign(to: \.value, on: obj)

//    ——— Example of: assign(to🔛) ———
//    A
//    B
}

example(of: "Custom Subscriber") {
    let publisher = (1...6).publisher
//    let publisher = ["A", "B", "C", "D", "E", "F"].publisher // ERROR: Input 类型不匹配

    // 自定义订阅者
    final class IntSubscriber: Subscriber {
        typealias Input = Int
        typealias Failure = Never

        // 由发布者调用
        func receive(subscription: Subscription) {
            // 指定最多可接收 3 个值
            subscription.request(.max(3))
        }

        // 值事件
        func receive(_ input: Int) -> Subscribers.Demand {
            print("收到的值为：", input)
            return .none // 等同于 .max(0)，不再调整接收数量
//            return .unlimited // 升级为无限制
        }

        // 完成事件
        func receive(completion: Subscribers.Completion<Never>) {
            print("收到完成")
        }
    }

    let subscriber = IntSubscriber()
    publisher.subscribe(subscriber)

//    ——— Example of: Custom Subscriber ———
//    收到的值为： 1
//    收到的值为： 2
//    收到的值为： 3
}

var subscriptions = Set<AnyCancellable>()

//example(of: "Future") {
//    // 发出整数，并永不失败
//    func futureIncrement(
//        integer: Int,
//        afterDelay delay: TimeInterval
//    ) -> Future<Int, Never> {
//        // public init(_ attemptToFulfill: @escaping (@escaping Future<Output, Failure>.Promise) -> Void)
//        Future<Int, Never>.init { promise in
//            print("Original") // 立即打印说明，创建后会立即执行（无需订阅者）
//            DispatchQueue.global().asyncAfter(deadline: .now() + delay) {
//                promise(.success(integer + 1)) // 发射出一个值，之后完成事件（需要订阅者）
//            }
//        }
//    }
//
//    // Future 是一个发布者，本质上是制造一个单一值并结束，或者失败。当值或者错误可用时，将通过调用闭包实现，该闭包被称作 Promise。
//
//    let future = futureIncrement(integer: 1, afterDelay: 3)
//
////    sleep(3);
//    future.sink(receiveCompletion: { print($0) }, receiveValue: { print($0) }).store(in: &subscriptions)
//
//    // ⚠️：Future 并没有重新执行 Promise，而是共享（或者说重播）了其输出
//    future.sink(receiveCompletion: { print("Second", $0) }, receiveValue: { print("Second", $0) }) .store(in: &subscriptions)
//
////    ——— Example of: Future ———
////    Original
////    2
////    finished
////    Second 2
////    Second finished
//}

example(of: "PassthoughSubject") {
    enum MyError: Error {
        case test
    }

    final class StringSubscriber: Subscriber {
        typealias Input = String
        typealias Failure = MyError

        func receive(subscription: Subscription) {
            subscription.request(.max(2))
        }

        func receive(_ input: String) -> Subscribers.Demand {
            print("接收到值", input)
            return input == "World" ? .max(1) : .none
        }

        func receive(completion: Subscribers.Completion<MyError>) {
            print("接收到完成", completion)
        }
    }

    let subscriber = StringSubscriber()

    // PassthroughSubject：可以按需要发布新值（其乐于传递值和完成事件），但仍需声明其发出值和错误的类型，且订阅者需要与其匹配
    let subject = PassthroughSubject<String, MyError>()
    // 订阅者订阅 subject
    subject.subscribe(subscriber)

    // 创建另外的订阅
    let subscription = subject.sink(receiveCompletion: { completion in
        print("Received completion (sink)", completion)
    }) { value in
        print("Received value (sink)", value)
    }

    subject.send("Hello")
    subject.send("World")

    subscription.cancel() // 取消订阅

    subject.send("Test")

//    subject.send(completion: .failure(MyError.test))
    subject.send(completion: .finished) // subscriber 的订阅取消
    subject.send("Test again")

//     通过 PassthroughSubject 传递值是将命令式代码桥接到 Combine 的声明性世界的一种方法。
}

example(of: "CurrentValueSubject") {
    // 发射 Int 值，不会出现错误
    let subject = CurrentValueSubject<Int, Never>(0)

    subject
        .print() // 打印 log 信息
        .sink(receiveValue: { print($0) }) // 订阅，并处理
        .store(in: &subscriptions) // inout：更新同一集合

    subject.send(1)
    // 访问其当前值
    print("print", subject.value)
    subject.send(2)
    print("print", subject.value)

    subject.value = 100
    print("print", subject.value)

    subject
         .print()
        .sink(receiveValue: { print("222:", $0) }) // 首个为当前值
        .store(in: &subscriptions)

    // CurrentValueSubject 的值只能为值，不能为完成事件
    // ERROR: Type 'Int' has no member 'finished'
    // subject.value = .finished

    // 手动取消
//    subscriptions.forEach {
//        $0.cancel()
//    }
    // 发射完成事件(无需 cancel）
    subject.send(completion: .finished)

    // 可以将预订存储在AnyCancellable的实例或集合中，以在取消初始化时接收自动取消。

}

example(of: "动态调整 Demand") {
    final class IntSubscriber: Subscriber {
        typealias Input = Int
        typealias Failure = Never
        func receive(subscription: Subscription) {
            subscription.request(.max(2))
        }

        func receive(_ input: Int) -> Subscribers.Demand {
            print("Received value", input)

            switch input {
            case 1:
                return .max(2) // 1
            case 3:
                return .max(1) // 2
            default:
                return .none // 3
            }
        }

        func receive(completion: Subscribers.Completion<Never>) {
          print("Received completion", completion)
        }
    }

    let subscriber = IntSubscriber()  // 2
    let subject = PassthroughSubject<Int, Never>()
    subject.subscribe(subscriber)
    subject.send(1) // 4
    subject.send(2) // 5
    subject.send(3)
    subject.send(4)
    subject.send(5)
    subject.send(6)
}

example(of: "类型擦除 Type erasure") {
    // 希望让订阅者订阅以接收来自发布者的事件，而又无法访问有关该发布者的其他详细信息。
    let subject = PassthroughSubject<Int, Never>()

    // 类型擦除的发布者：对订阅者隐藏发布者的详细信息
    // AnyPublisher<Int, Never>
    let publisher = subject.eraseToAnyPublisher()

    publisher.sink(receiveValue: { print($0) })
    .store(in: &subscriptions)

    subject.send(0)


    // AnyCancellable 是遵守 Cancellable 的类型擦除的类，调用者可以取消订阅，但不能访问订阅内部的信息


    // 另外一种情况是，当我们想使用一对公开和私有的属性，来允许这些属性的拥有者在私有的发布者上发送值，并让外界调用者访问公开的发布值来订阅，但不能发送值

    // AnyPublisher 没有 send(_:) 操作者，所以新值不能被再加到发布者上
    // publisher.send(1)

    // eraseToAnyPublisher() 将提供的发布者作为 AnyPublisher 的实例（不能指定为 Publish 是因为 Publish 是协议，无法作为实例的类型去约束），隐藏其是 PassthroughSubject 的事实


}

import Foundation
import Combine


public func example(of description: String, action: () -> Void) {
    print("\n——— Example of:", description, "———")
    action()
}


var subscriptions = [AnyCancellable]()

// 操作者是发布者
// 处理来自发布者的值的方法被称作操作者，
example(of: "收集") {
    ["A", "B", "C", "D", "E"].publisher
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })
    .store(in: &subscriptions)

    // 注意：
    // 使用 collect 和其它不限定数量或限制的缓冲操作符时要小心，其将使用无限制的内存来存储收到的值
    ["A", "B", "C", "D", "E"].publisher
//    .collect() // ["A", "B", "C", "D", "E"]
    .collect(2) // ["A", "B"], ["C", "D"], ["E"] // 限定为 2
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })
    .store(in: &subscriptions)
}

example(of: "map") {
    let formatter = NumberFormatter()
    formatter.numberStyle = .spellOut
    // map 是 Swift 里的高阶函数
    [123, 4, 56].publisher.map {
        formatter.string(from: NSNumber(integerLiteral: $0)) ?? ""
    }
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)
}

struct Coordinate {
    var x: Int
    var y: Int
}

func quadrantOf(x: Int, y: Int) -> String {
    return "\(x + y)"
}

example(of: "map key path") {
//    • map<T>(_:)
//    • map<T0, T1>(_:_:)
//    • map<T0, T1, T2>(_:_:_:)
    // T 代表给定键路径中找到的值的类型
    // 象限（Quadrants）是坐标几何（coordinate geometry）的一部分。有关更多信息，请访问 mathworld.wolfram.com/Quadrant.html。

    let publisher = PassthroughSubject<Coordinate, Never>()

    publisher
        .map(\.x, \.y) // 将 Coordinate keyPath map 出来(永远不会出错）
        .sink(receiveValue: { x, y in
            print("\(x) + \(y) =", quadrantOf(x: x, y: y))
        })
    .store(in: &subscriptions)

    publisher.send(Coordinate(x: 10, y: -8))
    publisher.send(Coordinate(x: 0, y: 5))
}

example(of: "tryMap(_:)") {
    // 尝试 map，出错时为 failure completion
    Just("Directory name that does not exist")
        .tryMap { try FileManager.default.contentsOfDirectory(atPath: $0) }
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)
}

public struct Chatter {
  public let name: String
  public let message: CurrentValueSubject<String, Never>

    public init(name: String, message: String) {
        self.name = name
        self.message = CurrentValueSubject(message)
    }
}

example(of: "flattern") {
    // flattern 可用于将多个上游发布者摊平为一个下游发布者，或者特别的，将这些发布者的发射物摊平

    // 上游发布者接收到的类型与 flatMap 返回的发布者通常会不一致
    // 常用情况：当我们想订阅发布者发射的值，但这些值本身又是他们自己的发布者

    let charlotte = Chatter(name: "Charlotte", message: "Hi I am Charlotte")
    let james = Chatter(name: "James", message: "Hi I am James")

    let chat = CurrentValueSubject<Chatter, Never>(charlotte) // 最初为 charlotte

//    chat
//        .sink(receiveValue: { print($0.message.value) })
//        .store(in: &subscriptions)

    chat
        .flatMap { $0.message } // 转换为 message 的 发布者
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

    // messgae 变更也可触发（不做 flatMap 的时候是不可以的）
    charlotte.message.value = "Charlotte: How's it going?"

    // 均可触发
    chat.value = james
}

example(of: "flatMap(maxPublishers") {
    // flatMap 将所有接收到的发布者的输出压缩为一个发布者。
    // 这带来了内存问题，因为它将缓存与您发送的发布者一样多的发布者，以更新它在下游发出的单个发布者。

    let charlotte = Chatter(name: "Charlotte", message: "Hi I am Charlotte")
    let james = Chatter(name: "James", message: "Hi I am James")

    let chat = CurrentValueSubject<Chatter, Never>(charlotte) // 最初发布者为 charlotte

    chat
        .flatMap(maxPublishers: .max(2)) { $0.message } // 最多接收上游 n 个发布者，其他将忽略，默认为无限
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

    // 更新第一个发布者
    charlotte.message.value = "Charlotte: How's it going?"

    // 第二个发布者为 james
    chat.value = james

    // 第三个发布者，但被忽略了
    let morgan = Chatter(name: "Morgan",
                         message: "Hey guys, what are you up to?")
    chat.value = morgan

    // 用的是 charlotte
    charlotte.message.value = "Did you hear something?"

//Hi I am Charlotte
//Charlotte: How's it going?
//Hi I am James
//Did you hear something?

}

example(of: "replaceNil") {
    ["A", nil, "B"]
        .publisher
        .replaceNil(with: "-") // 只做替换，并不改变类型
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

//    Optional("A")
//    Optional("-")
//    Optional("B")


    ["A", nil, "B"]
        .publisher
        .replaceNil(with: "-")
        .map { $0! } // 使用 map 转换类型（强制解包）
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

//    A
//    -
//    B

    let formatter = NumberFormatter()
    formatter.numberStyle = .spellOut

    [123, 4, 56].publisher.map {
        formatter.string(from: NSNumber(integerLiteral: $0)) ?? "" // ?? 可以返回另外一个可选类型，但 replaceNil 不可
    }
    .sink(receiveValue: { print($0) })
    .store(in: &subscriptions)

//    ["A", nil, "B"]
//    .publisher
//    .replaceNil(with: "-" as? String?)
//    .map { $0! } // 使用 map 转换类型（强制解包）
//    .sink(receiveValue: { print($0) })
//    .store(in: &subscriptions)
}

example(of: "replaceEmpty(with:)") {
    // 空发布者，立即发出完成事件；也可通过传递 false 到 completeImmediately 中，从而不发射任何东西，默认为 true
    // 可用于测试或演示只有完成事件时
//    let empty = Empty<Int, Never>.init(completeImmediately: false)
    let empty = Empty<Int, Never>()
    empty
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

//    finished


    empty
        .replaceEmpty(with: 1) // 替换空为发射 1
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

//    1
//    finished
}

example(of: "scan") {
    // 计算属性，随机数
    var dailyGainLoss: Int {
        .random(in: -10...10)
    }

    let august2019 = (0..<22)
        .map { _ in dailyGainLoss } // 数组
        .publisher

    august2019
        // 起始 50
        .scan(50) { latest, current in
            max(0, latest + current) // ➡️ 可预览
        }
    .sink(receiveValue: { _ in }) // 不做操作
//    .sink(receiveCompletion: { print($0) },
//          receiveValue: { print($0) })
    .store(in: &subscriptions)


//    •对发布者的输出执行操作的方法称为运算符。
//    •操作者也是发布者。
//使用任何缓冲值（例如collect或flatMap）的运算符时，请避免内存问题。
//    应用来自Swift标准库的现有功能知识时要谨记。 一些名称相似的Combine运算符的工作原理相同，而另一些则完全不同。
    // 操作符可以组合用于一个订阅
}

import Foundation
import Combine

var subscriptions = [AnyCancellable]()

// 带 try 的操作符，将提供 throw 的闭包，您在闭包内引发的任何错误都将以引发的错误终止发布者。

enum MyError: Error {
    case foo
}

example(of: "filter") {
    let numbers = (1...10).publisher
        numbers.filter {
            $0.isMultiple(of: 3) // 3 的倍数
        }
        .sink(receiveValue: {
            print($0)
        })
        .store(in: &subscriptions)

        // ---

        func throwFunc(val: Int) throws -> Bool {
            if val % 3 == 0 {
                return true
            }
            throw MyError.foo
        }

        numbers
            .tryFilter {
                try throwFunc(val: $0) // 失败后，即结束
        }
            .sink(receiveCompletion: {
                print($0) // failure(__lldb_expr_10.(unknown context at $1136214cc).(unknown context at $1136214d4).(unknown context at $1136214dc).MyError.foo)
            }, receiveValue: {
                print($0)
            })
            .store(in: &subscriptions)
}

example(of: "removeDuplicates") {
    let words = "a a a a b b b b" .components(separatedBy: " ")
        .publisher

    words
        .removeDuplicates() // 无需参数（遵守 Equable 协议）
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

//    a
//    b

        words
            .tryRemoveDuplicates(by: { first, second -> Bool in
                print(first, second)
//                throw Error
                return true
            })
            .sink(receiveCompletion: {
                print($0)
            }, receiveValue: {
                print($0)
            })
            .store(in: &subscriptions)
}

example(of: "compactMap") {
    let strings = ["a", "1.24", "3", "def", "45", "0.23"].publisher

    strings
        .compactMap {
            Float($0) // 无法转换会返回 nil
        }
        .sink(receiveValue: {
            print($0)
        })
        .store(in: &subscriptions)
//    1.24
//    3.0
//    45.0
//    0.23

    func throwFunc(val: String) throws -> Float? {
        if let fV = Float(val) {
            return fV
        }

        throw MyError.foo
    }

    strings
        .tryCompactMap {
            try throwFunc(val: $0)
        }
        .sink(receiveCompletion: {
            print($0)
        }, receiveValue: {
            print($0)
        })
        .store(in: &subscriptions)

    // failure(__lldb_expr_1.MyError.foo)
}

example(of: "ignoreOutput") {
    let numbers = (1...10_000).publisher

    numbers
        .ignoreOutput() // 忽略值输出
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)
//    finished
}

example(of: "first(where:)") {
    let numbers = (1...9).publisher

    numbers
        .print("numbers")
        .first(where: { $0 % 2 == 0 }) // 找到第一个即取消订阅并完成
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

//    numbers: receive subscription: (1...9)
//    numbers: request unlimited
//    numbers: receive value: (1)
//    numbers: receive value: (2)
//    numbers: receive cancel
//    2
}


example(of: "last(where:)") {
    let numbers = (1...9).publisher

    numbers
        .print("numbers")
        .last(where: { $0 % 2 == 0 }) // 需等待所有值发出完成后再执行
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

//numbers: receive subscription: (1...9)
//numbers: request unlimited
//numbers: receive value: (1)
//numbers: receive value: (2)
//numbers: receive value: (3)
//numbers: receive value: (4)
//numbers: receive value: (5)
//numbers: receive value: (6)
//numbers: receive value: (7)
//numbers: receive value: (8)
//numbers: receive value: (9)
//numbers: receive finished
//8
}

example(of: "last(where:)") {
    let numbers = PassthroughSubject<Int, Never>()
    numbers
        .last(where: { $0 % 2 == 0 })
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)
    numbers.send(1)
    numbers.send(2)
    numbers.send(3)
    numbers.send(4)
    numbers.send(5)

    // 没有完成就不会执行
//    numbers.send(completion: .finished)
}


example(of: "drop") {
    let numbers = (1...5).publisher

    numbers
        .dropFirst() // 抛弃第一个值
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

//    ——— Example of: dropFirst ———
//    2
//    3
//    4
//    5
//    finished
}

example(of: "drop(while:)") {
    let numbers = (1...10).publisher

    numbers
    .drop(while: {
        print("dropping")
        return $0 % 5 != 0
    }) // 条件成立时，丢弃，第一个不成立条件达成时，输出它以及后续的所有值（之后不再进入）
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)
//    dropping
//    dropping
//    dropping
//    dropping
//    dropping
//    5
//    6
//    7
//    8
//    9
//    10

    numbers
        .filter { $0 % 5 != 0 }
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)
//    1
//    2
//    3
//    4
//    6
//    7
//    8
//    9

    // 与 filter 的区别：
    // 1. 闭包内返回 true 时，while 会跳过这些值，filter 则是允许通过
    // 2. filter 会在后续执行相同的过滤策略，而 while 则第一次满足后（返回 false）就不再过滤
}

example(of: "drop(untilOutputFrom:)") {
    // 如果我们想当一个发布者准备好的时候，再去接受另外一个发布者的值，就可以使用该操作者
    let isReady = PassthroughSubject<Void, Never>()
    let taps = PassthroughSubject<Int, Never>()

    taps
        .drop(untilOutputFrom: isReady) // isReady 发射才开始接收
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

    (1...5).forEach { n in
        taps.send(n)
        if n == 3 {
//            isReady.send(completion: .finished) 发射完成事件是不行的
            isReady.send() // 发射
        }
    }

//    4
//    5
}

example(of: "prefix") {
    let numbers = (1...10).publisher

    // 类似 first 惰性，仅占用所需数量的值，即终止
    numbers
        .prefix(2) // 只接受前两个
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

//    1
//    2
//    finished

    numbers.prefix(while: { $0 < 3 }) // 只接受小于 3 的前缀，当起始不满足，则直接 finished
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })
    .store(in: &subscriptions)

//    1
//    2
//    finished
}

example(of: "prefix(untilOutputFrom:)") {
    let isReady = PassthroughSubject<Void, Never>()
    let taps = PassthroughSubject<Int, Never>()

    taps
        .prefix(untilOutputFrom: isReady) // 当 isReady 发出后不再可以发送
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

    (1...5).forEach { n in taps.send(n)
        if n == 2 {
            isReady.send() // 发射
        }
    }

//    1
//    2
//    finished


}

example(of: "Challenge") {
    let p = (1...100).publisher

    p
        .drop(while: { $0 < 50 })
        .prefix { $0 <= 70 }
        .filter { $0.isMultiple(of: 2) }
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

    // 惰性：接收所需数量的值，然后完成
    // 贪婪：确定值的范围，再接收满足条件的值
}

import UIKit
import Combine

var subscriptions = [AnyCancellable]()

enum MyError: Error {
    case foo
}


example(of: "prepend(Output...)") {

    let publisher = [3, 4].publisher

    publisher
        .prepend(1, 2) // 可变参数，插入前置的值
        .prepend(-1, 0) // 插入上流的最前面
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)
//-1
//0
//1
//2
//3
//4
}

example(of: "prepend(Sequence)") {

    let publisher = [3, 4].publisher

    publisher
        .prepend([3, 4])
        .prepend(Set(1...2)) // 集合无序，可能 1，2 或 2，1
        .prepend(stride(from: 6, to: 11, by: 2))
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

}

example(of: "prepend(Publisher)") {
    let publisher1 = [3, 4].publisher
    let publisher2 = [1, 2].publisher
// 类型需相同
    publisher1
        .prepend(publisher2) // 插入发布者，当插入的发布者完成后才发出第二个
        .sink(receiveCompletion: { print($0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

    let publisher3 = PassthroughSubject<Int, MyError>()
    let publisher4 = PassthroughSubject<Int, MyError>()

    publisher4
    .prepend(publisher3) // publisher3 不结束，则 publisher4 不继续
    .sink(receiveCompletion: { print($0) },
          receiveValue: { print($0) })
        .store(in: &subscriptions)

    publisher4.send(11)
    publisher4.send(22)

    publisher3.send(1)
    publisher3.send(2)
    publisher3.send(completion: .finished) // 完成后，4 开始

    publisher4.send(33)
    publisher4.send(44)
//    publisher3.send(completion: .failure(.foo)) // Error 则也停止

//    1
//    2
//    33
//    44
}

example(of: "append(Output...)") {
    let publisher = [1].publisher

    publisher
        .append(2, 3)
        .append(4)
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

//    1
//    2
//    3
//    4
}

example(of: "append(Output...) #2") { // 1
    let publisher = PassthroughSubject<Int, Never>()
    publisher
        .append(3, 4)
        .append(5)
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

    publisher.send(1)
    publisher.send(2)
    // 只有 publisher 完成，append 才会附加
    publisher.send(completion: .finished)

//    1
//    2
//    3
//    4
//    5
}

example(of: "append(Sequence)") {
    let publisher = [1, 2, 3].publisher
    publisher
        .append([4, 5])
        .append(Set([6, 7])) // 注意集合无序
        .append(stride(from: 8, to: 11, by: 2))
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)

//    1
//    2
//    3
//    4
//    5
//    7
//    6
//    8
//    10
}

example(of: "append(Publisher)") {
    let publisher1 = [1, 2].publisher
    let publisher2 = [3, 4].publisher

    publisher1
        .append(publisher2)
        .sink(receiveValue: { print($0) })
        .store(in: &subscriptions)
}

example(of: "switchToLatest") {
    let publisher1 = PassthroughSubject<Int, Never>()
    let publisher2 = PassthroughSubject<Int, Never>()
    let publisher3 = PassthroughSubject<Int, Never>()

    let publishers = PassthroughSubject<PassthroughSubject<Int, Never>, Never>()

    publishers
        .switchToLatest() // 只能用于发布发布者的发布者上
        .sink(receiveCompletion: { _ in print("Completed!") },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

    publishers.send(publisher1)
    publisher1.send(1)
    publisher1.send(2)

    publishers.send(publisher2)
    publisher1.send(3) // 已经切换到 publisher2，忽略 publisher1
    publisher2.send(4)
    publisher2.send(5)

    publishers.send(publisher3)
    publisher2.send(6)  // 已经切换到 publisher3，忽略 publisher2
    publisher3.send(7)
    publisher3.send(8)
    publisher3.send(9)

    publisher3.send(completion: .finished)
    publishers.send(completion: .finished)

    // 用途，用户点击按钮就会发送请求，再次点击触发新的请求，那么就可以用 switchToLatest 切换到只处理最新的请求
}

//example(of: "switchToLatest - Network Request") {
//    let url = URL(string: "https://source.unsplash.com/random")!
//
//    func getImage() -> AnyPublisher<UIImage?, Never> {
//        return URLSession.shared
//            .dataTaskPublisher(for: url) // URLSession 的 Combine 扩展
//            .map { data, _ in UIImage(data: data) }
//            .print("image")
//            .replaceError(with: nil) // 替代 error
//            .eraseToAnyPublisher() // 类型擦除
//    }
//
//    let taps = PassthroughSubject<Void, Never>()
//    taps
//        .map { _ in getImage() }
//        .switchToLatest()
//        .sink(receiveValue: { _ in })
//        .store(in: &subscriptions)
//    taps.send()
//
//    DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
//        taps.send()
//    }
//
//    DispatchQueue.main.asyncAfter(deadline: .now() + 3.1) {
//        taps.send()
//    }
//
////    ——— Example of: switchToLatest - Network Request ———
////    image: receive subscription: (DataTaskPublisher)
////    image: request unlimited
////    image: receive cancel // 被取消
////    image: receive subscription: (DataTaskPublisher)
////    image: request unlimited
////    image: receive cancel // 被取消
////    image: receive subscription: (DataTaskPublisher)
////    image: request unlimited
////    image: receive value: (Optional(<UIImage:0x600001beca20 anonymous {1080, 1350}>))
////    image: receive finished
//}

example(of: "merge") {
    let pub1 = PassthroughSubject<Int, Never>()
    let pub2 = PassthroughSubject<Int, Never>()

    // 最多合并 8 个
//    pub2.merge(with: <#T##Publisher#>, <#T##c: Publisher##Publisher#>, <#T##d: Publisher##Publisher#>, <#T##e: Publisher##Publisher#>, <#T##f: Publisher##Publisher#>, <#T##g: Publisher##Publisher#>, <#T##h: Publisher##Publisher#>)
    pub2.merge(with: pub1)
        .sink(receiveCompletion: { _ in
        print("completion")
    }, receiveValue: { print($0) })
        .store(in: &subscriptions)

    pub1.send(11)
    pub2.send(22)
    pub1.send(13)
    pub2.send(24)

    pub1.send(completion: .finished)

    pub2.send(25)

    pub2.send(completion: .finished) // 以最后一个结束为整个结束

//11
//22
//13
//24
//25
//completion
}

example(of: "combineLatest") {
    // combineLatest 可组合多个发布者，当其中任何一个发出值时，都会发布一个具有所有发布者最新值的元组

    let pub1 = PassthroughSubject<Int, Never>()
    let pub2 = PassthroughSubject<String, Never>()

    pub1.combineLatest(pub2)
        .sink(receiveCompletion: { print("completion", $0) },
              receiveValue: { print($0) })
        .store(in: &subscriptions)

    pub1.send(11) // combineLatest 首次需等待另一个发布者发布才会一同发布
    pub1.send(12)
    pub2.send("23")
    pub1.send(14)
    pub2.send("25")

    pub1.send(completion: .finished)

    pub2.send("26")

    pub2.send(completion: .finished) // 以最后一个结束为整个结束

//    (12, "23")
//    (14, "23")
//    (14, "25")
//    (14, "26")
//    completion finished
}

example(of: "zip") {
    let pub1 = PassthroughSubject<Int, Never>()
    let pub2 = PassthroughSubject<String, Never>()

    pub1.zip(pub2)
    .sink(receiveCompletion: { print("completion", $0) },
          receiveValue: { print($0) })
    .store(in: &subscriptions)

    pub1.send(11)
    pub1.send(12)
    pub2.send("23") // 取和他同为第一个的 11
    pub1.send(14)
    pub2.send("25") // 取和他同为第二个的 12

    pub1.send(completion: .finished)

    pub2.send("26") // 取和他同为第三个的 14
    pub2.send("27") // 没有和他同一个的 pub1 了，就不输出

    pub2.send(completion: .finished) // 以最后一个结束为整个结束

//    ——— Example of: zip ———
//    (11, "23")
//    (12, "25")
//    (14, "26")
//    completion finished

//    merge（with :)使您可以交错多个发布者的值。

}

import UIKit
import SwiftUI
import Combine
import PlaygroundSupport

// ObservableObject 只能由 class 实现，对应了 assign 只接受分配到 class 类型
public final class DisplayTimer: ObservableObject {
    @Published var current: CGFloat = 0
    var cancellable: Cancellable? = nil

    init() {
        DispatchQueue.main.async {
            self.cancellable = self.start()
        }
    }

    func start() -> Cancellable {
        return Timer.publish(every: 1.0 / 30,
                             on: .main,
                             in: .common)
            .autoconnect()
            .scan(CGFloat(0)) { counter, _ in counter + 1 } // 扫描每个值，做自增 + 1
            .sink(receiveValue: { counter in self.current = counter }) // 将 Timer 发射一次就更新到 current 上
    }

    func stop(after: TimeInterval) {
        DispatchQueue.main.asyncAfter(deadline: .now() + after) {
            self.cancellable?.cancel()
        }
    }
}

// 事件枚举，值、完成、失败
public enum Event: Equatable {
    case value
    case completion
    case failure
}

struct CombineEvent {
    let index: Int
    let time: TimeInterval
    let event: Event

    var groupTime: Int { Int(floor(time * 10.0)) }
    var isValue: Bool { self.event == .value }
    var isCompletion: Bool { self.event == .completion }
}

struct CombineEvents: Identifiable {
    let events: [CombineEvent]

    var time: TimeInterval { events[0].time }
    var id: Int { (events[0].groupTime << 16) | events.count } // id，为每个事件创造一个 id
}

class EventsHolder {
    var events: [CombineEvent]
    var startDate = Date()
    var nextIndex = 1

    init(events: [CombineEvent] = []) {
        self.events = events
    }

    func capture(_ event: Event) {
        let time = Date().timeIntervalSince(startDate)
        // if case 模式匹配
        if case .completion = event, // 校验 event 是否符合 .completion 匹配
            let lastEvent = events.last,
            (time - lastEvent.time) < 1.0 {
            events.append(CombineEvent(index: nextIndex, time: lastEvent.time + 1, event: event))
        } else {
            events.append(CombineEvent(index: nextIndex, time: time, event: event))
        }
        nextIndex += 1

        // 超过 15 秒，则移除事件
        while let e = events.first {
            guard (time - e.time) > 15.0 else { break }
            events.removeFirst()
        }
    }
}

// 事件值的视图
struct EventValueView: View {
    let index: Int

    var body: some View {
        Text("\(self.index)")
        .padding(3)
        .frame(width: 28, height: 28)
        // Sets whether text can compress the space between characters when necessary to fit text in a line.
        // 允许一行中挤压文字间空隙
// .allowsTightening(true)
        .minimumScaleFactor(0.1)
        .foregroundColor(.white)
        .background(Circle().fill(Color.blue))
        .fixedSize()
    }
}

struct EventCompletedView: View {
    var body: some View {
        Rectangle()
            .frame(width: 4, height: 38)
            .offset(x: 0, y: -3)
            .foregroundColor(.gray)
    }
}

struct EventFailureView: View {
    var body: some View {
        Text("X")
            .padding(3.0)
            .frame(width: 28, height: 28)
            .foregroundColor(.white)
            .background(Circle().fill(Color.red))
    }
}

struct EventView: View {
    let event: CombineEvent
    // An AnyView allows changing the type of view used in a given view hierarchy. Whenever the type of view used with an AnyView changes, the old hierarchy is destroyed and a new hierarchy is created for the new type.
    var body: some View {
        switch self.event.event {
        case .value:
            return AnyView(EventValueView(index: self.event.index))
        case .completion:
            return AnyView(EventCompletedView())
        case .failure:
            return AnyView(EventFailureView())
        }
    }
}

struct SimultaneousEventsView: View {
    let events: [CombineEvent]

    var body: some View {
        VStack(alignment: .center, spacing: 0.0) {
            ForEach(0..<self.events.count) {
                EventView(event: self.events[$0])
            }
        }
    }
}

extension SimultaneousEventsView: Identifiable {
    var id: Int { return events[0].groupTime }
}

struct ContentView: View {
    @ObservedObject var time = DisplayTimer()
    let holder: EventsHolder
    let title: String
    var groupEvents: [CombineEvents] {
        let d = Dictionary<Int, [CombineEvent]>.init(grouping: self.holder.events) {
            $0.groupTime
        }
        return d.keys.sorted().map {
            CombineEvents(events: d[$0]!.sorted { $0.index < $1.index })
        }
    }

    init(title: String, holder: EventsHolder = EventsHolder()) {
        self.title = title
        self.holder = holder
    }

    var body: some View {
        VStack(alignment: .leading) {
            Text(title)
                .fixedSize(horizontal: false, vertical: true)
                .padding(.bottom, 8)
            ZStack {
                Rectangle()
                    .frame(height: 2)
                    .foregroundColor(.gray)
                    .offset(x: 0, y: 14)
                ForEach(groupEvents) { group in
                    SimultaneousEventsView(events: group.events)
                        .offset(x: CGFloat(group.time) * 30.0 - self.time.current - 32, y: 0)
                }
            }
            .frame(minHeight: 32)
            .onReceive(time.objectWillChange) { _ in
                if self.holder.events.contains(where: { $0.event != .value }) {
                    self.time.stop(after: 0.5)
                }
            }

        }
    }

    func capture<T, F>(publisher: AnyPublisher<T, F>) {
        // 传入的 publisher 与其订阅者绑定，触发事件更新
        // 类型擦除的 订阅者
        let observer = AnySubscriber(receiveSubscription: { subscription in
            subscription.request(.unlimited)
        }, receiveValue: { (value: T) -> Subscribers.Demand in
            self.holder.capture(.value)
            return .unlimited
        }, receiveCompletion: { (completion: Subscribers.Completion<F>) in
            switch completion {
            case .finished: self.holder.capture(.completion)
            case .failure: self.holder.capture(.failure)
            }
        })

        publisher.subscribe(on: DispatchQueue.main)
        .subscribe(observer)
    }
}

extension Publisher {
    func displayEvents(in view: ContentView) {
        view.capture(publisher: self.eraseToAnyPublisher())
    }
}

var subscriptions = [AnyCancellable]()

enum MyError: Error {
    case foo
}

let valuesPerSecond = 1.0
let delayInSeconds = 1.5



// 延迟
let pub = PassthroughSubject<Date, Never>() // 发射 Date 值的发布者
let delayedPub = pub.delay(for: .seconds(delayInSeconds),
                           scheduler: DispatchQueue.main) // 延迟发布者

let sub = Timer
    .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
    .autoconnect() // autoconnect，它将在第一个订阅时立即连接。
    .subscribe(pub) // pub.send

let sourceTimeline = ContentView(title: "Emitted values (\(valuesPerSecond) per sec.):")
let delayedTimeline = ContentView(title: "Delayed values (with a \(delayInSeconds)s delay):")
let view = VStack(spacing: 50) {
  sourceTimeline
  delayedTimeline
}

PlaygroundPage.current.liveView = UIHostingController(rootView: view)

pub.displayEvents(in: sourceTimeline)
delayedPub.displayEvents(in: delayedTimeline)


//pub.sink(receiveValue: { print($0) }).store(in: &subscriptions)