ReactiveSwift的GitHub地址

官方文檔

原來(lái) RAC 中和 Swift 平臺(tái)相關(guān)的核?心代碼被單獨(dú)抽取成了了?一個(gè)新框架:ReactiveSwift 。
Swift 正在快速成?長(zhǎng)并且成?長(zhǎng)為?一個(gè)跨平臺(tái)的語(yǔ)?言重抖。把只和 Swift 相關(guān)的代碼抽取出來(lái)后露氮，ReactiveSwift 就 可以在其他平臺(tái)上被使?，?而不不只是局限在 CocoaTouch 和 Cocoa 中钟沛。

什么是熱信號(hào)和冷信號(hào)

• 熱信號(hào)是主動(dòng)的畔规，不管有沒(méi)有訂閱者，他都會(huì)啟動(dòng)恨统。如果發(fā)送當(dāng)時(shí)有訂閱者叁扫，就會(huì)同時(shí)接收到信號(hào)，如果沒(méi)有就不會(huì)接收到消息畜埋，給訂閱者發(fā)送消息不一定會(huì)收到莫绣，它可以有多個(gè)訂閱者，是一對(duì)多悠鞍，集合可以與訂閱者共享信息

• 冷信號(hào)是被動(dòng)的对室，只有當(dāng)你訂閱的時(shí)候，它才會(huì)發(fā)布消咖祭。當(dāng)有不同的訂閱者软驰，消息是重新完整發(fā)送。

如何給變量添加reactive計(jì)算屬性

public protocol ReactiveExtensionsProvider {}

extension ReactiveExtensionsProvider {
  /// 成員計(jì)算變量
    public var reactive: Reactive<Self> {
        return Reactive(self)
    }
 /// 靜態(tài)成員計(jì)算變量
    public static var reactive: Reactive<Self>.Type {
        return Reactive<Self>.self
    }
}

/// 定義泛型結(jié)構(gòu)體
public struct Reactive<Base> {
    public let base: Base
    fileprivate init(_ base: Base) {
        self.base = base
    }
}

如何使用

struct Person {
   var name = "小明"
}

extension Person:ReactiveExtensionsProvider{}

extension Reactive where Base == Person{
    func printName(){
        print("Person.Name:\(base.name)")
    }
}

let person = Person.init()
person.reactive.printName()//輸出：Person.Name:小明

Event

extension Signal {
  public enum Event {
        /// 信號(hào)中的值
        case value(Value)

        /// 信號(hào)由于錯(cuò)誤而終止心肪。不會(huì)再收到其他事件锭亏。
        case failed(Error)

        /// 信號(hào)成功終止。 不會(huì)再收到其他事件硬鞍。
        case completed

        /// 信號(hào)的事件產(chǎn)生已中斷慧瘤。 不會(huì)再收到其他事件。
        /// 重要：此事件并不表示成功或失敗信號(hào)完成固该。
        case interrupted
    
    /// 此事件是否為已完成事件锅减。
        public var isCompleted: Bool {
            switch self {
            case .completed:
                return true
            case .value, .failed, .interrupted:
                return false
            }
        }

        /// 此事件是否表示信號(hào)終止（即不再事件將被接收）。
        public var isTerminating: Bool {
            switch self {
            case .value:
                return false
            case .failed, .completed, .interrupted:
                return true
            }
        }
  }
}

Signal

Signal是ReactiveSwift中熱信號(hào)的實(shí)現(xiàn),會(huì)主動(dòng)將產(chǎn)出的事件Event向外發(fā)送, 而不會(huì)等到有人訂閱后才開(kāi)始發(fā)送. 這意味著如果訂閱的時(shí)機(jī)晚于發(fā)送的時(shí)機(jī), 那么訂閱者是不會(huì)收到訂閱時(shí)機(jī)之前的事件的.

Signal的實(shí)現(xiàn)

public final class Signal<Value, Error: Swift.Error> 

Signal只包含一個(gè)存儲(chǔ)屬性

private let core: Core 

Signal本質(zhì)上只是 Core的一個(gè)殼伐坏，Signal 添加觀察者怔匣，其實(shí)就是core 添加觀察者，所以接下來(lái)我們主要介紹的其實(shí)是Core. 我們來(lái)看看Core的主要定義:

