Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Migrate six operators to the new abstraction. #814

Merged
merged 3 commits into from
Jan 11, 2021
Merged

Migrate six operators to the new abstraction. #814

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d2b52b5
Collect, CombinePrevious, SkipRepeats, UniqueValues
andersio Jan 1, 2021
a85539e
Reduce, ScanMap
andersio Jan 1, 2021
d7bb822
Merge branch 'master' into anders/more-operators-3
andersio Jan 9, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions ReactiveSwift.xcodeproj/project.pbxproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,30 @@
9A1D067D1D948A2300ACF44C /* UnidirectionalBindingSpec.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A1D067C1D948A2200ACF44C /* UnidirectionalBindingSpec.swift */; };
9A1D067E1D948A2300ACF44C /* UnidirectionalBindingSpec.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A1D067C1D948A2200ACF44C /* UnidirectionalBindingSpec.swift */; };
9A1D067F1D948A2300ACF44C /* UnidirectionalBindingSpec.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A1D067C1D948A2200ACF44C /* UnidirectionalBindingSpec.swift */; };
9A2D5CDB259F8398005682ED /* Collect.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CDA259F8398005682ED /* Collect.swift */; };
9A2D5CDC259F8398005682ED /* Collect.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CDA259F8398005682ED /* Collect.swift */; };
9A2D5CDD259F8398005682ED /* Collect.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CDA259F8398005682ED /* Collect.swift */; };
9A2D5CDE259F8398005682ED /* Collect.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CDA259F8398005682ED /* Collect.swift */; };
9A2D5CE5259F852B005682ED /* CombinePrevious.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CE4259F852B005682ED /* CombinePrevious.swift */; };
9A2D5CE6259F852B005682ED /* CombinePrevious.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CE4259F852B005682ED /* CombinePrevious.swift */; };
9A2D5CE7259F852B005682ED /* CombinePrevious.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CE4259F852B005682ED /* CombinePrevious.swift */; };
9A2D5CE8259F852B005682ED /* CombinePrevious.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CE4259F852B005682ED /* CombinePrevious.swift */; };
9A2D5CEF259F85AE005682ED /* SkipRepeats.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CEE259F85AE005682ED /* SkipRepeats.swift */; };
9A2D5CF0259F85AE005682ED /* SkipRepeats.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CEE259F85AE005682ED /* SkipRepeats.swift */; };
9A2D5CF1259F85AE005682ED /* SkipRepeats.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CEE259F85AE005682ED /* SkipRepeats.swift */; };
9A2D5CF2259F85AE005682ED /* SkipRepeats.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CEE259F85AE005682ED /* SkipRepeats.swift */; };
9A2D5CF9259F8634005682ED /* UniqueValues.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CF8259F8634005682ED /* UniqueValues.swift */; };
9A2D5CFA259F8634005682ED /* UniqueValues.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CF8259F8634005682ED /* UniqueValues.swift */; };
9A2D5CFB259F8634005682ED /* UniqueValues.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CF8259F8634005682ED /* UniqueValues.swift */; };
9A2D5CFC259F8634005682ED /* UniqueValues.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5CF8259F8634005682ED /* UniqueValues.swift */; };
9A2D5D03259F8C39005682ED /* Reduce.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D02259F8C39005682ED /* Reduce.swift */; };
9A2D5D04259F8C39005682ED /* Reduce.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D02259F8C39005682ED /* Reduce.swift */; };
9A2D5D05259F8C39005682ED /* Reduce.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D02259F8C39005682ED /* Reduce.swift */; };
9A2D5D06259F8C39005682ED /* Reduce.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D02259F8C39005682ED /* Reduce.swift */; };
9A2D5D0D259F8D1F005682ED /* ScanMap.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D0C259F8D1F005682ED /* ScanMap.swift */; };
9A2D5D0E259F8D1F005682ED /* ScanMap.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D0C259F8D1F005682ED /* ScanMap.swift */; };
9A2D5D0F259F8D1F005682ED /* ScanMap.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D0C259F8D1F005682ED /* ScanMap.swift */; };
9A2D5D10259F8D1F005682ED /* ScanMap.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5D0C259F8D1F005682ED /* ScanMap.swift */; };
9A2D5C9F259F8059005682ED /* TakeFirst.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5C9E259F8059005682ED /* TakeFirst.swift */; };
9A2D5CA0259F8059005682ED /* TakeFirst.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5C9E259F8059005682ED /* TakeFirst.swift */; };
9A2D5CA1259F8059005682ED /* TakeFirst.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9A2D5C9E259F8059005682ED /* TakeFirst.swift */; };
Expand Down Expand Up @@ -308,6 +332,12 @@
9A1A4F981E16961C006F3039 /* ValidatingPropertySpec.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ValidatingPropertySpec.swift; sourceTree = "<group>"; };
9A1B824020835EEC00EB7C09 /* ResultExtensions.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ResultExtensions.swift; sourceTree = "<group>"; };
9A1D067C1D948A2200ACF44C /* UnidirectionalBindingSpec.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = UnidirectionalBindingSpec.swift; sourceTree = "<group>"; };
9A2D5CDA259F8398005682ED /* Collect.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Collect.swift; sourceTree = "<group>"; };
9A2D5CE4259F852B005682ED /* CombinePrevious.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CombinePrevious.swift; sourceTree = "<group>"; };
9A2D5CEE259F85AE005682ED /* SkipRepeats.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = SkipRepeats.swift; sourceTree = "<group>"; };
9A2D5CF8259F8634005682ED /* UniqueValues.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = UniqueValues.swift; sourceTree = "<group>"; };
9A2D5D02259F8C39005682ED /* Reduce.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Reduce.swift; sourceTree = "<group>"; };
9A2D5D0C259F8D1F005682ED /* ScanMap.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ScanMap.swift; sourceTree = "<group>"; };
9A2D5C9E259F8059005682ED /* TakeFirst.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TakeFirst.swift; sourceTree = "<group>"; };
9A2D5CAD259F8112005682ED /* TakeLast.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TakeLast.swift; sourceTree = "<group>"; };
9A2D5CB7259F8199005682ED /* TakeWhile.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TakeWhile.swift; sourceTree = "<group>"; };
Expand Down Expand Up @@ -472,6 +502,12 @@
9AFA491A24E9A925003D263C /* Filter.swift */,
9AFA491F24E9A988003D263C /* CompactMap.swift */,
9AFA492424E9B15C003D263C /* Operators.swift */,
9A2D5CDA259F8398005682ED /* Collect.swift */,
9A2D5CE4259F852B005682ED /* CombinePrevious.swift */,
9A2D5CEE259F85AE005682ED /* SkipRepeats.swift */,
9A2D5CF8259F8634005682ED /* UniqueValues.swift */,
9A2D5D02259F8C39005682ED /* Reduce.swift */,
9A2D5D0C259F8D1F005682ED /* ScanMap.swift */,
9A2D5C9E259F8059005682ED /* TakeFirst.swift */,
9A2D5CAD259F8112005682ED /* TakeLast.swift */,
9A2D5CB7259F8199005682ED /* TakeWhile.swift */,
Expand Down Expand Up @@ -978,18 +1014,25 @@
57A4D1B61BA13D7A00F7D4B1 /* Event.swift in Sources */,
57A4D1B81BA13D7A00F7D4B1 /* Scheduler.swift in Sources */,
9A9100E21E0E6E680093E346 /* ValidatingProperty.swift in Sources */,
9A2D5CF2259F85AE005682ED /* SkipRepeats.swift in Sources */,
9A2D5CBB259F8199005682ED /* TakeWhile.swift in Sources */,
57A4D1B91BA13D7A00F7D4B1 /* Action.swift in Sources */,
57A4D1BA1BA13D7A00F7D4B1 /* Property.swift in Sources */,
9A2D5C5C259F7B31005682ED /* Materialize.swift in Sources */,
9A090C171DA0309E00EE97CA /* Reactive.swift in Sources */,
57A4D1BB1BA13D7A00F7D4B1 /* Signal.swift in Sources */,
9AFA492824E9B15C003D263C /* Operators.swift in Sources */,
9A2D5D10259F8D1F005682ED /* ScanMap.swift in Sources */,
9A2D5CE8259F852B005682ED /* CombinePrevious.swift in Sources */,
9A67963E1F6059440058C5B4 /* UninhabitedTypeGuards.swift in Sources */,
9A2D5D06259F8C39005682ED /* Reduce.swift in Sources */,
9AFA491424E9A196003D263C /* Map.swift in Sources */,
9A2D5CFC259F8634005682ED /* UniqueValues.swift in Sources */,
9A2D5C66259F7B47005682ED /* MaterializeAsResult.swift in Sources */,
9AFA491424E9A196003D263C /* Map.swift in Sources */,
9A2D5C8E259F7ED5005682ED /* Dematerialize.swift in Sources */,
9AFA491E24E9A925003D263C /* Filter.swift in Sources */,
9A2D5CDE259F8398005682ED /* Collect.swift in Sources */,
57A4D1BC1BA13D7A00F7D4B1 /* SignalProducer.swift in Sources */,
57A4D1BD1BA13D7A00F7D4B1 /* Atomic.swift in Sources */,
57A4D1BE1BA13D7A00F7D4B1 /* Bag.swift in Sources */,
Expand Down Expand Up @@ -1050,18 +1093,25 @@
A9B315BE1B3940810001CB9C /* Event.swift in Sources */,
A9B315C01B3940810001CB9C /* Scheduler.swift in Sources */,
9A9100E11E0E6E680093E346 /* ValidatingProperty.swift in Sources */,
9A2D5CF1259F85AE005682ED /* SkipRepeats.swift in Sources */,
9A2D5CBA259F8199005682ED /* TakeWhile.swift in Sources */,
A9B315C11B3940810001CB9C /* Action.swift in Sources */,
A9B315C21B3940810001CB9C /* Property.swift in Sources */,
9A2D5C5B259F7B31005682ED /* Materialize.swift in Sources */,
9A090C161DA0309E00EE97CA /* Reactive.swift in Sources */,
A9B315C31B3940810001CB9C /* Signal.swift in Sources */,
9AFA492724E9B15C003D263C /* Operators.swift in Sources */,
9A2D5D0F259F8D1F005682ED /* ScanMap.swift in Sources */,
9A2D5CE7259F852B005682ED /* CombinePrevious.swift in Sources */,
9A67963D1F6059430058C5B4 /* UninhabitedTypeGuards.swift in Sources */,
9A2D5D05259F8C39005682ED /* Reduce.swift in Sources */,
9AFA491324E9A196003D263C /* Map.swift in Sources */,
9A2D5CFB259F8634005682ED /* UniqueValues.swift in Sources */,
9A2D5C65259F7B47005682ED /* MaterializeAsResult.swift in Sources */,
9AFA491324E9A196003D263C /* Map.swift in Sources */,
9A2D5C8D259F7ED5005682ED /* Dematerialize.swift in Sources */,
9AFA491D24E9A925003D263C /* Filter.swift in Sources */,
9A2D5CDD259F8398005682ED /* Collect.swift in Sources */,
A9B315C41B3940810001CB9C /* SignalProducer.swift in Sources */,
A9B315C51B3940810001CB9C /* Atomic.swift in Sources */,
A9B315C61B3940810001CB9C /* Bag.swift in Sources */,
Expand Down Expand Up @@ -1094,18 +1144,25 @@
D0C312D319EF2A5800984962 /* Disposable.swift in Sources */,
9A9100DF1E0E6E620093E346 /* ValidatingProperty.swift in Sources */,
EBCC7DBC1BBF010C00A2AE92 /* Signal.Observer.swift in Sources */,
9A2D5CEF259F85AE005682ED /* SkipRepeats.swift in Sources */,
9A2D5CB8259F8199005682ED /* TakeWhile.swift in Sources */,
D03B4A3D19F4C39A009E02AC /* FoundationExtensions.swift in Sources */,
9A090C141DA0309E00EE97CA /* Reactive.swift in Sources */,
9A2D5C59259F7B31005682ED /* Materialize.swift in Sources */,
D08C54B31A69A2AE00AD8286 /* Signal.swift in Sources */,
D85C652A1C0D84C7005A77AD /* Flatten.swift in Sources */,
9AFA492524E9B15C003D263C /* Operators.swift in Sources */,
9A2D5D0D259F8D1F005682ED /* ScanMap.swift in Sources */,
9A2D5CE5259F852B005682ED /* CombinePrevious.swift in Sources */,
9A67963B1F6056B90058C5B4 /* UninhabitedTypeGuards.swift in Sources */,
9A2D5D03259F8C39005682ED /* Reduce.swift in Sources */,
9AFA491124E9A196003D263C /* Map.swift in Sources */,
9A2D5CF9259F8634005682ED /* UniqueValues.swift in Sources */,
9A2D5C63259F7B47005682ED /* MaterializeAsResult.swift in Sources */,
9AFA491124E9A196003D263C /* Map.swift in Sources */,
9A2D5C8B259F7ED5005682ED /* Dematerialize.swift in Sources */,
9AFA491B24E9A925003D263C /* Filter.swift in Sources */,
9A2D5CDB259F8398005682ED /* Collect.swift in Sources */,
D0C312CF19EF2A5800984962 /* Bag.swift in Sources */,
4A0E10FF1D2A92720065D310 /* Lifetime.swift in Sources */,
D0C312E719EF2A5800984962 /* Scheduler.swift in Sources */,
Expand Down Expand Up @@ -1166,18 +1223,25 @@
D0C312D419EF2A5800984962 /* Disposable.swift in Sources */,
D08C54B91A69A9D100AD8286 /* SignalProducer.swift in Sources */,
9A9100E01E0E6E670093E346 /* ValidatingProperty.swift in Sources */,
9A2D5CF0259F85AE005682ED /* SkipRepeats.swift in Sources */,
9A2D5CB9259F8199005682ED /* TakeWhile.swift in Sources */,
9ABCB1861D2A5B5A00BCA243 /* Deprecations+Removals.swift in Sources */,
EBCC7DBD1BBF01E100A2AE92 /* Signal.Observer.swift in Sources */,
9A2D5C5A259F7B31005682ED /* Materialize.swift in Sources */,
9A090C151DA0309E00EE97CA /* Reactive.swift in Sources */,
D85C652B1C0E70E3005A77AD /* Flatten.swift in Sources */,
9AFA492624E9B15C003D263C /* Operators.swift in Sources */,
9A2D5D0E259F8D1F005682ED /* ScanMap.swift in Sources */,
9A2D5CE6259F852B005682ED /* CombinePrevious.swift in Sources */,
9A67963C1F6059420058C5B4 /* UninhabitedTypeGuards.swift in Sources */,
9A2D5D04259F8C39005682ED /* Reduce.swift in Sources */,
9AFA491224E9A196003D263C /* Map.swift in Sources */,
9A2D5CFA259F8634005682ED /* UniqueValues.swift in Sources */,
9A2D5C64259F7B47005682ED /* MaterializeAsResult.swift in Sources */,
9AFA491224E9A196003D263C /* Map.swift in Sources */,
9A2D5C8C259F7ED5005682ED /* Dematerialize.swift in Sources */,
9AFA491C24E9A925003D263C /* Filter.swift in Sources */,
9A2D5CDC259F8398005682ED /* Collect.swift in Sources */,
4A0E11001D2A92720065D310 /* Lifetime.swift in Sources */,
D08C54BB1A69C54400AD8286 /* Property.swift in Sources */,
D03B4A3E19F4C39A009E02AC /* FoundationExtensions.swift in Sources */,
Expand Down
171 changes: 14 additions & 157 deletions Sources/Event.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,38 +325,6 @@ extension Signal.Event where Value: OptionalProtocol {
}
}

