Observable Object

Question

I have created an observer and an observable object class. The idea being that you can create an observer that does some action whenever object being observed changes value.

Design goals:

• Be able to choose to notify all observers every time a value is assigned to the object being observed (even if that value is the same as its current value) or only if the value actually is different.
• If the object being observed is to notify its observers every time it is updated, the underlying object should not be required to define operator==.
• If the object being observed goes out of scope, it should automatically disassociate itself with all of its observers.
• If an observer goes out of scope, it should automatically disassociate itself with the object it's observing.
• The observer and wrapped object being observed should be thread-safe.

Here's what I came up with:

//observable.h
#ifndef OBSERVABLE_T_H
#define OBSERVABLE_T_H

#include <unordered_map>
#include <functional>
#include <mutex>
#include <utility>
#include <memory>
#include <type_traits>

// make things slightly cleaner for pre C++14 compilers
template< bool B>
using enable_ = typename std::enable_if<B, void>::type;

template<typename T, bool NEW_ONLY> class Observer_t;

template<typename T, bool NEW_ONLY = false>
class Observable_t
{
public:
    Observable_t(): object_{} {}

    Observable_t(const T& obj): object_(obj) {}

    ~Observable_t()
    {
        std::lock_guard<std::mutex> lock(mutex_);
        for (auto& obs : observers_)
        {
            obs.second.get().remote_detach(object_);
        }
    }

    unsigned attach(Observer_t<T, NEW_ONLY> &obs)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        unsigned observer_ID = observer_index_++;
        observers_.emplace(std::make_pair(observer_ID, std::ref(obs)));
        return observer_ID;
    }

    bool detach(unsigned observer_ID)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if (observers_.find(observer_ID) == observers_.end())
        {
            return false;
        }
        observers_.erase(observer_ID);
        return true;
    }

    const T& operator=(const T& new_data)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        update_observers(new_data);
        return new_data;
    }

    template<typename... Args>
    void update(Args... args)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        update_observers(T(std::forward<Args>(args)...));
    }

    T operator()() const
    {
        std::lock_guard<std::mutex> lock(mutex_);
        return object_;
    }

private:
    unsigned observer_index_ = 1;

    std::unordered_map<unsigned,
        std::reference_wrapper<Observer_t<T, NEW_ONLY>>> observers_;
    T object_;
    mutable std::mutex mutex_;

    template<bool trigger = NEW_ONLY>
    enable_<trigger>
    update_observers(const T& new_data)
    {
        if (new_data != object_)
        {
            for (auto& obs : observers_)
            {
                obs.second.get() = new_data;
            }
            object_ = new_data;
        }
    }

    template<bool trigger = NEW_ONLY>
    enable_<not trigger>
    update_observers(const T& new_data)
    {
        for (auto& obs : observers_)
        {
            obs.second.get() = new_data;
        }
        object_ = new_data;
    }

};

template<typename T, bool NEW_ONLY>
class Observer_t
{

public:
    Observer_t(){}
    Observer_t(Observable_t<T, NEW_ONLY> *object,
        std::function<void(const T&)> callback): observable_ptr_(object),
        callback_(callback)
    {
        attach();
    }

    ~Observer_t(){detach();}

    void attach()
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if(observer_ID_ != 0)
        {
            observable_ptr_->detach(observer_ID_);
        }
        observer_ID_ = observable_ptr_->attach(*this);
    }

    void detach()
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if(observer_ID_ != 0)
        {
            cached_object_ = (*observable_ptr_)();
            observable_ptr_->detach(observer_ID_);
            observer_ID_ = 0;
        }
    }

    void remote_detach(T &obj)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        cached_object_ = obj;
        observer_ID_ = 0;
    }

    T operator()() const
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if (observer_ID_ == 0) //i.e. we're detached
            return cached_object_;
        else
            return (*observable_ptr_)();
    }

private:
    friend class Observable_t<T, NEW_ONLY>;
    const T& operator=(const T& new_object)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        callback_(new_object);
        return new_object;
    }

    Observable_t<T, NEW_ONLY>* observable_ptr_;
    std::function<void(const T&)> callback_;
    unsigned observer_ID_ = 0;
    T cached_object_;
    mutable std::mutex mutex_;
};

template<typename T>
using Observable_Object_t = Observable_t<T, false>;

template <typename T>
using Object_Observer_t = Observer_t<T, false>;

template<typename T>
using Observable_State_t = Observable_t<T, true>;

template<typename T>
using State_Observer_t = Observer_t<T, true>;