///信號(hào)的狀態(tài)
private enum State {
        // `TerminationKind` is constantly pointer-size large to keep `Signal.Core`
        // allocation size independent of the actual `Value` and `Error` types.
        enum TerminationKind {
            case completed
            case interrupted
            case failed(Swift.Error)
            case silent

            init(_ event: Event) {
                switch event {
                case .value:
                    fatalError()
                case .interrupted:
                    self = .interrupted
                case let .failed(error):
                    self = .failed(error)
                case .completed:
                    self = .completed
                }
            }

            func materialize() -> Event? {
                switch self {
                case .completed:
                    return .completed
                case .interrupted:
                    return .interrupted
                case let .failed(error):
                    return .failed(error as! Error)
                case .silent:
                    return nil
                }
            }
        }

        /// 信號(hào)還是存活的
        case alive(Bag<Observer>, hasDeinitialized: Bool)

        /// “信號(hào)”已收到終止事件，并且即將終止每瞒。
        case terminating(Bag<Observer>, TerminationKind)

        /// 信號(hào)已終止金闽。
        case terminated
}


private final class Core {
        /// 與信號(hào)關(guān)聯(lián)的一次性對(duì)象。
        private let disposable: CompositeDisposable

        ///信號(hào)的狀態(tài)
        private var state: State

        /// 狀態(tài)鎖剿骨，確保所有狀態(tài)的同步 
        private let stateLock: Lock

        /// 事件鎖 用于確保事件在交付給觀察者的過(guò)程中的同步
        private let sendLock: Lock

    ///Core 的構(gòu)造方法
        fileprivate init(_ generator: (Observer, Lifetime) -> Void) {
            state = .alive(Bag(), hasDeinitialized: false)

            stateLock = Lock.make()
            sendLock = Lock.make()
            disposable = CompositeDisposable()

            // The generator observer retains the `Signal` core.
            generator(Observer(action: self.send, interruptsOnDeinit: true), Lifetime(disposable))
        }

        private func send(_ event: Event) {
      //判斷信號(hào)是否終止代芜，也就是 是否發(fā)送了  .failed, .completed, .interrupted這些事件
            if event.isTerminating {
                // Recursive events are disallowed for `value` events, but are permitted
                // for termination events. Specifically:
                //
                // - `interrupted`
                // It can inadvertently be sent by downstream consumers as part of the
                // `SignalProducer` mechanics.
                //
                // - `completed`
                // If a downstream consumer weakly references an object, invocation of
                // such consumer may cause a race condition with its weak retain against
                // the last strong release of the object. If the `Lifetime` of the
                // object is being referenced by an upstream `take(during:)`, a
                // signal recursion might occur.
                //
                // So we would treat termination events specially. If it happens to
                // occur while the `sendLock` is acquired, the observer call-out and
                // the disposal would be delegated to the current sender, or
                // occasionally one of the senders waiting on `sendLock`.

                self.stateLock.lock()

                if case let .alive(observers, _) = state {
                    self.state = .terminating(observers, .init(event))
                    self.stateLock.unlock()
                } else {
                    self.stateLock.unlock()
                }

                tryToCommitTermination()
            } else {
        //信號(hào)沒(méi)有終止
                self.sendLock.lock()
                self.stateLock.lock()

                if case let .alive(observers, _) = self.state {
                    self.stateLock.unlock()

                    for observer in observers {
                        observer.send(event)
                    }
                } else {
                    self.stateLock.unlock()
                }

                self.sendLock.unlock()

                // Check if the status has been bumped to `terminating` due to a
                // terminal event being sent concurrently or recursively.
                //
                // The check is deliberately made outside of the `sendLock` so that it
                // covers also any potential concurrent terminal event in one shot.
                //
                // Related PR:
                // https://github.com/ReactiveCocoa/ReactiveSwift/pull/112
                //
                // While calling `tryToCommitTermination` is sufficient, this is a fast
                // path for the recurring value delivery.
                //
                // Note that this cannot be `try` since any concurrent observer bag
                // manipulation might then cause the terminating state being missed.
                stateLock.lock()
                if case .terminating = state {
                    stateLock.unlock()
                    tryToCommitTermination()
                } else {
                    stateLock.unlock()
                }
            }
        }

