MaterialRestableStream.swift

public class MaterialRestableStream<T>: MaterialStream<T> {
  public var atRest = true {
    didSet {
      if atRest != oldValue {
        for listener in listeners {
          (listener as! AnyObserver<Bool>).next(atRest)
        }
      }
    }
  }

  public func subscribeAtRest(_ next: @escaping (Bool) -> Void) -> Subscription {
    let strongStream: MaterialStream<Bool>
    if let stream = weakAtRestStream {
      strongStream = stream
    } else {
      strongStream = MaterialStream<Bool>(named: "atRest") { observer in
        observer.next(self.atRest) // Give the current state.
        self.listeners.add(observer)
        return Subscription {
          self.listeners.remove(observer)
        }
      }

      // Store the weak reference after storing to a strong reference,
      // otherwise the weak stream will be released on assignment.
      weakAtRestStream = strongStream
    }
    return strongStream.subscribe(next: next)
  }

  // atRestStream indirectly holds a strong reference to self,
  // so we keep a weak reference here to avoid a retain cycle.
  private weak var weakAtRestStream: (MaterialStream<Bool>)?
  private var listeners = NSMutableOrderedSet()
}