Rxswift: DelegateProxy for background threads

[rkawajiri]

I would like to use DelegateProxy even on background threads. e.g. CoreBluetooth https://github.com/SideEffects-xyz/RxBluetooth

However, DelegateProxy can be used only from Main thread for now.

/**
Base class for DelegateProxyType protocol.
This implementation is not thread safe and can be used only from one thread (Main thread).
*/
public class DelegateProxy : _RXDelegateProxy {

https://github.com/ReactiveX/RxSwift/blob/83bac6db0cd4f7dd3e706afc6747bd5797ea16ff/RxCocoa/Common/DelegateProxy.swift#L20

My idea is that DelegateProxy can be specified an SerialScheduler or something.

[kzaher]

Hi @rkawajiri,

I don't have any experience with bluetooth :) Why do you think there is a threading problem here?

This sentence

This implementation is not thread safe and can be used only from one thread (Main thread).

means that you can only touch those delegate convenience properties from main thread and that delegate methods are expected to fire on main thread.

I would just want to have more information about this.

[rkawajiri]

Core Bluetooth delegates can be invoked on background threads.
And to set some delegates needs information given on the other delegates.

An example snippet is like this.

rx_didDiscoverPeripheral
  .subscribeNext { (peripheral, _, _) in  // on background thread
    self.rx_didConnectPeripheral  // throw main thread error
      .filter { $0.identifier == peripheral.identifier }
      .take(1)
      .subscribeNext { peripheral in  // on background thread
        do_something_more(peripheral)
    }
    self.connectPeripheral(peripheral, options: nil)
}

The delegate didDiscoverPeripheral can be called on background thread (peripheral means an BLE accessary). Then before connect, we need to set didConnectPeripheral delegate.

Do you have any idea to avoid the error?

[rkawajiri]

This modification works on my project for now and passed the unit tests of RxSwift project.

https://github.com/ReactiveX/RxSwift/compare/develop...rkawajiri:feature/delegate_proxy_force_on_main_thread
It synchronously run the delegate proxy tasks on main thread instead of throwing fatal error.

[kzaher]

Hi @rkawajiri ,

I wouldn't suggest making a dispatch_sync(dispatch_get_main_queue()) because you would create a deadlock that way.

I think we'll take this work for the 2.4.0 release.

[rkawajiri]

Thanks for your warning and considering to enhance for future release.

As you mentioned, the deadlocks occurred. For now, I avoid the problem by using scheduler like this.

Observable.just(Void)
  .observeOn(MainScheduler.instance)
  .flatMap { _ in
    let ret = self.rx_didDiscoverPeripheral
    trigger_discovering()
    return ret
  }
  .subscribeNext { (peripheral, _, _) in  // on background thread
    Observable.just(Void)
      .observeOn(MainScheduler.instance)
      .flatMap { _ in
        let ret = self.rx_didConnectPeripheral
        trigger_connection(peripheral)
        return ret
      }
      .filter { $0.identifier == peripheral.identifier }
      .take(1)
      .subscribeNext { peripheral in  // on background thread
        do_something_more(peripheral)
    }
}

[kzaher]

Hi @rkawajiri ,

I was now performing some changes around DelegateProxyType, but it seems to me that this is a more complex request then I originally assumed.

We can make DelegateProxyType thread safe, but that would mean we would be setting delegates on the original object potentially on various dispatch queues and that is problematic for sure.

Otherwise we would need add another static method to DelegateProxyType that describes on which dispatch_queue to perform those operations, and always do dispatch_async to that dispatch queue to prevent deadlocks.

Also, some API take operation queues.

This is starting to look more and more like a not simple thing to add :(

I'm starting to think that in those more complex scenarios it would be wiser, more performant and vastly simpler to manually add Subjects and install delegates on necessary queues.

E.g.

    /**
    Reactive wrapper for `contentOffset`.
    */
    public var rx_contentOffset: ControlProperty<CGPoint> {

        let bindingObserver = UIBindingObserver(UIElement: self) { scrollView, contentOffset in
            scrollView.contentOffset = contentOffset
        }

        return ControlProperty(values: privateContentOffsetSubject, valueSink: bindingObserver)
    }

[kzaher]

I don't see any simple way of supporting this at the moment, if I figure out something reasonably simple, we can reopen this.