        /// 將任何將來(lái)的事件發(fā)送給給定的觀察者來(lái)觀察信號(hào)。
        ///
        /// - parameters:
        ///   - observer: An observer to forward the events to.
        ///
        /// - returns: A `Disposable` which can be used to disconnect the observer,
        ///            or `nil` if the signal has already terminated.
        fileprivate func observe(_ observer: Observer) -> Disposable? {
            var token: Bag<Observer>.Token?

            stateLock.lock()

            if case let .alive(observers, hasDeinitialized) = state {
                var newObservers = observers
                token = newObservers.insert(observer)
                self.state = .alive(newObservers, hasDeinitialized: hasDeinitialized)
            }

            stateLock.unlock()

            if let token = token {
                return AnyDisposable { [weak self] in
                    self?.removeObserver(with: token)
                }
            } else {
                observer.sendInterrupted()
                return nil
            }
        }

        /// 通過(guò)觀察者關(guān)聯(lián)Token屬性來(lái)移除觀察者
        ///
        /// - parameters:
        ///   - token: The token of the observer to remove.
        private func removeObserver(with token: Bag<Observer>.Token) {
            stateLock.lock()

            if case let .alive(observers, hasDeinitialized) = state {
                var newObservers = observers
                let observer = newObservers.remove(using: token)
                self.state = .alive(newObservers, hasDeinitialized: hasDeinitialized)

                // Ensure `observer` is deallocated after `stateLock` is
                // released to avoid deadlocks.
                withExtendedLifetime(observer) {
                    // Start the disposal of the `Signal` core if the `Signal` has
                    // deinitialized and there is no active observer.
                    tryToDisposeSilentlyIfQualified(unlocking: stateLock)
                }
            } else {
                stateLock.unlock()
            }
        }

        /// Try to commit the termination, or in other words transition the signal from a
        /// terminating state to a terminated state.
        ///
        /// It fails gracefully if the signal is alive or has terminated. Calling this
        /// method as a result of a false positive `terminating` check is permitted.
        ///
        /// - precondition: `stateLock` must not be acquired by the caller.
        private func tryToCommitTermination() {
            // Acquire `stateLock`. If the termination has still not yet been
            // handled, take it over and bump the status to `terminated`.
            stateLock.lock()

            if case let .terminating(observers, terminationKind) = state {
                // Try to acquire the `sendLock`, and fail gracefully since the current
                // lock holder would attempt to commit after it is done anyway.
                if sendLock.try() {
                    state = .terminated
                    stateLock.unlock()

                    if let event = terminationKind.materialize() {
                        for observer in observers {
                            observer.send(event)
                        }
                    }

                    sendLock.unlock()
                    disposable.dispose()
                    return
                }
            }

            stateLock.unlock()
        }

        /// Try to dispose of the signal silently if the `Signal` has deinitialized and
        /// has no observer.
        ///
        /// It fails gracefully if the signal is terminating or terminated, has one or
        /// more observers, or has not deinitialized.
        ///
        /// - precondition: `stateLock` must have been acquired by the caller.
        ///
        /// - parameters:
        ///   - stateLock: The `stateLock` acquired by the caller.
        private func tryToDisposeSilentlyIfQualified(unlocking stateLock: Lock) {
            assert(!stateLock.try(), "Calling `unconditionallyTerminate` without acquiring `stateLock`.")

            if case let .alive(observers, true) = state, observers.isEmpty {
                // Transition to `terminated` directly only if there is no event delivery
                // on going.
                if sendLock.try() {
                    self.state = .terminated
                    stateLock.unlock()
                    sendLock.unlock()

                    disposable.dispose()
                    return
                }

                self.state = .terminating(Bag(), .silent)
                stateLock.unlock()

                tryToCommitTermination()
                return
            }

            stateLock.unlock()
        }

