Understanding RACSignal Subscription, Multicast, and Replay in ReactiveCocoa

ReactiveCocoa is an implementation of Functional Reactive Programming (FRP) in Objective‑C and is widely used in Meituan projects. This article does not repeat the basic usage of ReactiveCocoa; instead it focuses on the implementation principles of RACSignal. Readers are expected to be familiar with the basic usage of RACSignal before reading.

The article consists of two parts: the first part analyses the subscription process of RACSignal, and the second part dives deeper into two often‑confusing operations – multicast and replay. For clarity, only the next event is discussed (error and completed are omitted). The analysis is based on ReactiveCocoa 2.x.

Common Usage of RACSignal

-(RACSignal *)signInSignal {
  // part 1: obtain a signal via createSignal
  return [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
    [self.signInService
     signInWithUsername:self.usernameTextField.text
     password:self.passwordTextField.text
     complete:^(BOOL success) {
       // part 3: inside didSubscribe, send next value
       [subscriber sendNext:@(success)];
       [subscriber sendCompleted];
     }];
    return nil;
  }];
}

// part 2: subscribeNext to get a subscriber and start subscription
[[self signInSignal] subscribeNext:^(id x) {
    NSLog(@"Sign in result: %@", x);
}];

Summary of the Subscription Process

The subscription process of RACSignal can be divided into three steps:

Obtain a signal via [RACSignal createSignal] .

Obtain a subscriber via [signal subscribeNext:] and perform the subscription.

Enter didSubscribe and execute the next block with [subscriber sendNext:] .

Step 1: [RACSignal createSignal] to Obtain a Signal

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    return [RACDynamicSignal createSignal:didSubscribe];
}

+ (RACSignal *)createSignal:(RACDisposable * (^)(id<RACSubscriber> subscriber))didSubscribe {
    RACDynamicSignal *signal = [[self alloc] init];
    signal->_didSubscribe = [didSubscribe copy];
    return [signal setNameWithFormat:@"+createSignal:"];
}

[RACSignal createSignal]

calls the subclass RACDynamicSignal 's createSignal, which returns a signal and stores the didSubscribe block inside the signal.

Step 2: [signal subscribeNext:] to Obtain a Subscriber and Perform Subscription

- (RACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
    RACSubscriber *o = [RACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
    return [self subscribe:o];
}

+ (instancetype)subscriberWithNext:(void (^)(id x))next error:(void (^)(NSError *error))error completed:(void (^)(void))completed {
    RACSubscriber *subscriber = [[self alloc] init];
    subscriber->_next = [next copy];
    subscriber->_error = [error copy];
    subscriber->_completed = [completed copy];
    return subscriber;
}

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
    if (self.didSubscribe != NULL) {
        RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
            RACDisposable *innerDisposable = self.didSubscribe(subscriber);
            [disposable addDisposable:innerDisposable];
        }];
        [disposable addDisposable:schedulingDisposable];
    }
    return disposable;
}

[signal subscribeNext] first creates a RACSubscriber that stores the next , error , and completed blocks.

Because the signal is actually a RACDynamicSignal , the subscribe method invokes the previously saved didSubscribe block, passing the subscriber created in step 1 as an argument.

Step 3: Enter didSubscribe and Execute the next Block via [subscriber sendNext:]

- (void)sendNext:(id)value {
    @synchronized (self) {
        void (^nextBlock)(id) = [self.next copy];
        if (nextBlock == nil) return;
        nextBlock(value);
    }
}

Whenever [subscriber sendNext:] is called, the stored next block is invoked immediately.

Review of the Signal Subscription Process

From the three steps we can see:

By calling createSignal and subscribeNext , we declare how to handle values when they arrive.

After the asynchronous work in the didSubscribe block finishes, the subscriber automatically handles the emitted values.

Understanding the subscription process allows us to discuss two easily confused operations: multicast and replay.

Why Understand Multicast and Replay

