Add replayValues and shareReplayValues operators to LoadingSignal.

Sources/Connectable.swift

@@ -111,3 +111,21 @@ extension SignalProtocol {
     return replay(limit: limit).refCount()
   }
 }
+
+extension SignalProtocol where Element: LoadingStateProtocol {
+
+  /// Ensure that all observers see the same sequence of elements. Connectable.
+  public func replayValues(limit: Int = Int.max) -> ConnectableSignal<Signal<LoadingState<LoadingValue, LoadingError>, Error>> {
+    if limit == 0 {
+      return ConnectableSignal(source: map { $0.asLoadingState }, subject: PublishSubject())
+    } else {
+      return ConnectableSignal(source: map { $0.asLoadingState }, subject: ReplayLoadingValueSubject(bufferSize: limit))
+    }
+  }
+
+  /// Ensure that all observers see the same sequence of elements.
+  /// Shorthand for `replay(limit).refCount()`.
+  public func shareReplayValues(limit: Int = Int.max) -> Signal<LoadingState<LoadingValue, LoadingError>, Error> {
+    return replayValues(limit: limit).refCount()
+  }
+}

Sources/Subjects.swift

@@ -150,6 +150,62 @@ public final class ReplayOneSubject<Element, Error: Swift.Error>: Subject<Elemen
 }
 
 
+/// A subject that replies accumulated sequence of loading values to each observer.
+public final class ReplayLoadingValueSubject<Val, LoadingError: Swift.Error, Error: Swift.Error>: Subject<LoadingState<Val, LoadingError>, Error> {
+
+  private enum State {
+    case notStarted
+    case loading
+    case loadedOrFailedAtLeastOnce
+  }
+
+  private var state: State = .notStarted
+  private var buffer: ArraySlice<LoadingState<Val, LoadingError>> = []
+  private var terminalEvent: Event<LoadingState<Val, LoadingError>, Error>? = nil
+
+  public let bufferSize: Int
+
+  public init(bufferSize: Int = Int.max) {
+    self.bufferSize = bufferSize
+  }
+
+  public override func send(_ event: Event<LoadingState<Val, LoadingError>, Error>) {
+    switch event {
+    case .next(let loadingState):
+      switch loadingState {
+      case .loading:
+        if state == .notStarted {
+          state = .loading
+        }
+      case .loaded:
+        state = .loadedOrFailedAtLeastOnce
+        buffer.append(loadingState)
+        buffer = buffer.suffix(bufferSize)
+      case .failed:
+        state = .loadedOrFailedAtLeastOnce
+        buffer = [loadingState]
+      }
+    case .failed, .completed:
+      terminalEvent = event
+    }
+    super.send(event)
+  }
+
+  public override func willAdd(observer: @escaping Observer<LoadingState<Val, LoadingError>, Error>) {
+    switch state {
+    case .notStarted:
+      break
+    case .loading:
+      observer(.next(.loading))
+    case .loadedOrFailedAtLeastOnce:
+      buffer.forEach { observer(.next($0)) }
+    }
+    if let event = terminalEvent {
+      observer(event)
+    }
+  }
+}
+
 @available(*, deprecated, message: "All subjects now inherit 'Subject' that can be used in place of 'AnySubject'.")
 public final class AnySubject<Element, Error: Swift.Error>: SubjectProtocol {
   private let baseObserve: (@escaping Observer<Element, Error>) -> Disposable