Repeated signal with variable interval

[eimantas]

I have a PoC of a class using repeated signal with variable interval:

// Repeat.h
#import <Foundation/Foundation.h>

@interface Repeat : NSObject

- (instancetype)initWithInterval:(NSInteger)interval;

@property (assign, nonatomic) NSInteger interval;
@property (readonly, nonatomic) RACSubject *dataSubject;

@end

// Repeat.m
#import <ReactiveCocoa/ReactiveCocoa.h>

#import "Repeat.h"

@interface Repeat ()

@property (readwrite, strong, nonatomic) RACSubject *dataSubject;
@property (readwrite, strong, nonatomic) RACSignal *dataSignal;

@end

@implementation Repeat

- (instancetype)initWithInterval:(NSInteger)interval
{
    self = [super init];
    if (self) {
        self.dataSubject = [RACSubject subject];
        self.dataSignal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
            static int i = 0;
            [subscriber sendNext:@(i++)];
            return nil;
        }];

        self.interval = interval;

        [RACObserve(self, interval) subscribeNext:^(id x) {
            RACSignal *limit = [RACObserve(self, interval) skip:1];
            RACSignal *repeater = [[RACSignal interval:[x integerValue]
                                           onScheduler:[RACScheduler mainThreadScheduler]]
                                   takeUntil:limit];
            RACSignal *tickSignal = [repeater doCompleted:^{
                                         NSLog(@"interval changed to: %lu", self.interval);
                                     }];
            [[tickSignal flattenMap:^RACStream *(id value) { // *
                return self.dataSignal;
            }] subscribeNext:^(id x) {
                [self.dataSubject sendNext:x];
            }];
        }];
    }
    return self;
}

@end

The usage would be following:

Repeat *repeater = [[Repeat alloc] initWithInterval:3];
[repeater.dataSubject subscribeNext:^(id x) {
    // do something with x
}];
...
// somehwere else in the code
repeater.interval = 5;

This allows me to keep the subscription and change only the interval of the delivery of signal values. I'm worried about the part that consumer must actually subscribe to the subject rather than to the signal. I understand this is not a common scenario and guidelines actually encourages avoiding subjects if possible. I tried exposing dataSignal in public interface and replacing the part marked with // * as follows:

[tickSignal subscribeNext:^(id x) {
    [self.dataSignal replayLast];
}];

But this would not work at all and only the initial value would get sent. Has anyone else tried implementing what I am trying to achieve here? I am confident RAC is capable of doing this and my PoC is just an ugly duckling that can be greatly improved.

The gist can be found here: https://gist.github.com/eimantas/51d9dfc81136f340a829

Update: based on @jspahrsummers answer to the question on stackoverflow: http://stackoverflow.com/questions/15075075/when-to-use-racreplaysubject-vs-racmulticastconnection

It seems that RACMulticastConnection is doing the same thing - exposing RACSubject as RACSignal, so I guess I should do the same then?

[jspahrsummers]

If the timer is never really meant to terminate, you could create a recursive method which runs concat:[RACSignal delay:N] and invokes itself from concat:[RACSignal defer:^{}] after that.

In other words, insert delays of N seconds between an infinite series of your own signal.

[eimantas]

The timer can be terminated on user's command. And regarding the recursion - my CS might be rusty, but wouldn't that eventually create the stack frame overflow?

[jspahrsummers]

@eimantas No, because the delay won't be synchronous. RAC will leave the frame and return to the code later after the delay has elapsed.

[eimantas]

I'm sorry, but I still don't get your idea .( specifically - on what should I call concat:[RACSignal delay:N]?

[jspahrsummers]

- (RACSignal *)doStuffRepeatedly {
    return [[[RACSignal
        defer:^{
            // Do some stuff here
            return [RACSignal empty];
        }]
        // Wait 5 seconds
        concat:[RACSignal delay:5]]
        then:^{
            // Do more stuff
            return [self doStuffRepeatedly];
        }];
}

[ikesyo]

Closing this due to inactivity.