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?