/// A reference type which wraps an array to auxiliate the collection of values
/// for `collect` operator.
private final class CollectState<Value> {
var values: [Value] = []

/// Collects a new value.
func append(_ value: Value) {
values.append(value)
}

/// Check if there are any items remaining.
///
/// - note: This method also checks if there weren't collected any values
/// and, in that case, it means an empty array should be sent as the
/// result of collect.
var isEmpty: Bool {
/// We use capacity being zero to determine if we haven't collected any
/// value since we're keeping the capacity of the array to avoid
/// unnecessary and expensive allocations). This also guarantees
/// retro-compatibility around the original `collect()` operator.
return values.isEmpty && values.capacity > 0
}

/// Removes all values previously collected if any.
func flush() {
// Minor optimization to avoid consecutive allocations. Can
// be useful for sequences of regular or similar size and to
// track if any value was ever collected.
values.removeAll(keepingCapacity: true)
}
}

extension Signal.Event {
internal static var collect: Transformation<[Value], Error> {
return collect { _, _ in false }
Expand All @@ -368,54 +336,14 @@ extension Signal.Event {
}

internal static func collect(_ shouldEmit: @escaping (_ collectedValues: [Value]) -> Bool) -> Transformation<[Value], Error> {
return { action, _ in
let state = CollectState<Value>()

return Signal.Observer { event in
switch event {
case let .value(value):
state.append(value)
if shouldEmit(state.values) {
action(.value(state.values))
state.flush()
}
case .completed:
if !state.isEmpty {
action(.value(state.values))
}
action(.completed)
case let .failed(error):
action(.failed(error))
case .interrupted:
action(.interrupted)
}
}
return { downstream, _ in
Operators.Collect(downstream: downstream, shouldEmit: shouldEmit)
}
}

internal static func collect(_ shouldEmit: @escaping (_ collected: [Value], _ latest: Value) -> Bool) -> Transformation<[Value], Error> {
return { action, _ in
let state = CollectState<Value>()

return Signal.Observer { event in
switch event {
case let .value(value):
if shouldEmit(state.values, value) {
action(.value(state.values))
state.flush()
}
state.append(value)
case .completed:
if !state.isEmpty {
action(.value(state.values))
}
action(.completed)
case let .failed(error):
action(.failed(error))
case .interrupted:
action(.interrupted)
}
}
return { downstream, _ in
Operators.Collect(downstream: downstream, shouldEmit: shouldEmit)
}
}

Expand All @@ -424,83 +352,26 @@ extension Signal.Event {
/// `nil` literal would be materialized as `Optional<Value>.none` instead of `Value`,
/// thus changing the semantic.
internal static func combinePrevious(initial: Value?) -> Transformation<(Value, Value), Error> {
return { action, _ in
var previous = initial

return Signal.Observer { event in
switch event {
case let .value(value):
if let previous = previous {
action(.value((previous, value)))
}
previous = value
case .completed:
action(.completed)
case let .failed(error):
action(.failed(error))
case .interrupted:
action(.interrupted)
}
}
return { downstream, _ in
Operators.CombinePrevious(downstream: downstream, initial: initial)
}
}

internal static func skipRepeats(_ isEquivalent: @escaping (Value, Value) -> Bool) -> Transformation<Value, Error> {
return { action, _ in
var previous: Value?

return Signal.Observer { event in
switch event {
case let .value(value):
if let previous = previous, isEquivalent(previous, value) {
return
}
previous = value
fallthrough
case .completed, .interrupted, .failed:
action(event)
}
}
return { downstream, _ in
Operators.SkipRepeats(downstream: downstream, isEquivalent: isEquivalent)
}
}

internal static func uniqueValues<Identity: Hashable>(_ transform: @escaping (Value) -> Identity) -> Transformation<Value, Error> {
return { action, _ in
var seenValues: Set<Identity> = []

return Signal.Observer { event in
switch event {
case let .value(value):
let identity = transform(value)
let (inserted, _) = seenValues.insert(identity)
if inserted {
fallthrough
}

case .failed, .completed, .interrupted:
action(event)
}
}
return { downstream, _ in
Operators.UniqueValues(downstream: downstream, extract: transform)
}
}

internal static func reduce<U>(into initialResult: U, _ nextPartialResult: @escaping (inout U, Value) -> Void) -> Transformation<U, Error> {
return { action, _ in
var accumulator = initialResult

return Signal.Observer { event in
switch event {
case let .value(value):
nextPartialResult(&accumulator, value)
case .completed:
action(.value(accumulator))
action(.completed)
case .interrupted:
action(.interrupted)
case let .failed(error):
action(.failed(error))
}
}
return { downstream, _ in
Operators.Reduce(downstream: downstream, initial: initialResult, nextPartialResult: nextPartialResult)
}
}

Expand All @@ -520,22 +391,8 @@ extension Signal.Event {
}

internal static func scanMap<State, U>(into initialState: State, _ next: @escaping (inout State, Value) -> U) -> Transformation<U, Error> {
return { action, _ in
var accumulator = initialState

return Signal.Observer { event in
switch event {
case let .value(value):
let output = next(&accumulator, value)
action(.value(output))
case .completed:
action(.completed)
case .interrupted:
action(.interrupted)
case let .failed(error):
action(.failed(error))
}
}
return { downstream, _ in
Operators.ScanMap(downstream: downstream, initial: initialState, next: next)
}
}

Expand Down
Loading