- (RACMulticastConnection *)publish;
- (RACMulticastConnection *)multicast:(RACSubject *)subject;
- (RACSignal *)replay;
- (RACSignal *)replayLast;
- (RACSignal *)replayLazily;

These five operators differ in subtle ways; using the wrong one can easily cause bugs.

Misunderstanding them may lead to side‑effects being executed multiple times, resulting in program errors.

Application Scenarios of Multicast & Replay

"Side effects occur for each subscription by default, but there are certain situations where side effects should only occur once – for example, a network request typically should not be repeated when a new subscriber is added."

// Example from ReactiveCocoa source documentation
// This signal starts a new request on each subscription.
RACSignal *networkRequest = [RACSignal createSignal:^(id<RACSubscriber> subscriber) {
    AFHTTPRequestOperation *operation = [client HTTPRequestOperationWithRequest:request
        success:^(AFHTTPRequestOperation *operation, id response) {
            [subscriber sendNext:response];
            [subscriber sendCompleted];
        }
        failure:^(AFHTTPRequestOperation *operation, NSError *error) {
            [subscriber sendError:error];
        }];
    [client enqueueHTTPRequestOperation:operation];
    return [RACDisposable disposableWithBlock:^{
        [operation cancel];
    }];
}];

// Starts a single request, no matter how many subscriptions `connection.signal` gets.
RACMulticastConnection *connection = [networkRequest multicast:[RACReplaySubject subject]];
[connection connect];

[connection.signal subscribeNext:^(id response) {
    NSLog(@"subscriber one: %@", response);
}];

[connection.signal subscribeNext:^(id response) {
    NSLog(@"subscriber two: %@", response);
}];

If we do not use RACMulticastConnection , each subscription would trigger a separate network request.

After applying multicast , we subscribe to connection.signal instead of the original networkRequest . This is the key to ensuring that side effects happen only once.

Multicast Principle Analysis

replay

is just a special case of multicast. The whole multicast process can be split into two steps.

Multicast Mechanism – Part 1

- (id)initWithSourceSignal:(RACSignal *)source subject:(RACSubject *)subject {
    NSCParameterAssert(source != nil);
    NSCParameterAssert(subject != nil);
    self = [super init];
    if (self == nil) return nil;
    _sourceSignal = source;
    _serialDisposable = [[RACSerialDisposable alloc] init];
    _signal = subject;
    return self;
}

In the earlier example, sourceSignal is the networkRequest , while _signal refers to [RACReplaySubject subject] .

- (RACDisposable *)connect {
    BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);
    if (shouldConnect) {
        self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
    }
    return self.serialDisposable;
}

Calling [self.sourceSignal subscribe:_signal] executes the didSubscribe block of the source signal, using _signal (the replay subject) as the subscriber. After that, all subsequent sendNext or additional subscriptions interact only with the replay subject, not the original source signal. This explains why "side effects only occur once".

Multicast Mechanism – Part 2

Before the second step, we need to understand RACSubject and RACReplaySubject.

--- Divider Start ---

RACSubject

"A subject can be thought of as a signal that you can manually control by sending next, completed, and error."

Typical usage:

RACSubject *letters = [RACSubject subject];
// Outputs: A B
[letters subscribeNext:^(id x) {
    NSLog(@"%@ ", x);
}];
[letters sendNext:@"A"];
[letters sendNext:@"B"];

Implementation details:

- (id)init {
    self = [super init];
    if (self == nil) return nil;
    _disposable = [RACCompoundDisposable compoundDisposable];
    _subscribers = [[NSMutableArray alloc] initWithCapacity:1];
    return self;
}