#endif // OBSERVABLE_T_H

A brief test driver:

#include <iostream>
#include <sstream>
#include "observable_t.h"


struct Attitude_t
{
    Attitude_t(double yaw = 0, double pitch = 0, double roll = 0):
        yaw_(yaw), pitch_(pitch), roll_(roll) {}

    std::string str() const
    {
        std::ostringstream oss;
        oss << "Attitude: Yaw: " << yaw_ << " pitch: " << pitch_ << " roll: "
            << roll_;
        return oss.str();
    }
    Attitude_t& operator=(const Attitude_t& other) = default;

private:
    double yaw_;
    double pitch_;
    double roll_;
};

int main()
{
    Observable_State_t<int> oi;

    oi = 3;

    std::cout <<"oi is currently: " << oi() << std::endl;

    State_Observer_t<int> so_i(&oi,[&](const int &new_int)
    {
       std::cout << "so_i received new int: " << new_int <<std::endl;
    });

    oi = 5;

    oi = 5;

    oi = 7;

    oi.update(123442);

    so_i.detach();

    oi = 12;

    std::cout <<"after detaching so_i is " << so_i() << std::endl;

    Observable_Object_t<Attitude_t> attitude;

    attitude = Attitude_t(1,2,3);
    std::cout << attitude().str() << std::endl;

    Object_Observer_t<Attitude_t> att_obs(&attitude, [&](const Attitude_t& new_att)
    {
        std::cout << "att_obs received: " << new_att.str() << std::endl;
    });

    attitude = Attitude_t(4,5,6);
    attitude = Attitude_t(5,6,7);
    attitude.update(12, 34, 52);

    att_obs.detach();

    attitude = Attitude_t(7,8,9);

    std::cout << "Attitude is now: " << attitude().str() << std::endl;

    std::cout << att_obs().str() << std::endl;

    return 0;
}

See it run at coliru.

Thoughts?

Answer 1

Avoid convenience aliases unless it really improves the code

You created the type alias enable_, but it is only used twice. Furthermore, if you wouldn't have seen the alias, and first see enable_<trigger>, then you would first have to find the definition of enable_ to know what this expression does, whereas if you wrote std::enable_if<trigger, void>::type directly, someone who already knows about the standard library's SFINAE types can see immediately what is happening.

Beware of dangling pointers

You have observers and observables that reference each other, and each of them can go out of scope independently. You have to be extra careful here to ensure you don't have dangling references. And it looks like you do, consider:

auto oip = new Observable_State_t<int>;
State_Observer_t<int> soi(oip, [](const int &i){});
delete oip; // detaches soi correctly
soi.attach(); // tries to reattach to deleted object

To solve this, you need to set observable_ptr_ to nullptr when you detach an observer. Of course, that makes it impossible to reattach later. It is probably best to remove the public attach() and detach() functions, make remote_detach() a private function and make Observer_t a friend of Observable_t.

Another option is to use std::shared_ptr and std::weak_ptr to safely manage references between observers and observables. This requires you to always create std::shared_ptrs of these types, and they store std::weak_ptr references to each other. That would look like this (I simplified the code significantly here):

class Observer;

class Observable {
public:
    static shared_ptr<Observable> create() {
        return std::make_shared<Observable>();
    }

    unsigned attach(const std::shared_ptr<Observer> &observer) {
        observers[++index_] = observer;
        return index_;
    }
    
    bool detach(unsigned index) {
        auto it = observers_.find(index);
        if (it != observers_.end()) {
            observers_.erase(it);
            return true;
        } else {
            return false;
        }
    }

    void update() {
        for (auto &weak_obs: observers_) {
            auto obs = weak_obs.lock();
            if (obs) {
                obs->update();
            }
        }
    }

private:
    Observable() {}

    std::unordered_map<unsigned, std::weak_ptr<Observer>> observers_;
    unsigned index_{};
}

class Observer {
public:
    static shared_ptr<Observer> create(const std::shared_ptr<Observable> &object) {
        return std::make_shared<Observer>(object);
    }

    void update() {
        ...
    }

    void operator()() {
        auto obj = object_.lock();
        if (obj) {
            // access the observable directly
        } else {
            // observable has gone out of scope
        }
    }       

private:
    Observer(const std::shared_ptr<Observable> &object): object_(object) {}
    std::weak_ptr<Observable> object_;
}

The above misses the templatization and locking of course. Although the locking is now only necessary to guard object_ inside Observable; updating the shared and weak pointers is already thread-safe.