Reactivecocoa: Best design pattern for starting only one async operation and share signal result

[notxcain]

Let's pretend that RACSignal *updateSignal encapsulates some async task. When subscription occurs I want every new subscription during the async task to receive same result as the first subscriber.
Right now my code looks like this:

- (RACSignal *)doWork
{
    if (_sharedWorkSignal) return _sharedWorkSignal;

    RACSignal *requestSignal = ...;
    _sharedWorkSignal = [[[[requestSignal publish] autoconnect] doCompleted:^{
        _sharedWorkSignal = nil;
    }] doError:^(NSError *error) {
        _sharedWorkSignal = nil;
    }];

    return _sharedWorkSignal;
}

It works but I don't really like how it looks. Do I miss something important here?

[jspahrsummers]

You can use -replay or -replayLazily to achieve this.

[notxcain]

But still I had to use this instnace variable to hold replayed signal while async work is being done?

[jspahrsummers]

I'm not entirely sure I understand your question. You may want to check out my example of fetching an authentication token before API calls, since it uses a similar pattern.

[notxcain]

Well may be I've provided a wrong description. While the async operation is running because of the first subscription, I want any other subscription to share the result, but when this work is done and signal completes, I want any new subscription to start this work again. Sorry for my english ;)

[jspahrsummers]

In that case, I'm not sure there's a significantly better approach than what you've done. However, there are some things to note:

1. Reading and writing _sharedWorkSignal is not thread-safe. You should use an atomic property for this.
2. You probably want to use -replayLazily instead of -publish and -autoconnect, or else new subscribers won't receive values sent before they attached.
3. You can use -finally: instead of -doCompleted: and -doError:.

[notxcain]

Great! Thank you so much Justin! How could I miss -finally. And thanks for your notes! Closed.

[jspahrsummers]

:sparkles:

[zakdances]

If you don't want to use methods, instance variables, or properties, you could try something like this:

RACSignal *testSig = [[RACSignal return:[NSMutableArray arrayWithCapacity:1]] map:^id(NSMutableArray *array) {

            if (array.count == 0) {

                RACSignal *sig = [[RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {


                    // Do stuff
                        [subscriber sendNext:something];
                        [subscriber sendCompleted];

                        return nil;
                    }] doCompleted:^{
                        [array removeAllObjects];
                    }];


                    [array addObject:sig.replayLast];
                }

                return array.lastObject;
}].replayLazily;

Using an array or subject, you can keep a reference to something without declaring a variable or using a property.

[jspahrsummers]

That code isn't thread safe. Sometimes using a variable really is the best answer.

[zakdances]

Ok...so it sounds like the safest strategy is to use atomic properties as "anchors", especially for any variables that you weave in and out of RAC blocks.

[fabiomassimo]

I'm facing a kind of same issue with a shared signal that I won't to "re-trigger" when it's needed.
I've tried a lot of possible solution in RACMultiConnection fashion but no matter what I'm always facing the same problem: every time I want to re-trigger my source signal it sends next: to all previous subscribers.

More in detail:

1) the network request is made by a convenience methods that wraps up an old delegate implementation:

RACSignal *performRequestSignal =
    [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
        // Create data received signal
        RACSignal *dataReceivedSignal = [[self rac_signalForSelector:@selector(dataReceived:forRequest:) fromProtocol:@protocol(MainDataControllerDelegate)] filter:^BOOL(RACTuple *returnTuple) {
            DataRequest *request = returnTuple.second;
            DataResponse *dataResponse = returnTuple.first;

            return ((request.type == dataRequest.type) && ([dataResponse.value isKindOfClass:dataResponseClass]));
        }];

        // Subscribe
        [dataReceivedSignal subscribeNext:^(RACTuple *returnTuple) {
            DataResponse *dataResponse = returnTuple.first;
            [subscriber sendNext:dataResponse.value];
            [subscriber sendCompleted];

        }];

        // Create data request failed signal
        RACSignal *requestFailedSignal = [self rac_signalForSelector:@selector(requestFailed:error:) fromProtocol:@protocol(MainDataControllerDelegate)];

        // Subscribe
        [requestFailedSignal subscribeNext:^(RACTuple *returnTuple) {
            NSError *error = returnTuple.second;
            [subscriber sendError:error];
        }];

        // Trigger the request to API
        [[MainDataController sharedInstance] addRequest:dataRequest delegate:self priority:queuePriority];

        return [RACDisposable disposableWithBlock:^{
            [[MainDataController sharedInstance] cancelRequest:dataRequest];
        }];
    }];

    return performRequestSignal;

2) This method uses previous one to create a convenience signal which we are calling "retrieveThingsSignal"

- (RACSignal *)retrieveThings
{
    if (_retrieveDebtors) return _retrieveDebtors;

    RACSignal *requestSignal = [self retrieveThingsSignal];
    _retrieveDebtors = [[requestSignal replayLast] finally:^{
        _retrieveThings = nil;
    }];

    return _retrieveThings;

3) every time I won't to re-trigger the signal a run this code

[[self retrieveThings] subscribeNext:^(NSArray *things) {
        @strongify(self);
        self.things = things;
    } error:^(NSError *error) {
        @strongify(self)
        self.things = nil;
    }];

My self.things gets updated accordingly to every time I need to make a new request to retreiveThings but, by putting a breakpoint inside the 1) step, I'm receiving that the next: is sent to ALL previous subscribers. If I call retrieveThings twice, I'm receiving, from that breakpoint, that the base signal is sending next: to two different subscribers (one will actually run the subscribeNext: block). If three to three subscribers, and so on...

I can I prevent this behaviour?

I would like to share the last sent value from my retrieveThings method without trigger EVERY time the network request but, when needed, I need to clean it such that next subscribe will actually trigger the network request.