RACSubject maintains an array of subscribers.

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    NSCParameterAssert(subscriber != nil);
    RACCompoundDisposable *disposable = [RACCompoundDisposable compoundDisposable];
    subscriber = [[RACPassthroughSubscriber alloc] initWithSubscriber:subscriber signal:self disposable:disposable];
    NSMutableArray *subscribers = self.subscribers;
    @synchronized (subscribers) {
        [subscribers addObject:subscriber];
    }
    return [RACDisposable disposableWithBlock:^{
        @synchronized (subscribers) {
            NSUInteger index = [subscribers indexOfObjectWithOptions:NSEnumerationReverse passingTest:^BOOL(id<RACSubscriber> obj, NSUInteger idx, BOOL *stop) {
                return obj == subscriber;
            }];
            if (index != NSNotFound) [subscribers removeObjectAtIndex:index];
        }
    }];
}

Each subscription adds the subscriber to the internal subscribers array.

- (void)sendNext:(id)value {
    [self enumerateSubscribersUsingBlock:^(id<RACSubscriber> subscriber) {
        [subscriber sendNext:value];
    }];
}

When sendNext: is called, the subject forwards the value to every stored subscriber.

RACReplaySubject

"A replay subject saves the values it is sent (up to its defined capacity) and resends those to new subscribers." It inherits from RACSubject and adds buffering capabilities.

- (instancetype)initWithCapacity:(NSUInteger)capacity {
    self = [super init];
    if (self == nil) return nil;
    _capacity = capacity;
    _valuesReceived = (capacity == RACReplaySubjectUnlimitedCapacity ? [NSMutableArray array] : [NSMutableArray arrayWithCapacity:capacity]);
    return self;
}

The object stores a capacity and an array valuesReceived that buffers sent values.

- (RACDisposable *)subscribe:(id<RACSubscriber>)subscriber {
    RACCompoundDisposable *compoundDisposable = [RACCompoundDisposable compoundDisposable];
    RACDisposable *schedulingDisposable = [RACScheduler.subscriptionScheduler schedule:^{
        @synchronized (self) {
            for (id value in self.valuesReceived) {
                if (compoundDisposable.disposed) return;
                [subscriber sendNext:(value == RACTupleNil.tupleNil ? nil : value)];
            }
            if (compoundDisposable.disposed) return;
            if (self.hasCompleted) {
                [subscriber sendCompleted];
            } else if (self.hasError) {
                [subscriber sendError:self.error];
            } else {
                RACDisposable *subscriptionDisposable = [super subscribe:subscriber];
                [compoundDisposable addDisposable:subscriptionDisposable];
            }
        }
    }];
    [compoundDisposable addDisposable:schedulingDisposable];
    return compoundDisposable;
}

When a new subscriber subscribes, the replay subject first re‑emits all buffered values, then adds the subscriber to the normal subscribers array.

- (void)sendNext:(id)value {
    @synchronized (self) {
        [self.valuesReceived addObject:value ?: RACTupleNil.tupleNil];
        [super sendNext:value];
        if (self.capacity != RACReplaySubjectUnlimitedCapacity && self.valuesReceived.count > self.capacity) {
            [self.valuesReceived removeObjectsInRange:NSMakeRange(0, self.valuesReceived.count - self.capacity)];
        }
    }
}

The subject buffers each value, forwards it to current subscribers, and discards old values when the capacity is exceeded.

--- Divider End ---

Having introduced RACReplaySubject, we continue with the second part of multicast.

Multicast Part 2

In the earlier example, two subscriptions are made to connection.signal. According to the replay subject's subscribe: implementation:

The RACReplaySubject stores both subscribers in its internal array.

When the network request succeeds, the subject receives sendNext: , buffers the value, and immediately forwards it to all stored subscribers.

Further Thoughts

If a sendNext occurs before any subscription, the buffered values will still be delivered to later subscribers because of the replay logic. Consider this scenario as an exercise.

The RACSignal+Operation header defines five operators related to multicast & replay: publish , multicast , replay , replayLast , and replayLazily . Their subtle differences become clear after understanding the mechanisms described above. A recommended blog post is listed in the references.

References

ReactiveCocoa GitHub repository ReactiveCocoa Documentation ReactiveCocoa articles on raywenderlich.com