Observable is the easiest way to observe values in Swift.

As discussed in https://github.com/roberthein/Observable/issues/24, this PR tries to fix the issue. @roberthein What do you think about this? I would like to know your opinion on this. Thanks

I restore back mutex because it less expensive for usage than thread. Deadlock problem developer can solve with add observer dispatch queue.

Subject sometimes can be usefull. E.g. when we call something from net, and waiting result. For this moment you don't have value.

This PR fixes the issue #32 where the first time we call the observer, right after subscribing, we where not using the given queue from the parameter.

I have a scenario where I'm using an observer in a singleton, so it's shared between some classes. The problem arises when this observer triggers the instantiation of another class which also uses the same observable (It looks like this is related to #19 ). In this scenario, the app freezes because since the first event is not using the queue, the second time I subscribe it reaches lock for the second time, before the first has a chance to unlock.

I noticed that in line 45, the queue is not really used at all. https://github.com/roberthein/Observable/blob/b8e87082d65f3f322f35973384e23590031f078c/Observable/Classes/Observable.swift#L45

Is there any reason for that? Or do you think we could use the queue on that line, like this:

   if let queue = queue {
       queue.async {
           observer(self.value, nil)
       }
   } else {
       observer(value, nil)
   }

Here are some test scenarios comparing the old/new implementation:

    // Hangs with current implementation
    // Succeeds with new implementation
    func test_whenChangingValue_shouldSucceed() {
        let exp = expectation(description: "")
        let observable = Observable(0)

        observable.observe(DispatchQueue.main) { _, _ in
            observable.removeAllObservers()
            observable.value = 1
            exp.fulfill()
        }.add(to: &disposal)

        wait(for: [exp], timeout: 1.0)
        XCTAssert(true)
    }

    // Succeeds with current implementation
    // Succeeds with new implementation
    func test_whenUsingDispatchMain_shouldSucceed() {
        let exp = expectation(description: "")
        let singleton = Singleton.shared
        let callback: (() -> Void) = {
            exp.fulfill()
        }
        var aClass: AClass?

        singleton.observable.observe { _, _ in
            singleton.observable.removeAllObservers()
            DispatchQueue.main.async {
                aClass = AClass(callback: callback)
            }
        }.add(to: &disposal)

        print(aClass.debugDescription)
        wait(for: [exp], timeout: 1.0)
        XCTAssert(true)
    }

    // Hangs with current implementation
    // Succeeds with new implementation
    func test_whenPassingDispatchMain_shouldSucceed() {
        let exp = expectation(description: "")
        let singleton = Singleton.shared
        let callback: (() -> Void) = {
            exp.fulfill()
        }
        var aClass: AClass?

        singleton.observable.observe(DispatchQueue.main) { _, _ in
            singleton.observable.removeAllObservers()
            aClass = AClass(callback: callback)
        }.add(to: &disposal)

        print(aClass.debugDescription)
        wait(for: [exp], timeout: 2.0)
        XCTAssert(true)
    }

    // Hangs with current implementation
    // Succeeds with new implementation
    func test_whenUsingDispatchGlobal_shouldSucceed() {
        let exp = expectation(description: "")
        let singleton = Singleton.shared
        let callback: (() -> Void) = {
            exp.fulfill()
        }
        var aClass: AClass?

        singleton.observable.observe(DispatchQueue.global()) { _, _ in
            singleton.observable.removeAllObservers()
            aClass = AClass(callback: callback)
        }.add(to: &disposal)

        print(aClass.debugDescription)
        wait(for: [exp], timeout: 1.0)
        XCTAssert(true)
    }

    // Hangs with current implementation
    // Hangs with new implementation
    func test_whenNotUsingDispatch_shouldHang() {
        let exp = expectation(description: "")
        let singleton = Singleton.shared
        let callback: (() -> Void) = {
            exp.fulfill()
        }
        var aClass: AClass?

        singleton.observable.observe { _, _ in
            singleton.observable.removeAllObservers()
            aClass = AClass(callback: callback)
        }.add(to: &disposal)

        print(aClass.debugDescription)
        wait(for: [exp], timeout: 1.0)
        XCTAssert(true)
    }

    class AClass {
        var disposal = Disposal()

        init(callback: @escaping () -> Void) {
            let singleton = Singleton.shared
            singleton.observable.observe { _, _ in
                callback()
            }.add(to: &disposal)
        }
    }

    class Singleton {
        static let shared = Singleton()
        var observable = Observable(0)
        private init() {}
    }

Replaces Lock with DispatchQueues to avoid deadlock when updating an Observable value from within its Observation.

Added a read/write queue for set/get the value and added another queue to decoupled the changes broadcasting to observers from the read/write.

I opened this issue, to continue the conversation started in https://github.com/roberthein/Observable/pull/12 about tests.

Do you want create the Xcode project to add the test target, or should I create it? 🙂

Our project has an extension to wrap some existing Observable-like APIs in the Realm framework so that they match the Observable API. In order to do this we need to create a Disposable object, but the initializer for Disposable isn't marked as public so it defaults to internal.

Can the initializer for Disposable be made public?

The ImmutableObservable class is great, since it allows you to have a public API in your classes that is safe from external side effects, but the naming semantics are backwards.

The current name would imply that ImmutableObservable is a special kind of Observable that adds immutability, but that's not the case. Instead Observable is a special kind of ImmutableObservable that adds mutability, but that's not apparent from reading the names.

I addition, it litters your public API with extra information. The client shouldn't know or care that the Observable is immutable, that's an implementation detail that makes usages more verbose and less readable.

I suggest flipping the names, so the standard implementation is Observable and the subclass is MutableObservable. That would make the relationship clear and clean up public usages.

class SomeViewModel {
    /// Public property
    var position: Observable<CGPoint> = {
        return positionSubject
    }
    
    /// Private property
    private let positionSubject = MutableObservable(CGPoint.zero)
}

I've just noticed that updating the value of an observable from within an observation results in deadlock because a mutex is being used to ensure thread safety. The mutex could be replaced by DispatchQueues and avoid this issue. Is there a specific reason as to why mutexes are being used instead of DispatchQueues?

I can do a PR for this.

Hello,

Firstly a big thanks for this library ! I'm using Observable in a lot of projects and I love it. It fit really well with mvvm. But if you have a lot of properties, the viewController get fat. 1 observable is about 3 lines of codes, so 10 observables is 30 lines. Readability and maintainability become less clear.

The idea is to reduce the number of line of code to juste one, for all simple cases.

What do you think about adding something like that :

func bind<O: NSObject>(to object: O, _ objectKeyPath: ReferenceWritableKeyPath<O, T>, disposal: inout Disposal)
or
func bind<O: NSObject, R>(to object: O, _ objectKeyPath: ReferenceWritableKeyPath<O, R>, disposal: inout Disposal, transform: @escaping (T) -> R)

usage:

class ViewModel {
   @observable var firstname: String 
}

class ViewController {

    @IBOutlet private weak var firstnameTextField: UITextField!

    func viewDidLoad() {
        viewModel.firstname.bind(to: self.firstnameTextField, \.text, disposal: &self.disposal)
        // or 
        viewModel.firstname.bind(to: self.firstnameTextField, \.text, disposal: &self.disposal) {
              v?.uppercased()
        }

    }
}

I have make some tests with simple cases and it's seems to works, but I'm sure that more work is needed to fit complexes cases and generic makes me sick.

if this proposal fits with the spirit of Observable, I can make a PR. (if not I keep that in an extension 😉 )

Hey, Currently Observable creates a deadlock when updating an Observable value from within its Observation as described https://github.com/roberthein/Observable/issues/19 and https://github.com/roberthein/Observable/pull/20.

This PR migrates the lock from pthread_mutex_t to NSRecursiveLock which avoids the deadlock problem. This is also the solution used in RxSwift 1 2.

Regarding performance, it has a small overhead, which is expected, but I think it's not something noticeable, and it will avoid the dead lock situation.

Here are some performance tests on pthread_mutex_t and NSRecursiveLock.

https://github.com/nyisztor/LockTest#the-results

https://gist.github.com/azinman/5410263c62157086943a#file-appdelegate-m-L30-L41

 Pthread mutex avg: 133 ns
 Pthread reentrant mutex avg: 194 ns
 Pthread reentrant mutex + exceptions avg: 191 ns
 Pthread 3x reentrant mutex + exceptions avg: 262 ns
 Pthread reentrant mutex contended on main avg: 742 ns
 Pthread reentrant mutex contended on high priority avg: 915 ns

 NSRecursiveLock avg: 188 ns
 NSRecursiveLock + exception handling avg: 179 ns
 NSRecursiveLock + exception handling 3x avg: 348 ns
 NSRecursiveLock under contention on main + exception handling avg: 758 ns
 NSRecursiveLock under contention on high priority + exception handling avg: 928 ns

This PR also has a unit tests that tests the deadlock situation.

@roberthein Please let me know if there is any additional changes to merge this 🙂 Thanks

What do you thinks about adding something like combineLatest function ? This could be really useful when handling 2 or more observable values.

Like that

Observable.combine([observables]) { (values, oldValues) in 

}.add()

What would be your approche to do that ?

Ben

Hi,

Is a skip method possible ?

This could simplify unit test and some edge case where first or more value must be ignore.

First I can simply change

public func observe(_ queue: DispatchQueue? = nil, _ observer: @escaping Observer) -> Disposable 

to

public func observe(_ queue: DispatchQueue? = nil, skipFirst: Bool = false,  _ observer: @escaping Observer) -> Disposable 

This could solve the first emitted null value when using Observable(String?) = Observable(nil)

But i can't add skip(count: Int) method without a lot of changes, maybe due to a lack of knowledge about the library.

✌️ Ben

Hello,

Test bundle seems to be broken. I can't find it in _Pod.project or Observable_Example.xcodeproj

A good reason to keep the auto generate pod format? We could just have a simple framework project with unit tests, examples, .podspec, and Package.swift files. I find the format proposed by cocoapods is complex.