        /// Acknowledge the deinitialization of the `Signal`.
        fileprivate func signalDidDeinitialize() {
            stateLock.lock()

            // Mark the `Signal` has now deinitialized.
            if case let .alive(observers, false) = state {
                state = .alive(observers, hasDeinitialized: true)
            }

            // Attempt to start the disposal of the signal if it has no active observer.
            tryToDisposeSilentlyIfQualified(unlocking: stateLock)
        }

        deinit {
            disposable.dispose()
        }
    }

Signal 實(shí)現(xiàn)了 SignalProtocol 和 SignalProducerConvertible

public protocol SignalProtocol: class {
    /// The type of values being sent by `self`.
    associatedtype Value

    /// The type of error that can occur on `self`.
    associatedtype Error: Swift.Error

    /// The materialized `self`.
    var signal: Signal<Value, Error> { get }
}
//返回Signal對(duì)象本身
extension Signal: SignalProtocol {
    public var signal: Signal<Value, Error> {
        return self
    }
}

public protocol SignalProducerConvertible {
    /// The type of values being sent by `self`.
    associatedtype Value

    /// The type of error that can occur on `self`.
    associatedtype Error: Swift.Error

    /// The `SignalProducer` representation of `self`.
    var producer: SignalProducer<Value, Error> { get }
}
//返回SignalProducer對(duì)象
extension Signal: SignalProducerConvertible {
    public var producer: SignalProducer<Value, Error> {
        return SignalProducer(self)
    }
}

Signal的使用

    /// Initialize a Signal that will immediately invoke the given generator,
    /// then forward events sent to the given observer.
    ///
    /// - note: The disposable returned from the closure will be automatically
    ///         disposed if a terminating event is sent to the observer. The
    ///         Signal itself will remain alive until the observer is released.
    ///
    /// - parameters:
    ///   - generator: A closure that accepts an implicitly created observer
    ///                that will act as an event emitter for the signal.
    public init(_ generator: (Observer, Lifetime) -> Void) {
        core = Core(generator)
    }


 /// Create a `Signal` that will be controlled by sending events to an
 /// input observer.
 ///
 /// - note: The `Signal` will remain alive until a terminating event is sent
 ///         to the input observer, or until it has no observers and there
 ///         are no strong references to it.
 ///
 /// - parameters:
 ///   - disposable: An optional disposable to associate with the signal, and
 ///                 to be disposed of when the signal terminates.
 ///
 /// - returns: A 2-tuple of the output end of the pipe as `Signal`, and the input end
 ///            of the pipe as `Signal.Observer`.
 public static func pipe(disposable: Disposable? = nil) -> (output: Signal, input: Observer) {
     var observer: Observer!

     let signal = self.init { innerObserver, lifetime in
         observer = innerObserver
         lifetime += disposable
     }

     return (signal, observer)
 }

enum MyError:Error {
    case customError
    var desc:String{
        return "這是自定義的錯(cuò)誤"
    }
}
typealias MySignal = Signal<String,MyError>

var obser:MySignal.Observer?

let signal = MySignal.init { (observer, lifetime) in
    obser = observer
    lifetime.observeEnded {
        print("信號(hào)生命周期結(jié)束")
    }
}
//添加觀察的對(duì)象
let observer = MySignal.Observer.init { (event) in
    print("MySignal.Observer: \(event)")
}
signal.observe(observer)

//觀察事件
signal.observe { (event) in
    print("observe方法: \(event)")
}
//觀察發(fā)送的值 推薦用這種方式 不要使用observeValues方法
signal.observeResult { (result) in
    switch result{
    case .success(let value):
        print(value)
    case .failure(let error):
        print(error)
    }
}
//觀察失敗事件
signal.observeFailed { (error) in
    print(error.desc)
}
//觀察完成事件
signal.observeCompleted {
    print("完成了")
}

obser?.send(value: "這是發(fā)送的值")
/*輸出：
 MySignal.Observer: VALUE 這是發(fā)送的值
 observe方法: VALUE 這是發(fā)送的值
 這是發(fā)送的值
 */

