ReactiveCocoa/ReactiveCocoa — pull request #808 by jspahrsummers (2013-09-22 07:14:57+00:00)
Title: [WIP] Replace +createSignal: with +generator:

**:warning: NOT READY TO MERGE :warning:**

At [Strange Loop 2013](https://thestrangeloop.com), Rich Hickey gave a talk about Clojure's [core.async](http://clojure.github.io/core.async) library, which presents a [CSP](http://en.wikipedia.org/wiki/Communicating_sequential_processes) abstraction known as "channels," similar to [the type by the same name in Go](http://golang.org/doc/effective_go.html#channels) and [TBQueue](http://hackage.haskell.org/packages/archive/stm/2.4.2/doc/html/Control-Concurrent-STM-TBQueue.html) in Haskell.

Among other things, channels support **backpressure**, which means that downstream consumers can throttle upstream producers to the rate that they're capable of handling — preventing production from outstripping consumption, which would normally result in dropped events and/or excessive memory usage.

Without going into too much detail, this is possible because channels are _pull-driven_ (i.e., you pull values only when you're ready for them), and putting a value into an unbuffered channel will "park" the producer until a consumer is ready to receive it. In some ways, this is similar to what the [Conduit](http://hackage.haskell.org/package/conduit) library offers in Haskell.

In ReactiveCocoa 2.x and earlier, this can sorta be accomplished using `RACSequence` (roughly equivalent to .NET's `IEnumerable` or Haskell's `List` type), but it's not very simple or easy. Sequences are much more suitable for enumeration and list processing than a queue abstraction.

But it occurred to me that we can offer backpressure in `RACSignal` (`IObservable` in .NET), without significantly disrupting the interface, by adding an alternative to `+createSignal:` (`Observable.Create`) that generates values incrementally instead of doing _all_ of the work up front.

This will make much more sense with an example. Let's use some [ObjectiveGit](https://github.com/libgit2/objective-git) pseudo-code to enumerate all of the commits in a repository's history.

First, the push-driven version:

```objc
[RACSignal createSignal:^ RACDisposable * (id subscriber) {
GTEnumerator *enumerator = [[GTEnumerator alloc] initWithRepository:self.repository error:NULL];

while (YES) {
NSError *error = nil;
BOOL success = YES;

GTCommit *commit = [enumerator nextObjectWithSuccess:&success error:&error];
if (commit == nil) {
if (success) {
[subscriber sendCompleted];
} else {
[subscriber sendError:error];
}

break;
}

[subscriber sendNext:commit];
}

return nil;
}];
```

Of course, this is impossible to cancel, since no disposable can be returned until the synchronous work has completed. To actually support cancelation, we'll need to dispatch the work asynchronously, and make sure to check a disposable frequently:

```objc
[RACSignal createSignal:^ RACDisposable * (id subscriber) {
GTEnumerator *enumerator = [[GTEnumerator alloc] initWithRepository:self.repository error:NULL];

RACSerialDisposable *serialDisposable = [[RACSerialDisposable alloc] init];
serialDisposable.disposable = [[RACScheduler scheduler] schedule:^{
while (!serialDisposable.disposed) {
NSError *error = nil;
BOOL success = YES;

GTCommit *commit = [enumerator nextObjectWithSuccess:&success error:&error];
if (commit == nil) {
if (success) {
[subscriber sendCompleted];
} else {
[subscriber sendError:error];
}

break;
}

[subscriber sendNext:commit];
}
}];

return serialDisposable;
}];
```

Compare the above to an incremental generator:

```objc
[RACSignal generator:^{
GTEnumerator *enumerator = [[GTEnumerator alloc] initWithRepository:self.repository error:NULL];

return ^ RACDisposable * (id subscriber) {
NSError *error = nil;
BOOL success = YES;
GTCommit *commit = [enumerator nextObjectWithSuccess:&success error:&error];

if (commit != nil) {
[subscriber sendNext:commit];
} else if (success) {
[subscriber sendCompleted];
} else {
[subscriber sendError:error];
}

return nil;
};
}];
```

The result is very similar to the first version, but without a messy loop, and including automatic support for cancelation. Disposal can be handled by RAC itself, since there's a natural window for it between each invocation of the inner "step" block.

Of course, the biggest win is that the step block doesn't need to be invoked until the consumer (subscriber) is ready for more events. In the most basic implementation, we invoke the step block only at the moment that the subscriber is ready, but — just like core.async — we could also add various policies (e.g., dropping latest, dropping earliest, buffering up to N and then parking) that allow production to get _partly_ ahead of consumption when desirable.

Without thinking too hard about the implementation details, do you guys like this idea? Any thoughts on the abstraction, or potential problems you see with it?

@jonsterling @paulcbetts @mattpodwysocki