2

I have a problem with an Observable<Data?> function that is called so many times and so fast that the function doesn't complete until the next one is run. This makes sense and is good in most cases. But in this case it becomes very problematic because the function in question uses a counter.

func sendMessage(input: MessageToSend) -> Observable<Data?> {
    input.counter = self.counter

    print("-- 1", input.counter)

    let transformMessage = transform(message: input)
    self.counter += 1

    print("-- 2", input.counter)

    return transformMessage
}

Obviously I need input.counter to increase by 1 every time the function is called. But unfortunately this isn't what's happening. Because of the async nature of rx sendMessage() is run and adds 0 to input.counter, but before it has had the chance to increment self.counter by 1, sendMessage() is run again and again adds 0 to input.counter. Then the first call reaches self.counter += 1, and immediately after, the second call reaches self.counter += 1 as well. So now the counter has reached 2. So when the third call is made, input.counter gets 2 as value.

The prints will look like this:

-- 1 0
-- 1 0
-- 2 0
-- 2 0
-- 1 2
-- 2 2

My initial idea on how to fix this was to force the call to a serial task. So instead of calling it like this:

disposeble = tsiHandler?.sendMessage(message: message).subscribe()

I called it like this:

disposeble = tsiHandler?.sendMessage(message: message)
    .subscribeOn(SerialDispatchQueueScheduler(internalSerialQueueName: "serial"))
    .subscribe()

In my world this would force it to become serial which would make all the calls to sendMessage() wait for other calls to complete. But for some reason it doesn't work. I get the exact same result in the prints. Am I missunderstanding how SerialDispatchQueueScheduler work?

1
  • What is calling sendMessage(message:) so quickly? There is nothing in the code you showed that would do that...
    – Daniel T.
    Apr 4, 2022 at 11:19

2 Answers 2

1

As it is written, each call to sendMessage is being placed on a different scheduler, so even if the schedulers are serial, that wouldn't accomplish what you are trying to do. Instead of creating a new scheduler for each call, have them all use the same scheduler.

Even so, this code looks fishy and your comments about it implies a fundamental misunderstanding of Rx. For example it is not inherently async. It merely sets up callback chains...

3
  • Thanks for your response! Could you elaborate a bit? How do I use the same scheduler? I thought I already did that. Or do you mean that every time it hooks on onSubscribe() it's a new serial task that doesn't consider the old one? Apr 4, 2022 at 12:13
  • No, I mean that SerialDispatchQueueScheduler(internalSerialQueueName: "serial") is a constructor. You are literally making a new scheduler every time that is called. Just call the constructor once and store the instance, then reuse that instance here.
    – Daniel T.
    Apr 4, 2022 at 13:11
  • Depending on how transform(message:) is implemented, the subscribeOn(_:) probably doesn't do what you think it does anyway...
    – Daniel T.
    Apr 4, 2022 at 15:43
0

I believe you should make sure:

  1. To hold one instance of the SerialDispatchQueueScheduler.
  2. When sending a message you have to observe(on: scheduler) and not subscribe(on: scheduler).

Hope that helps.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.