obser?.send(error: .customError)
obser?.send(value: "這是發(fā)送的值")
/*輸出  發(fā)送錯(cuò)誤之后浓利，信號(hào)的生命周期就結(jié)束了挤庇，在發(fā)送value也不會(huì)打印了
 MySignal.Observer: FAILED customError
 observe方法: FAILED customError
 customError
 這是自定義的錯(cuò)誤
 信號(hào)生命周期結(jié)束
 */

obser?.send(value: "這是發(fā)送的值 1111111")
obser?.sendCompleted()
obser?.send(value: "這是發(fā)送的值 2222222")
/*輸出  sendCompleted之后，信號(hào)的生命周期就結(jié)束了贷掖，在發(fā)送value也不會(huì)打印了
 MySignal.Observer: VALUE 這是發(fā)送的值 1111111
 observe方法: VALUE 這是發(fā)送的值 1111111
 這是發(fā)送的值 1111111
 MySignal.Observer: COMPLETED
 observe方法: COMPLETED
 完成了
 信號(hào)生命周期結(jié)束
 */

///接下來(lái)和上面的用法一樣
let (signal1,obser1) = MySignal.pipe()

SignalProducer

public struct SignalProducer<Value, Error: Swift.Error> 

SignalProducer是ReactiveSwift中冷信號(hào)的實(shí)現(xiàn)

SignalProducer也是只包含一個(gè)存儲(chǔ)屬性

SignalProducer本質(zhì)上只是 SignalProducerCore的一個(gè)殼嫡秕，所以接下來(lái)我們主要介紹的其實(shí)是SignalProducerCore. 我們來(lái)看看SignalProducerCore的主要定義:

fileprivate let core: SignalProducerCore<Value, Error>

///   SignalProducerCore 是SignalProducer的實(shí)際實(shí)現(xiàn)core 抽象類，需要SignalProducerCore的子類
///   才能創(chuàng)建SignalProducer
internal class SignalProducerCore<Value, Error: Swift.Error> {
    /// `Instance` represents an instance of `Signal` created from a
    /// `SignalProducer`. In addition to the `Signal` itself, it includes also the
    /// starting side effect and an interrupt handle for this particular instance.
    ///
    /// It is the responsibility of the `Instance` consumer to ensure the
    /// starting side effect is invoked exactly once, and is invoked after observations
    /// has properly setup.
    struct Instance {
        let signal: Signal<Value, Error>
        let observerDidSetup: () -> Void
        let interruptHandle: Disposable
    }
  //這是一個(gè)抽象方法苹威，需要子類重新實(shí)現(xiàn)
    func makeInstance() -> Instance {
        fatalError()
    }

    /// Start the producer with an observer created by the given generator.
    ///
    /// The created observer **must** manaully dispose of the given upstream interrupt
    /// handle iff it performs any event transformation that might result in a terminal
    /// event.
    ///
    /// - parameters:
    ///   - generator: The closure to generate an observer.
    ///
    /// - returns: A disposable to interrupt the started producer instance.
    @discardableResult
    func start(_ generator: (_ upstreamInterruptHandle: Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
        fatalError()
    }

    /// Perform an action upon every event from `self`. The action may generate zero or
    /// more events.
    ///
    /// - precondition: The action must be synchronous.
    ///
    /// - parameters:
    ///   - transform: A closure that creates the said action from the given event
    ///                closure.
    ///
    /// - returns: A producer that forwards events yielded by the action.
    internal func flatMapEvent<U, E>(_ transform: @escaping Signal<Value, Error>.Event.Transformation<U, E>) -> SignalProducer<U, E> {
        return SignalProducer<U, E>(TransformerCore(source: self, transform: transform))
    }
}

創(chuàng)建SignalProducer的方式

///通過(guò)SignalProducerConvertible協(xié)議去創(chuàng)建SignalProducer
///
/// - parameters:
///   - base: The entity to convert from.
    public init<T: SignalProducerConvertible>(_ base: T) where T.Value == Value, T.Error == Error {
        self = base.producer
    }

///通過(guò)Signal去創(chuàng)建SignalProducer
/// If the Disposable returned from `start()` is disposed or a terminating
/// event is sent to the observer, the given signal will be disposed.
///
/// - parameters:
///   - signal: A signal to observe after starting the producer.
    public init(_ signal: Signal<Value, Error>) {
        self.init { observer, lifetime in
            lifetime += signal.observe(observer)
        }
    }

/// 通過(guò)SignalProducerCore去創(chuàng)建SignalProducer
///
/// - parameters:
///   - core: The `SignalProducer` core.
    internal init(_ core: SignalProducerCore<Value, Error>) {
        self.core = core
    }

/// 通過(guò)GeneratorCore創(chuàng)建SignalProducer
///
/// - parameters:
///   - value: A value that should be sent by the `Signal` in a `value`
///            event.
public init(value: Value) {
  self.init(GeneratorCore { observer, _ in
    observer.send(value: value)
    observer.sendCompleted()
  })
}


/// 通過(guò)Sequence創(chuàng)建SignalProducer
///
/// - parameters:
///   - values: A sequence of values that a `Signal` will send as separate
///             `value` events and then complete.
    public init<S: Sequence>(_ values: S) where S.Iterator.Element == Value {
        self.init(GeneratorCore(isDisposable: true) { observer, disposable in
            for value in values {
                observer.send(value: value)

                if disposable.isDisposed {
                    break
                }
            }

            observer.sendCompleted()
        })
    }

