[WIP] Automatic backpressure for throttling producers

**:warning: NOT READY TO MERGE :warning:**

**TLDR:** Signals could support backpressure, where the rate at which events are produced is automatically throttled to the rate at which they can be consumed. Should they?

----

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<T>` 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<T>` 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<RACSubscriber> 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<RACSubscriber> 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:^(id<RACSubscriber> subscriber, RACCompoundDisposable *disposable) {
	GTEnumerator *enumerator = [[GTEnumerator alloc] initWithRepository:self.repository error:NULL];
 
	return ^{
		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];
		}
	};
}];
```

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, unbounded buffering) which describe how production should be allowed to get ahead of consumption when desirable.

My gut tells me that it'd also be possible to remove `RACSequence` and `RACStream` (collapsing them into `RACSignal`) if we do this, but we can talk about _that_ later. :wink:

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

**To do:**

 - [ ] ~~Should `-delay:` apply backpressure?~~ It could be useful, but it would also effectively double the delay, since signal values would not be produced until consumed.
 - [ ] ~~Should `-throttle:` apply backpressure?~~ It's arguably unnecessary, since throttling already reduces production to a level that the downstream consumer can handle. Backpressure would also emphasize _earlier_ values over the _later_ ones that `-throttle:` currently favors.
 - [ ] Support different policies (unbounded, bounded to N, drop latest, drop earliest) for backpressure, defaulting to unbounded.