Rx with ObjectiveC using NSInvocation

RxArraySource.h

#import "RxObservable.h"

@interface RxArraySource : RxObservable

+ (instancetype)createWithElements: (NSArray *)array;
- (void)subscribe:(NSInvocation *)invocation;

@end

RxArraySource.m

#import "RxArraySource.h"

@interface RxArraySource ()

@property (nonatomic, copy) NSArray *elements;

@end

@implementation RxArraySource

+ (instancetype)createWithElements:(NSArray *)elements;
{
  RxArraySource *instance = [RxArraySource new];
  instance.elements = elements;
  return instance;
}

- (void)subscribe:(NSInvocation *)invocation;
{
  for (id element in _elements) {
    [invocation setArgument:&element atIndex:2];
    [invocation invoke];
  }
}

@end

RxDriver.h

@interface RxDriver : NSObject

- (void)run;

@end

RxDriver.m

#import "RxDriver.h"
#import "NSInvocation+RxConvenience.h"
#import "RxObservable.h"

@implementation RxDriver

- (void)print:(NSNumber *)num
{
  NSLog(@"%@", num);
}

- (NSNumber *)tenTimesOfNumber:(NSNumber *)num
{
  return [NSNumber numberWithInteger:[num integerValue] * 10];
}

- (void)run
{

  NSInvocation *verbose = [NSInvocation invocationWithMethodSignature:[self methodSignatureForSelector:@selector(print:)]];
  [verbose setTarget:self];
  [verbose setSelector:@selector(print:)];

  RxObservable *stream = [RxObservable observableFromArray:@[@1, @2, @3]];

  // Simple subscribe
  NSInvocation *printInvoc = [NSInvocation invocationWithTarget:self
                                                       selector:@selector(print:)];
  [stream subscribe:printInvoc];


  // Square
  NSInvocation *mapInvoc = [NSInvocation invocationWithTarget:self
                                                     selector:@selector(tenTimesOfNumber:)];
  [[stream mapWithInvocation:mapInvoc] subscribe:printInvoc];
}

@end

RxMapOperator.h

#import "RxObservable.h"

@interface RxMapOperator : RxObservable

+ (instancetype) operatorFromStream:(RxObservable *)upstream
                     withInvocation:(NSInvocation *)invoc;

@end

RxMapOperator.m

#import "RxMapOperator.h"
#import "NSInvocation+RxConvenience.h"

@interface RxMapOperator ()
@property (nonatomic, strong) RxObservable *stream;
@property (nonatomic, strong) NSInvocation *invocUp;
@property (nonatomic, strong) NSInvocation *invocDown;
@end

@implementation RxMapOperator

+ (instancetype) operatorFromStream:(RxObservable *)stream
                     withInvocation:(NSInvocation *)invoc;
{
  RxMapOperator *instance = [RxMapOperator new];
  instance.stream = stream;
  instance.invocUp = invoc;
  return instance;
}

- (void)transform:(id)arg
{
  [self.invocUp setArgument:&arg atIndex:2];
  [self.invocUp invoke];

  id ret;
  [self.invocUp getReturnValue:&ret];

  [self.invocDown setArgument:&ret atIndex:2];
  [self.invocDown invoke];
}

- (void)subscribe:(NSInvocation *)invocation
{
  self.invocDown = invocation;
  [self.stream subscribe:[NSInvocation invocationWithTarget:self selector:@selector(transform:)]];
}

@end

RxObservable.h

@interface RxObservable : NSObject

+ (instancetype)observableFromArray:(NSArray *)array;
- (instancetype)mapWithInvocation:(NSInvocation *)invo;

- (void)subscribe:(NSInvocation *)invocation;

@end

RxObservable.m

#import "RxObservable.h"
#import "RxArraySource.h"
#import "RxMapOperator.h"

@implementation RxObservable

+ (instancetype)observableFromArray:(NSArray *)array;
{
  return [RxArraySource createWithElements:array];
}

- (instancetype)mapWithInvocation:(NSInvocation *)invo
{
  return [RxMapOperator operatorFromStream:self withInvocation:invo];
}

- (void)subscribe:(NSInvocation *)invocation;
{}

@end