    /// Creates a producer for a Signal that will immediately send the values
    /// from the given sequence, then complete.
    ///
    /// - parameters:
    ///   - first: First value for the `Signal` to send.
    ///   - second: Second value for the `Signal` to send.
    ///   - tail: Rest of the values to be sent by the `Signal`.
    public init(values first: Value, _ second: Value, _ tail: Value...) {
        self.init([ first, second ] + tail)
    }
//以上只是部分初始化方法昆咽，

SignalProducerCore 默認(rèn)有3個(gè)子類他們都是private final 類型，不能被外面調(diào)用

1.SignalCore

private final class SignalCore<Value, Error: Swift.Error>: SignalProducerCore<Value, Error> {
    private let _make: () -> Instance

    init(_ action: @escaping () -> Instance) {
        self._make = action
    }

    @discardableResult
    override func start(_ generator: (Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
        let instance = makeInstance()
        instance.signal.observe(generator(instance.interruptHandle))
        instance.observerDidSetup()
        return instance.interruptHandle
    }

    override func makeInstance() -> Instance {
        return _make()
    }
}

2.TransformerCore

/// `TransformerCore` composes event transforms, and is intended to back synchronous
/// `SignalProducer` operators in general via the core-level operator `Core.flatMapEvent`.
///
/// It takes advantage of the deferred, single-observer nature of SignalProducer. For
/// example, when we do:
///
/// ```
/// upstream.map(transform).compactMap(filteringTransform).start()
/// ```
///
/// It is contractually guaranteed that these operators would always end up producing a
/// chain of streams, each with a _single and persistent_ observer to its upstream. The
/// multicasting & detaching capabilities of Signal is useless in these scenarios.
///
/// So TransformerCore builds on top of this very fact, and composes directly at the
/// level of event transforms, without any `Signal` in between.
///
/// - note: This core does not use `Signal` unless it is requested via `makeInstance()`.
private final class TransformerCore<Value, Error: Swift.Error, SourceValue, SourceError: Swift.Error>: SignalProducerCore<Value, Error> {
    private let source: SignalProducerCore<SourceValue, SourceError>
    private let transform: Signal<SourceValue, SourceError>.Event.Transformation<Value, Error>

    init(source: SignalProducerCore<SourceValue, SourceError>, transform: @escaping Signal<SourceValue, SourceError>.Event.Transformation<Value, Error>) {
        self.source = source
        self.transform = transform
    }

