Port `groupBy` from REX #200

iv-mexx · 2016-12-30T10:52:32Z

As per #197, this is a Port of groupBy from REX.

I've only changed it to use Atomic instead of the NSRecursiveLock

andersio · 2016-12-30T12:39:43Z

Sources/SignalProducer.swift

+							let producer = SignalProducer(signal).replayLazily(upTo: Int.max)
+
+							// Start the buffering immediately.
+							producer.start()


This can be disposed.

andersio · 2016-12-30T12:46:48Z

Sources/SignalProducer.swift

+					let key = grouping(value)
+					var group: Signal<Value, Error>.Observer?
+					groups.modify {
+						group = $0[key]


[Nit] For multi-line modifications, it'd be better to use named parameter.

andersio · 2016-12-30T12:47:46Z

Sources/SignalProducer.swift

+				switch event {
+				case let .value(value):
+					let key = grouping(value)
+					var group: Signal<Value, Error>.Observer?


You can dodge the need of an optional by using:

let group: Signal<Value, Error>.Observer = groups.modify { groups in

nice, didn't know about this!

andersio · 2016-12-30T12:48:42Z

Sources/SignalProducer.swift

+						}
+					}
+					group!.send(value: value)					
+				case let .failed(error):


[Nit] Blank line before the pattern.

andersio · 2016-12-30T12:50:35Z

Sources/SignalProducer.swift

+		return SignalProducer<(Key, SignalProducer<Value, Error>), Error> { observer, disposable in
+			let groups = Atomic<[Key: Signal<Value, Error>.Observer]>([:])
+
+			self.start { event in


The start disposable should be added to the producer disposable.

andersio · 2016-12-30T12:51:53Z

Sources/SignalProducer.swift

+	///            to the group to which the value belongs to (as determined by the key)
+	public func groupBy<Key: Hashable>(_ grouping: @escaping (Value) -> Key) -> SignalProducer<(Key, SignalProducer<Value, Error>), Error> {
+		return SignalProducer<(Key, SignalProducer<Value, Error>), Error> { observer, disposable in
+			let groups = Atomic<[Key: Signal<Value, Error>.Observer]>([:])


It doesn't seem necessary to specialise it. Is it a workaround to the compiler's weirdness or something?

I'm afraid I don't know what exactly you mean.

I mean if you really need to write SignalProducer<(Key, SignalProducer<Value, Error>), Error> when the type parameters can be inferred.

Oh that line. Yeah that works without explicit annotations 👍

andersio · 2016-12-30T12:53:12Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

@@ -1666,5 +1666,47 @@ class SignalProducerLiftingSpec: QuickSpec {
 				expect(latestValues?.1) == 2
 			}
 		}
+
+		describe("groupBy") {
+			let (signal, observer) = Signal<Int, NoError>.pipe()


Please wrap test cases in it.

andersio · 2016-12-30T12:53:58Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

+
+			disposable += producer
+				.groupBy { $0 % 2 == 0 }
+				.start(Observer(value: { key, group in


Seems like start that takes an event action would be nicer here.

alright! 👍

andersio · 2016-12-30T12:55:26Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

+			disposable.dispose()
+
+			observer.send(value: 1)
+			expect(interrupted) == true


These should be checked immediately after the disposal.

andersio · 2016-12-30T12:57:54Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

@@ -1666,5 +1666,47 @@ class SignalProducerLiftingSpec: QuickSpec {
 				expect(latestValues?.1) == 2
 			}
 		}
+
+		describe("groupBy") {


It would be great if there are specialized test cases than one giant one. At least this single test case apparently did not catch the undisposed upstreams, caused by this line.

Say we could use a few test cases on how the inner producers behaves with regard to an outer terminal event, and keep this test case focusing on values.

yeah right, i've split this up into 2 separated cases and extended them to check the lifetime of the inner producers - now the cases catch the disposal issue you've found!

Perhaps one per terminal event, hmm?

Sure I can do that.
Should we check for all events that only this event is sent (e.g. when sending completed, also check that interrupted was not received), or can we trust that when completed is received, interrupted can not be received?

I think asserting just the specific event is enough. The Signal contract guarantees only one would ever be sent. If multiple terminal events are sent, this would be a fault in the Signal basics and should have been caught somewhere else.

andersio · 2016-12-30T16:47:23Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

+				expect(evens) == [2]
+				expect(odds) == [1, 3]
+			}
+			it("should terminate correctly on disposal") {


[Nit] A blank line please. 🙏

andersio · 2016-12-30T16:48:30Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

+						} else {
+							group.startWithValues { odds.append($0)}
+						}
+				}


[Nit] This close bracket could be aligned better.

andersio · 2016-12-30T16:50:44Z

Sources/SignalProducer.swift

+	/// - returns: A producer of producers amits one producer for each group and forwards
+	///            each value from the original producer to the inner producer corresponding 
+	///            to the group to which the value belongs to (as determined by the key)
+	public func groupBy<Key: Hashable>(_ grouping: @escaping (Value) -> Key) -> SignalProducer<(Key, SignalProducer<Value, Error>), Error> {


By the API convention though, it should be group(by:). Would the parameter be better named classifier or something?

I've tried that for a short moment, but on the call site that would then look like producer.group { ... } instead of producer.groupBy { ... } when using trailing closure syntax.
Not sure if thats a problem though...

We already have skip(while:) and take(while:) working in this way.

Alright! 👍

iv-mexx · 2016-12-30T17:13:54Z

@andersio all done 😃

andersio · 2016-12-30T17:19:14Z

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

+			expect(oddCompleted) == true
+		}
+
+		it("should terminate correctly receiving an error event") {


s/error/failed/

iv-mexx · 2017-01-23T22:28:06Z

@mdiep ping

mdiep · 2017-01-23T22:48:13Z

Oops! Sorry @iv-mexx. I put this back on my todo list and will get to it this week.

mdiep · 2017-01-29T19:21:11Z

Sources/SignalProducer.swift

+							return group
+						} else {
+							let (signal, innerObserver) = Signal<Value, Error>.pipe()
+							let producer = SignalProducer(signal).replayLazily(upTo: Int.max)


Is replaying values an essential part of group(by:)? This is my primary hesitation. In general, we try to discourage the use of replaying (which is why buffer was removed).

(You could just as easily send the Signal and not send values on the Signal until it's been sent—giving observers a chance to observe the signal.)

If you feel that it is, I'd love to see a larger code sample that demonstrates why.

I think so, otherwise consumers could miss events without realizing it. Hence why I erred on the side of buffering. I could be convinced otherwise, but I'm not sure how to best communicate those semantics at the operator level.

I think returning Signals instead of SignalProducers communicates the semantics.

But I think seeing an actual example of this operator will make it easier to see whether that's a viable change.

I've posted an example how we use the groupBy in a project in the Issue: #197 (comment)

If I'm reading this correctly, RxJS is using Hot Observables for the groups (via refCount)

As much as I agree with @mdiep about the semantics, in my experience having buffering in this operator (specifically RxJava's Observable#groupBy) has been valuable .

Perhaps should the buffering be an option?

@mdiep

(You could just as easily send the Signal and not send values on the Signal until it's been sent—giving observers a chance to observe the signal.)

While this is the ideal case, it seems the buffering is essential, say when using with flatten(.concat), which can delay the establishment of the observation.

Edit: I'd suggest to annotate it with warnings for the potential of indefinite buffering, and perhaps offer a non-buffering variant.

Edit: If we have to find a middle ground, perhaps the emitted inner producers should buffer until & only replay values at the first time it started (future work: a configurable #?). Subsequent starts would be invalid and emit interrupted. This should satisfy the common use cases, e.g. @iv-mexx's example, which do not retain and repeatedly start the inner producers.

neilpa · 2017-01-30T21:17:58Z

Sources/SignalProducer.swift

+	/// - parameters:
+	///   - grouping: a closure that determines the grouping key for a given value
+	///
+	/// - returns: A producer of producers amits one producer for each group and forwards


typo amits

mdiep · 2017-05-08T13:29:36Z

I've been thinking about this quite a bit. The replaying of values makes me pretty uncomfortable, so I've felt a bit stuck and unsure how to resolve this.

But after thinking about it, I no longer think that groupBy makes sense as a built-in operator. 😞 It requires semantics that we've said we'd like to discourage (for good reason, I think). And the example can be resolved by introducing a class, which implies the correct semantics.

class BluetoothManager {
  private let allBeaconsSignal: Signal<CLBeacon, NoError>

  private var _beacons: [(Int, Int): (Signal<_, _>, Signal<_, _>.Observer)]
  public let beacons: Signal<(CLBeacon, Signal<_, _>), NoError>
  private let observer: Signal<(CLBeacon, Signal<_, _>), NoError>.Observer

  init() {
     // I'm not exactly sure how this should all work. I was working off the given example code,
     // which is tightly focused on the groupBy. But this will hopefully communicate the idea,
     // which is that the class is responsible for grouping values, notifying interested parties
     // that new values have come in and giving them an opportunity to subscribe to events.
     let (beacons, observer) = Signal<(CLBeacon, Signal), NoError>.pipe()
     allBeaconsSignal.observeValues { beacon in
       let identifier = (beacon.major, beacon.minor)
       if _beacons[identifier] == nil {
         let (signal, observer) = Signal<_, _>.pipe()
         self.observer.send(value: (beacon, signal))
       }
     }
  }
}

But if the rest of the core team feels that this does make sense to add, they can definitely override me. 😄

andersio · 2020-05-22T17:45:50Z

In lieu of the rise of unidirectional data flow in the Swift community, perhaps the class of problems intended for groupBy is better solved that way.

Closing because of inactivity.

andersio reviewed Dec 30, 2016

View reviewed changes

andersio requested changes Dec 30, 2016

View reviewed changes

iv-mexx force-pushed the groupBy branch from 55ec62a to 033a00e Compare December 30, 2016 17:12

andersio reviewed Dec 30, 2016

View reviewed changes

andersio approved these changes Dec 30, 2016

View reviewed changes

andersio requested a review from mdiep December 30, 2016 17:22

andersio added the enhancement label Jan 15, 2017

mdiep reviewed Jan 29, 2017

View reviewed changes

neilpa reviewed Jan 30, 2017

View reviewed changes

iv-mexx added 7 commits February 16, 2017 17:33

Add groupBy operator from REX

9b8175b

Use Atomic for groupBy

918e119

Extend test to catch disposal misbehavior

a575411

Fix disposal misbehavior of inner producer

679f800

Dispose group producers immediately

dbaa6cf

Review changes to groupBy

0f0f48b

Implement test cases for all event types for groupBy

5bea41e

iv-mexx added 4 commits February 16, 2017 17:33

Change `groupBy(_:) to group(by:)

d53d396

Remove explicit type annotation that can be inferred

2587a6c

Realign braces

f4ae595

Fix naming for testcase

56ee8f6

iv-mexx force-pushed the groupBy branch from 446ad6a to 56ee8f6 Compare February 16, 2017 16:34

fix typo in documentation

1d52724

andersio added additive and removed enhancement labels Apr 15, 2017

andersio modified the milestone: 2.5/3.0 Jul 16, 2017

andersio removed this from the Next Minor milestone Nov 11, 2017

andersio closed this May 22, 2020

Port groupBy from REX #200

Port groupBy from REX #200

Conversation

iv-mexx commented Dec 30, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

iv-mexx Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Dec 30, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iv-mexx commented Dec 30, 2016

Choose a reason for hiding this comment

iv-mexx commented Jan 23, 2017

mdiep commented Jan 23, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iv-mexx Feb 16, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andersio Feb 22, 2017 • edited Loading

Choose a reason for hiding this comment

andersio Mar 28, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdiep commented May 8, 2017

andersio commented May 22, 2020

Port `groupBy` from REX #200

Port `groupBy` from REX #200

andersio Dec 30, 2016 •

edited

Loading

andersio Dec 30, 2016 •

edited

Loading

andersio Dec 30, 2016 •

edited

Loading

andersio Dec 30, 2016 •

edited

Loading

iv-mexx Dec 30, 2016 •

edited

Loading

andersio Dec 30, 2016 •

edited

Loading

andersio Dec 30, 2016 •

edited

Loading

iv-mexx Feb 16, 2017 •

edited

Loading

andersio Feb 22, 2017 •

edited

Loading

andersio Mar 28, 2017 •

edited

Loading