Imports code of the RX library.

21 files changed, 710 insertions, 0 deletions

diff --git a/src/objective-c/RxLibrary/GRXImmediateWriter.h b/src/objective-c/RxLibrary/GRXImmediateWriter.h
new file mode 100644
index 0000000000..568dbe6576
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXImmediateWriter.h
@@ -0,0 +1,40 @@
+#import <Foundation/Foundation.h>
+
+#import "GRXWriter.h"
+
+// Utility to construct GRXWriter instances from values that are immediately available when
+// required. The returned writers all support pausing and early termination.
+//
+// Unless the writeable callback pauses them or stops them early, these writers will do all their
+// interactions with the writeable before the start method returns.
+@interface GRXImmediateWriter : NSObject<GRXWriter>
+
+// Returns a writer that pulls values from the passed NSEnumerator instance and pushes them to
+// its writeable. The NSEnumerator is released when it finishes.
++ (id<GRXWriter>)writerWithEnumerator:(NSEnumerator *)enumerator;
+
+// Returns a writer that pushes to its writeable the successive values returned by the passed
+// block. When the block first returns nil, it is released.
++ (id<GRXWriter>)writerWithValueSupplier:(id (^)())block;
+
+// Returns a writer that iterates over the values of the passed container and pushes them to
+// its writeable. The container is released when the iteration is over.
+//
+// Note that the usual speed gain of NSFastEnumeration over NSEnumerator results from not having to
+// call one method per element. Because GRXWriteable instances accept values one by one, that speed
+// gain doesn't happen here.
++ (id<GRXWriter>)writerWithContainer:(id<NSFastEnumeration>)container;
+
+// Returns a writer that sends the passed value to its writeable and then finishes (releasing the
+// value).
++ (id<GRXWriter>)writerWithValue:(id)value;
+
+// Returns a writer that, as part of its start method, sends the passed error to the writeable
+// (then releasing the error).
++ (id<GRXWriter>)writerWithError:(NSError *)error;
+
+// Returns a writer that, as part of its start method, finishes immediately without sending any
+// values to its writeable.
++ (id<GRXWriter>)emptyWriter;
+
+@end
diff --git a/src/objective-c/RxLibrary/GRXImmediateWriter.m b/src/objective-c/RxLibrary/GRXImmediateWriter.m
new file mode 100644
index 0000000000..ebeb3f458a
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXImmediateWriter.m
@@ -0,0 +1,132 @@
+#import "GRXImmediateWriter.h"
+
+#import "NSEnumerator+GRXUtil.h"
+
+@implementation GRXImmediateWriter {
+  NSEnumerator *_enumerator;
+  NSError *_errorOrNil;
+  id<GRXWriteable> _writeable;
+}
+
+@synthesize state = _state;
+
+- (instancetype) init {
+  return [self initWithEnumerator:nil error:nil]; // results in an empty writer.
+}
+
+// Designated initializer
+- (instancetype)initWithEnumerator:(NSEnumerator *)enumerator error:(NSError *)errorOrNil {
+  if (((self = [super init]))) {
+    _enumerator = enumerator;
+    _errorOrNil = errorOrNil;
+    _state = GRXWriterStateNotStarted;
+  }
+  return self;
+}
+
+#pragma mark Convenience constructors
+
++ (instancetype)writerWithEnumerator:(NSEnumerator *)enumerator error:(NSError *)errorOrNil {
+  return [[self alloc] initWithEnumerator:enumerator error:errorOrNil];
+}
+
++ (id<GRXWriter>)writerWithEnumerator:(NSEnumerator *)enumerator {
+  return [self writerWithEnumerator:enumerator error:nil];
+}
+
++ (id<GRXWriter>)writerWithValueSupplier:(id (^)())block {
+  return [self writerWithEnumerator:[NSEnumerator grx_enumeratorWithValueSupplier:block]];
+}
+
++ (id<GRXWriter>)writerWithContainer:(id<NSFastEnumeration>)container {
+  return [self writerWithEnumerator:[NSEnumerator grx_enumeratorWithContainer:container]];;
+}
+
++ (id<GRXWriter>)writerWithValue:(id)value {
+  if (value) {
+    return [self writerWithEnumerator:[NSEnumerator grx_enumeratorWithSingleValue:value]];
+  } else {
+    return [self emptyWriter];
+  }
+}
+
++ (id<GRXWriter>)writerWithError:(NSError *)error {
+  if (error) {
+    return [self writerWithEnumerator:nil error:error];
+  } else {
+    return [self emptyWriter];
+  }
+}
+
++ (id<GRXWriter>)emptyWriter {
+  static GRXImmediateWriter *emptyWriter;
+  static dispatch_once_t onceToken;
+  dispatch_once(&onceToken, ^{
+    emptyWriter = [self writerWithEnumerator:nil error:nil];
+  });
+  return emptyWriter;
+}
+
+#pragma mark Conformance with GRXWriter
+
+// Most of the complexity in this implementation is the result of supporting pause and resumption of
+// the GRXWriter. It's an important feature for instances of GRXWriter that are backed by a
+// container (which may be huge), or by a NSEnumerator (which may even be infinite).
+
+- (void)writeUntilPausedOrStopped {
+  id value;
+  while (value = [_enumerator nextObject]) {
+    [_writeable didReceiveValue:value];
+    // If the writeable has a reference to us, it might change our state to paused or finished.
+    if (_state == GRXWriterStatePaused || _state == GRXWriterStateFinished) {
+      return;
+    }
+  }
+  [self finishWithError:_errorOrNil];
+}
+
+- (void)startWithWriteable:(id<GRXWriteable>)writeable {
+  _state = GRXWriterStateStarted;
+  _writeable = writeable;
+  [self writeUntilPausedOrStopped];
+}
+
+- (void)finishWithError:(NSError *)errorOrNil {
+  _state = GRXWriterStateFinished;
+  _enumerator = nil;
+  _errorOrNil = nil;
+  id<GRXWriteable> writeable = _writeable;
+  _writeable = nil;
+  [writeable didFinishWithError:errorOrNil];
+}
+
+- (void)setState:(GRXWriterState)newState {
+  // Manual transitions are only allowed from the started or paused states.
+  if (_state == GRXWriterStateNotStarted || _state == GRXWriterStateFinished) {
+    return;
+  }
+
+  switch (newState) {
+    case GRXWriterStateFinished:
+      _state = newState;
+      _enumerator = nil;
+      _errorOrNil = nil;
+      // Per GRXWriter's contract, setting the state to Finished manually
+      // means one doesn't wish the writeable to be messaged anymore.
+      _writeable = nil;
+      return;
+    case GRXWriterStatePaused:
+      _state = newState;
+      return;
+    case GRXWriterStateStarted:
+      if (_state == GRXWriterStatePaused) {
+        _state = newState;
+        [self writeUntilPausedOrStopped];
+      }
+      return;
+    case GRXWriterStateNotStarted:
+      return;
+  }
+}
+
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriteable.h b/src/objective-c/RxLibrary/GRXWriteable.h
new file mode 100644
index 0000000000..bbcdb6a2ba
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriteable.h
@@ -0,0 +1,27 @@
+#import <Foundation/Foundation.h>
+
+// A GRXWriteable is an object to which a sequence of values can be sent. The
+// sequence finishes with an optional error.
+@protocol GRXWriteable <NSObject>
+
+// Push the next value of the sequence to the receiving object.
+// TODO(jcanizales): Name it enumerator:(id<GRXEnumerator>) didProduceValue:(id)?
+- (void)didReceiveValue:(id)value;
+
+// Signal that the sequence is completed, or that an error ocurred. After this
+// message is sent to the instance, neither it nor didReceiveValue: may be
+// called again.
+// TODO(jcanizales): enumerator:(id<GRXEnumerator>) didFinishWithError:(NSError*)?
+- (void)didFinishWithError:(NSError *)errorOrNil;
+@end
+
+typedef void (^GRXValueHandler)(id value);
+typedef void (^GRXCompletionHandler)(NSError *errorOrNil);
+
+// Utility to create objects that conform to the GRXWriteable protocol, from
+// blocks that handle each of the two methods of the protocol.
+@interface GRXWriteable : NSObject<GRXWriteable>
+- (instancetype)initWithValueHandler:(GRXValueHandler)valueHandler
+                   completionHandler:(GRXCompletionHandler)completionHandler
+    NS_DESIGNATED_INITIALIZER;
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriteable.m b/src/objective-c/RxLibrary/GRXWriteable.m
new file mode 100644
index 0000000000..3b4f0811aa
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriteable.m
@@ -0,0 +1,33 @@
+#import "GRXWriteable.h"
+
+@implementation GRXWriteable {
+  GRXValueHandler _valueHandler;
+  GRXCompletionHandler _completionHandler;
+}
+
+- (instancetype)init {
+  return [self initWithValueHandler:nil completionHandler:nil];
+}
+
+// Designated initializer
+- (instancetype)initWithValueHandler:(GRXValueHandler)valueHandler
+                   completionHandler:(GRXCompletionHandler)completionHandler {
+  if ((self = [super init])) {
+    _valueHandler = valueHandler;
+    _completionHandler = completionHandler;
+  }
+  return self;
+}
+
+- (void)didReceiveValue:(id)value {
+  if (_valueHandler) {
+    _valueHandler(value);
+  }
+}
+
+- (void)didFinishWithError:(NSError *)errorOrNil {
+  if (_completionHandler) {
+    _completionHandler(errorOrNil);
+  }
+}
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriter+Immediate.h b/src/objective-c/RxLibrary/GRXWriter+Immediate.h
new file mode 100644
index 0000000000..2d397d7655
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriter+Immediate.h
@@ -0,0 +1,33 @@
+#import "GRXWriter.h"
+
+@interface GRXWriter (Immediate)
+
+// Returns a writer that pulls values from the passed NSEnumerator instance and pushes them to
+// its writeable. The NSEnumerator is released when it finishes.
++ (instancetype)writerWithEnumerator:(NSEnumerator *)enumerator;
+
+// Returns a writer that pushes to its writeable the successive values returned by the passed
+// block. When the block first returns nil, it is released.
++ (instancetype)writerWithValueSupplier:(id (^)())block;
+
+// Returns a writer that iterates over the values of the passed container and pushes them to
+// its writeable. The container is released when the iteration is over.
+//
+// Note that the usual speed gain of NSFastEnumeration over NSEnumerator results from not having to
+// call one method per element. Because GRXWriteable instances accept values one by one, that speed
+// gain doesn't happen here.
++ (instancetype)writerWithContainer:(id<NSFastEnumeration>)container;
+
+// Returns a writer that sends the passed value to its writeable and then finishes (releasing the
+// value).
++ (instancetype)writerWithValue:(id)value;
+
+// Returns a writer that, as part of its start method, sends the passed error to the writeable
+// (then releasing the error).
++ (instancetype)writerWithError:(NSError *)error;
+
+// Returns a writer that, as part of its start method, finishes immediately without sending any
+// values to its writeable.
++ (instancetype)emptyWriter;
+
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriter+Immediate.m b/src/objective-c/RxLibrary/GRXWriter+Immediate.m
new file mode 100644
index 0000000000..841ea8a30f
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriter+Immediate.m
@@ -0,0 +1,31 @@
+#import "GRXWriter+Immediate.h"
+
+#import "GRXImmediateWriter.h"
+
+@implementation GRXWriter (Immediate)
+
++ (instancetype)writerWithEnumerator:(NSEnumerator *)enumerator {
+  return [[self alloc] initWithWriter:[GRXImmediateWriter writerWithEnumerator:enumerator]];
+}
+
++ (instancetype)writerWithValueSupplier:(id (^)())block {
+  return [[self alloc] initWithWriter:[GRXImmediateWriter writerWithValueSupplier:block]];
+}
+
++ (instancetype)writerWithContainer:(id<NSFastEnumeration>)container {
+  return [[self alloc] initWithWriter:[GRXImmediateWriter writerWithContainer:container]];
+}
+
++ (instancetype)writerWithValue:(id)value {
+  return [[self alloc] initWithWriter:[GRXImmediateWriter writerWithValue:value]];
+}
+
++ (instancetype)writerWithError:(NSError *)error {
+  return [[self alloc] initWithWriter:[GRXImmediateWriter writerWithError:error]];
+}
+
++ (instancetype)emptyWriter {
+  return [[self alloc] initWithWriter:[GRXImmediateWriter emptyWriter]];
+}
+
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriter+Transformations.h b/src/objective-c/RxLibrary/GRXWriter+Transformations.h
new file mode 100644
index 0000000000..4c9335b675
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriter+Transformations.h
@@ -0,0 +1,9 @@
+#import "GRXWriter.h"
+
+@interface GRXWriter (Transformations)
+
+// Returns a writer that wraps the receiver, and has all the values the receiver would write
+// transformed by the provided mapping function.
+- (GRXWriter *)map:(id (^)(id value))map;
+
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriter+Transformations.m b/src/objective-c/RxLibrary/GRXWriter+Transformations.m
new file mode 100644
index 0000000000..30e5000afd
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriter+Transformations.m
@@ -0,0 +1,14 @@
+#import "GRXWriter+Transformations.h"
+
+#import "transformations/GRXMappingWriter.h"
+
+@implementation GRXWriter (Transformations)
+
+- (GRXWriter *)map:(id (^)(id))map {
+  if (!map) {
+    return self;
+  }
+  return [[GRXMappingWriter alloc] initWithWriter:self map:map];
+}
+
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriter.h b/src/objective-c/RxLibrary/GRXWriter.h
new file mode 100644
index 0000000000..03b3ee18cd
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriter.h
@@ -0,0 +1,94 @@
+#import <Foundation/Foundation.h>
+
+#import "GRXWriteable.h"
+
+typedef NS_ENUM(NSInteger, GRXWriterState) {
+
+  // The writer has not yet been given a writeable to which it can push its
+  // values. To have an writer transition to the Started state, send it a
+  // startWithWriteable: message.
+  //
+  // An writer's state cannot be manually set to this value.
+  GRXWriterStateNotStarted,
+
+  // The writer might push values to the writeable at any moment.
+  GRXWriterStateStarted,
+
+  // The writer is temporarily paused, and won't send any more values to the
+  // writeable unless its state is set back to Started. The writer might still
+  // transition to the Finished state at any moment, and is allowed to send
+  // didFinishWithError: to its writeable.
+  //
+  // Not all implementations of writer have to support pausing, and thus
+  // trying to set an writer's state to this value might have no effect.
+  GRXWriterStatePaused,
+
+  // The writer has released its writeable and won't interact with it anymore.
+  //
+  // One seldomly wants to set an writer's state to this value, as its
+  // writeable isn't notified with a didFinishWithError: message. Instead, sending
+  // finishWithError: to the writer will make it notify the writeable and then
+  // transition to this state.
+  GRXWriterStateFinished
+};
+
+// An object that conforms to this protocol can produce, on demand, a sequence
+// of values. The sequence may be produced asynchronously, and it may consist of
+// any number of elements, including none or an infinite number.
+//
+// GRXWriter is the active dual of NSEnumerator. The difference between them
+// is thus whether the object plays an active or passive role during usage: A
+// user of NSEnumerator pulls values off it, and passes the values to a writeable.
+// A user of GRXWriter, though, just gives it a writeable, and the
+// GRXWriter instance pushes values to the writeable. This makes this protocol
+// suitable to represent a sequence of future values, as well as collections
+// with internal iteration.
+//
+// An instance of GRXWriter can start producing values after a writeable is
+// passed to it. It can also be commanded to finish the sequence immediately
+// (with an optional error). Finally, it can be asked to pause, but the
+// conforming instance is not required to oblige.
+//
+// Unless otherwise indicated by a conforming class, no messages should be sent
+// concurrently to a GRXWriter. I.e., conforming classes aren't required to
+// be thread-safe.
+@protocol GRXWriter <NSObject>
+
+// This property can be used to query the current state of the writer, which
+// determines how it might currently use its writeable. Some state transitions can
+// be triggered by setting this property to the corresponding value, and that's
+// useful for advanced use cases like pausing an writer. For more details,
+// see the documentation of the enum.
+@property(nonatomic) GRXWriterState state;
+
+// Start sending messages to the writeable. Messages may be sent before the method
+// returns, or they may be sent later in the future. See GRXWriteable.h for the
+// different messages a writeable can receive.
+//
+// If this writer draws its values from an external source (e.g. from the
+// filesystem or from a server), calling this method will commonly trigger side
+// effects (like network connections).
+//
+// This method might only be called on writers in the NotStarted state.
+- (void)startWithWriteable:(id<GRXWriteable>)writeable;
+
+// Send didFinishWithError:errorOrNil immediately to the writeable, and don't send
+// any more messages to it.
+//
+// This method might only be called on writers in the Started or Paused
+// state.
+//
+// TODO(jcanizales): Consider adding some guarantee about the immediacy of that
+// stopping. I know I've relied on it in part of the code that uses this, but
+// can't remember the details in the presence of concurrency.
+- (void)finishWithError:(NSError *)errorOrNil;
+@end
+
+// A "proxy" class that simply forwards values, completion, and errors from its
+// input writer to its writeable.
+// It is useful as a superclass for pipes that act as a transformation of their
+// input writer, and for classes that represent objects with input and
+// output sequences of values, like an RPC.
+@interface GRXWriter : NSObject<GRXWriter>
+- (instancetype)initWithWriter:(id<GRXWriter>)writer NS_DESIGNATED_INITIALIZER;
+@end
diff --git a/src/objective-c/RxLibrary/GRXWriter.m b/src/objective-c/RxLibrary/GRXWriter.m
new file mode 100644
index 0000000000..67d928fed5
--- /dev/null
+++ b/src/objective-c/RxLibrary/GRXWriter.m
@@ -0,0 +1,79 @@
+#import "GRXWriter.h"
+
+@interface GRXWriter () <GRXWriteable>
+@end
+
+@implementation GRXWriter {
+  id<GRXWriter> _writer;
+  id<GRXWriteable> _writeable;
+}
+
+- (instancetype)init {
+  return [self initWithWriter:nil];
+}
+
+// Designated initializer
+- (instancetype)initWithWriter:(id<GRXWriter>)writer {
+  if (!writer) {
+    [NSException raise:NSInvalidArgumentException format:@"writer can't be nil."];
+  }
+  if ((self = [super init])) {
+    _writer = writer;
+  }
+  return self;
+}
+
+// This is used to send a completion or an error to the writeable. It nillifies
+// our reference to it in order to guarantee no more messages are sent to it,
+// and to release it.
+- (void)finishOutputWithError:(NSError *)errorOrNil {
+  id<GRXWriteable> writeable = _writeable;
+  _writeable = nil;
+  [writeable didFinishWithError:errorOrNil];
+}
+
+// This is used to stop the input writer. It nillifies our reference to it
+// to release it.
+- (void)finishInput {
+  id<GRXWriter> writer = _writer;
+  _writer = nil;
+  writer.state = GRXWriterStateFinished;
+}
+
+#pragma mark GRXWriteable implementation
+
+- (void)didReceiveValue:(id)value {
+  [_writeable didReceiveValue:value];
+}
+
+- (void)didFinishWithError:(NSError *)errorOrNil {
+  _writer = nil;
+  [self finishOutputWithError:errorOrNil];
+}
+
+#pragma mark GRXWriter implementation
+
+- (GRXWriterState)state {
+  return _writer ? _writer.state : GRXWriterStateFinished;
+}
+
+- (void)setState:(GRXWriterState)state {
+  if (state == GRXWriterStateFinished) {
+    _writeable = nil;
+    [self finishInput];
+  } else {
+    _writer.state = state;
+  }
+}
+
+- (void)startWithWriteable:(id<GRXWriteable>)writeable {
+  _writeable = writeable;
+  [_writer startWithWriteable:self];
+}
+
+- (void)finishWithError:(NSError *)errorOrNil {
+  [self finishOutputWithError:errorOrNil];
+  [self finishInput];
+}
+
+@end
diff --git a/src/objective-c/RxLibrary/NSEnumerator+GRXUtil.h b/src/objective-c/RxLibrary/NSEnumerator+GRXUtil.h
new file mode 100644
index 0000000000..ecd8f2de79
--- /dev/null
+++ b/src/objective-c/RxLibrary/NSEnumerator+GRXUtil.h
@@ -0,0 +1,18 @@
+#import <Foundation/Foundation.h>
+
+@interface NSEnumerator (GRXUtil)
+
+// Returns a NSEnumerator instance that iterates through the elements of the passed container that
+// supports fast enumeration. Note that this negates the speed benefits of fast enumeration over
+// NSEnumerator. It's only intended for the rare cases when one needs the latter and only has the
+// former, e.g. for iteration that needs to be paused and resumed later.
++ (NSEnumerator *)grx_enumeratorWithContainer:(id<NSFastEnumeration>)container;
+
+// Returns a NSEnumerator instance that provides a single object before finishing. The value is then
+// released.
++ (NSEnumerator *)grx_enumeratorWithSingleValue:(id)value;
+
+// Returns a NSEnumerator instance that delegates the invocations of nextObject to the passed block.
+// When the block first returns nil, it is released.
++ (NSEnumerator *)grx_enumeratorWithValueSupplier:(id (^)())block;
+@end
diff --git a/src/objective-c/RxLibrary/NSEnumerator+GRXUtil.m b/src/objective-c/RxLibrary/NSEnumerator+GRXUtil.m
new file mode 100644
index 0000000000..7da05d13c4
--- /dev/null
+++ b/src/objective-c/RxLibrary/NSEnumerator+GRXUtil.m
@@ -0,0 +1,21 @@
+#import "NSEnumerator+GRXUtil.h"
+
+#import "private/GRXNSBlockEnumerator.h"
+#import "private/GRXNSFastEnumerator.h"
+#import "private/GRXNSScalarEnumerator.h"
+
+@implementation NSEnumerator (GRXUtil)
+
++ (NSEnumerator *)grx_enumeratorWithContainer:(id<NSFastEnumeration>)container {
+  // TODO(jcanizales): Consider checking if container responds to objectEnumerator and return that?
+  return [[GRXNSFastEnumerator alloc] initWithContainer:container];
+}
+
++ (NSEnumerator *)grx_enumeratorWithSingleValue:(id)value {
+  return [[GRXNSScalarEnumerator alloc] initWithValue:value];
+}
+
++ (NSEnumerator *)grx_enumeratorWithValueSupplier:(id (^)())block {
+  return [[GRXNSBlockEnumerator alloc] initWithValueSupplier:block];
+}
+@end
diff --git a/src/objective-c/RxLibrary/README.md b/src/objective-c/RxLibrary/README.md
new file mode 100644
index 0000000000..88e90723b9
--- /dev/null
+++ b/src/objective-c/RxLibrary/README.md
@@ -0,0 +1,8 @@
+This is a generic Reactive Extensions library for Objective-C, created to ease
+the implementation of the gRPC Objective-C runtime.
+
+It has no dependencies on gRPC nor other libraries, and should eventually be
+moved under its own GitHub project.
+
+If you're trying to get started on the library, you might want to first read
+GRXWriter.h and then GRXWriteable.h.
diff --git a/src/objective-c/RxLibrary/private/GRXNSBlockEnumerator.h b/src/objective-c/RxLibrary/private/GRXNSBlockEnumerator.h
new file mode 100644
index 0000000000..0bb1f47764
--- /dev/null
+++ b/src/objective-c/RxLibrary/private/GRXNSBlockEnumerator.h
@@ -0,0 +1,9 @@
+#import <Foundation/Foundation.h>
+
+// Concrete subclass of NSEnumerator that delegates the invocations of nextObject to a block passed
+// on initialization.
+@interface GRXNSBlockEnumerator : NSEnumerator
+// The first time the passed block returns nil, the enumeration will end and the block will be
+// released.
+- (instancetype)initWithValueSupplier:(id (^)())block;
+@end
diff --git a/src/objective-c/RxLibrary/private/GRXNSBlockEnumerator.m b/src/objective-c/RxLibrary/private/GRXNSBlockEnumerator.m
new file mode 100644
index 0000000000..9a53531b12
--- /dev/null
+++ b/src/objective-c/RxLibrary/private/GRXNSBlockEnumerator.m
@@ -0,0 +1,28 @@
+#import "GRXNSBlockEnumerator.h"
+
+@implementation GRXNSBlockEnumerator {
+  id (^_block)();
+}
+
+- (instancetype)init {
+  return [self initWithValueSupplier:nil];
+}
+
+- (instancetype)initWithValueSupplier:(id (^)())block {
+  if ((self = [super init])) {
+    _block = block;
+  }
+  return self;
+}
+
+- (id)nextObject {
+  if (!_block) {
+    return nil;
+  }
+  id value = _block();
+  if (!value) {
+    _block = nil;
+  }
+  return value;
+}
+@end
diff --git a/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.h b/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.h
new file mode 100644
index 0000000000..e5f27b1cc7
--- /dev/null
+++ b/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.h
@@ -0,0 +1,10 @@
+#import <Foundation/Foundation.h>
+
+// This is a bridge to interact through NSEnumerator's interface with objects that only conform to
+// NSFastEnumeration. (There's nothing specifically fast about it - you certainly don't win any
+// speed by using this instead of a NSEnumerator provided by your container).
+@interface GRXNSFastEnumerator : NSEnumerator
+// After the iteration of the container (via the NSFastEnumeration protocol) is over, the container
+// is released. If the container is modified during enumeration, an exception is thrown.
+- (instancetype)initWithContainer:(id<NSFastEnumeration>)container;
+@end
diff --git a/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m b/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m
new file mode 100644
index 0000000000..817ff34d95
--- /dev/null
+++ b/src/objective-c/RxLibrary/private/GRXNSFastEnumerator.m
@@ -0,0 +1,55 @@
+#import "GRXNSFastEnumerator.h"
+
+@implementation GRXNSFastEnumerator {
+  id<NSFastEnumeration> _container;
+  NSFastEnumerationState _state;
+  // Number of elements of the container currently in the _state.itemsPtr array.
+  NSUInteger _count;
+  // The index of the next object to return from the _state.itemsPtr array.
+  NSUInteger _index;
+  // A "buffer of one element," for the containers that enumerate their elements one by one. Those
+  // will set _state.itemsPtr to point to this.
+  // The NSFastEnumeration protocol requires it to be __unsafe_unretained, but that's alright
+  // because the only use we'd make of its value is to return it immediately as the result of
+  // nextObject.
+  __unsafe_unretained id _bufferValue;
+  // Neither NSEnumerator nor NSFastEnumeration instances are required to work correctly when the
+  // underlying container is mutated during iteration. The expectation is that an exception is
+  // thrown when that happens. So we check for mutations.
+  unsigned long _mutationFlag;
+  BOOL _mutationFlagIsSet;
+}
+
+- (instancetype)init {
+  return [self initWithContainer:nil];
+}
+
+// Designated initializer.
+- (instancetype)initWithContainer:(id<NSFastEnumeration>)container {
+  NSAssert(container, @"container can't be nil");
+  if ((self = [super init])) {
+    _container = container;
+  }
+  return self;
+}
+
+- (id)nextObject {
+  if (_index == _count) {
+    _index = 0;
+    _count = [_container countByEnumeratingWithState:&_state objects:&_bufferValue count:1];
+    if (_count == 0) {
+      // Enumeration is over.
+      _container = nil;
+      return nil;
+    }
+    if (_mutationFlagIsSet) {
+      NSAssert(_mutationFlag == *(_state.mutationsPtr),
+               @"container was mutated while being enumerated");
+    } else {
+      _mutationFlag = *(_state.mutationsPtr);
+      _mutationFlagIsSet = YES;
+    }
+  }
+  return _state.itemsPtr[_index++];
+}
+@end
diff --git a/src/objective-c/RxLibrary/private/GRXNSScalarEnumerator.h b/src/objective-c/RxLibrary/private/GRXNSScalarEnumerator.h
new file mode 100644
index 0000000000..1130f52897
--- /dev/null
+++ b/src/objective-c/RxLibrary/private/GRXNSScalarEnumerator.h
@@ -0,0 +1,8 @@
+#import <Foundation/Foundation.h>
+
+// Concrete subclass of NSEnumerator whose instances return a single object before finishing.
+@interface GRXNSScalarEnumerator : NSEnumerator
+// Param value: the single object this instance will produce. After the first invocation of
+// nextObject, the value is released.
+- (instancetype)initWithValue:(id)value;
+@end
diff --git a/src/objective-c/RxLibrary/private/GRXNSScalarEnumerator.m b/src/objective-c/RxLibrary/private/GRXNSScalarEnumerator.m
new file mode 100644
index 0000000000..b2a1afd00e
--- /dev/null
+++ b/src/objective-c/RxLibrary/private/GRXNSScalarEnumerator.m
@@ -0,0 +1,24 @@
+#import "GRXNSScalarEnumerator.h"
+
+@implementation GRXNSScalarEnumerator {
+  id _value;
+}
+
+- (instancetype)init {
+  return [self initWithValue:nil];
+}
+
+// Designated initializer.
+- (instancetype)initWithValue:(id)value {
+  if ((self = [super init])) {
+    _value = value;
+  }
+  return self;
+}
+
+- (id)nextObject {
+  id value = _value;
+  _value = nil;
+  return value;
+}
+@end
diff --git a/src/objective-c/RxLibrary/transformations/GRXMappingWriter.h b/src/objective-c/RxLibrary/transformations/GRXMappingWriter.h
new file mode 100644
index 0000000000..13640c5bd6
--- /dev/null
+++ b/src/objective-c/RxLibrary/transformations/GRXMappingWriter.h
@@ -0,0 +1,7 @@
+#import "GRXWriter.h"
+
+// A "proxy" writer that transforms all the values of its input writer by using a mapping function.
+@interface GRXMappingWriter : GRXWriter
+- (instancetype)initWithWriter:(id<GRXWriter>)writer map:(id (^)(id value))map
+    NS_DESIGNATED_INITIALIZER;
+@end
diff --git a/src/objective-c/RxLibrary/transformations/GRXMappingWriter.m b/src/objective-c/RxLibrary/transformations/GRXMappingWriter.m
new file mode 100644
index 0000000000..3aa2a2503a
--- /dev/null
+++ b/src/objective-c/RxLibrary/transformations/GRXMappingWriter.m
@@ -0,0 +1,30 @@
+#import "GRXMappingWriter.h"
+
+static id (^kIdentity)(id value) = ^id(id value) {
+  return value;
+};
+
+@interface GRXWriter () <GRXWriteable>
+@end
+
+@implementation GRXMappingWriter {
+  id (^_map)(id value);
+}
+
+- (instancetype)initWithWriter:(id<GRXWriter>)writer {
+  return [self initWithWriter:writer map:nil];
+}
+
+// Designated initializer
+- (instancetype)initWithWriter:(id<GRXWriter>)writer map:(id (^)(id value))map {
+  if ((self = [super initWithWriter:writer])) {
+    _map = map ?: kIdentity;
+  }
+  return self;
+}
+
+// Override
+- (void)didReceiveValue:(id)value {
+  [super didReceiveValue:_map(value)];
+}
+@end