    @discardableResult
    internal override func start(_ generator: (Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
        // Collect all resources related to this transformed producer instance.
        let disposables = CompositeDisposable()

        source.start { upstreamInterrupter in
            // Backpropagate the terminal event, if any, to the upstream.
            disposables += upstreamInterrupter

            var hasDeliveredTerminalEvent = false

            // Generate the output sink that receives transformed output.
            let output = generator(disposables)

            // Wrap the output sink to enforce the "no event beyond the terminal
            // event" contract, and the disposal upon termination.
            let wrappedOutput: Signal<Value, Error>.Observer.Action = { event in
                if !hasDeliveredTerminalEvent {
                    output.send(event)

                    if event.isTerminating {
                        // Mark that a terminal event has already been
                        // delivered.
                        hasDeliveredTerminalEvent = true

                        // Disposed of all associated resources, and notify
                        // the upstream too.
                        disposables.dispose()
                    }
                }
            }

            // Create an input sink whose events would go through the given
            // event transformation, and have the resulting events propagated
            // to the output sink above.
            let input = transform(wrappedOutput, Lifetime(disposables))

            // Return the input sink to the source producer core.
            return Signal<SourceValue, SourceError>.Observer(input)
        }

        // Manual interruption disposes of `disposables`, which in turn notifies
        // the event transformation side effects, and the upstream instance.
        return disposables
    }

    internal override func flatMapEvent<U, E>(_ transform: @escaping Signal<Value, Error>.Event.Transformation<U, E>) -> SignalProducer<U, E> {
        return SignalProducer<U, E>(TransformerCore<U, E, SourceValue, SourceError>(source: source) { [innerTransform = self.transform] action, lifetime in
            return innerTransform(transform(action, lifetime), lifetime)
        })
    }

    internal override func makeInstance() -> Instance {
        let disposable = SerialDisposable()
        let (signal, observer) = Signal<Value, Error>.pipe(disposable: disposable)

        func observerDidSetup() {
            start { interrupter in
                disposable.inner = interrupter
                return observer
            }
        }

        return Instance(signal: signal,
                        observerDidSetup: observerDidSetup,
                        interruptHandle: disposable)
    }
}

3.GeneratorCore

/// `GeneratorCore` wraps a generator closure that would be invoked upon a produced
/// `Signal` when started. The generator closure is passed only the input observer and the
/// cancel disposable.
///
/// It is intended for constant `SignalProducers`s that synchronously emits all events
/// without escaping the `Observer`.
///
/// - note: This core does not use `Signal` unless it is requested via `makeInstance()`.
private final class GeneratorCore<Value, Error: Swift.Error>: SignalProducerCore<Value, Error> {
    private let isDisposable: Bool
    private let generator: (Signal<Value, Error>.Observer, Disposable) -> Void

    init(isDisposable: Bool = false, _ generator: @escaping (Signal<Value, Error>.Observer, Disposable) -> Void) {
        self.isDisposable = isDisposable
        self.generator = generator
    }

    @discardableResult
    internal override func start(_ observerGenerator: (Disposable) -> Signal<Value, Error>.Observer) -> Disposable {
        // Object allocation is a considerable overhead. So unless the core is configured
        // to be disposable, we would reuse the already-disposed, shared `NopDisposable`.
        let d: Disposable = isDisposable ? _SimpleDisposable() : NopDisposable.shared
        generator(observerGenerator(d), d)
        return d
    }

    internal override func makeInstance() -> Instance {
        let (signal, observer) = Signal<Value, Error>.pipe()
        let d = AnyDisposable(observer.sendInterrupted)

        return Instance(signal: signal,
                                     observerDidSetup: { self.generator(observer, d) },
                                     interruptHandle: d)
    }
}

SignalProducer 實(shí)現(xiàn)了SignalProducerConvertible 和 SignalProducerProtocol協(xié)議

public protocol SignalProducerConvertible {
    /// The type of values being sent by `self`.
    associatedtype Value

    /// The type of error that can occur on `self`.
    associatedtype Error: Swift.Error

    /// The `SignalProducer` representation of `self`.
    var producer: SignalProducer<Value, Error> { get }
}

/// A protocol for constraining associated types to `SignalProducer`.
public protocol SignalProducerProtocol {
    /// The type of values being sent by `self`.
    associatedtype Value

    /// The type of error that can occur on `self`.
    associatedtype Error: Swift.Error

    /// The materialized `self`.
    var producer: SignalProducer<Value, Error> { get }
}

extension SignalProducer: SignalProducerConvertible, SignalProducerProtocol {
    public var producer: SignalProducer {
        return self
    }
}