【Poco库源码解析】Poco库的事件

1.使用示例

#include "Poco/BasicEvent.h"
#include "Poco/Delegate.h"
#include <iostream>
using Poco::BasicEvent;
using Poco::Delegate;
class Source
{
	public:
	BasicEvent<int> theEvent;
	void fireEvent(int n)
	{
	theEvent(this, n);
	// theEvent.notify(this, n); // alternative syntax
	}
};

class Target
{
public:
 void onEvent(const void* pSender, int& arg)
 {
 	std::cout << "onEvent: " << arg << std::endl;
 }
};

int main(int argc, char** argv)
{
	Source source;
	Target target;
	source.theEvent += Poco::delegate(&target, &Target::onEvent);
	source.fireEvent(42);
	source.theEvent -= Poco::delegate(&target, &Target::onEvent);
	return 0;
}

2.Event流动过程

从图上看实现事件的类被分成了几类：

1. Delegate：
   AbstractDelegate,Delegate,Expire,FunctionDelegate,AbstractPriorityDelegate,PriorityDelegate,FunctionPriorityDelegate

2. Strategy：
   NotificationStrategy，PriorityStrategy，DefaultStrategy，FIFOStrategy

3. Event：
   AbstractEvent，PriorityEvent，FIFOEvent，BasicEvent

3.源码分析

Delegate 类：

template <class TObj, class TArgs, bool withSender = true> 
class Delegate: public AbstractDelegate<TArgs>
{
public:
	typedef void (TObj::*NotifyMethod)(const void*, TArgs&);

	Delegate(TObj* obj, NotifyMethod method):
		_receiverObject(obj), 
		_receiverMethod(method)
	{
	}

	Delegate(const Delegate& delegate):
		AbstractDelegate<TArgs>(delegate),
		_receiverObject(delegate._receiverObject),
		_receiverMethod(delegate._receiverMethod)
	{
	}

	~Delegate()
	{
	}
	
	Delegate& operator = (const Delegate& delegate)
	{
		if (&delegate != this)
		{
			this->_receiverObject = delegate._receiverObject;
			this->_receiverMethod = delegate._receiverMethod;
		}
		return *this;
	}

	bool notify(const void* sender, TArgs& arguments)
	{
		Mutex::ScopedLock lock(_mutex);
		if (_receiverObject)
		{
			(_receiverObject->*_receiverMethod)(sender, arguments);
			return true;
		}
		else return false;
	}

	bool equals(const AbstractDelegate<TArgs>& other) const
	{
		const Delegate* pOtherDelegate = reinterpret_cast<const Delegate*>(other.unwrap());
		return pOtherDelegate && _receiverObject == pOtherDelegate->_receiverObject && _receiverMethod == pOtherDelegate->_receiverMethod;
	}

	AbstractDelegate<TArgs>* clone() const
	{
		return new Delegate(*this);
	}
	
	void disable()
	{
		Mutex::ScopedLock lock(_mutex);
		_receiverObject = 0;
	}

protected:
	TObj*        _receiverObject;
	NotifyMethod _receiverMethod;
	Mutex        _mutex;

private:
	Delegate();
};

我们可以看到 Delegate 类中存储了目标类实例的指针_receiverObject，同时存储了目标类处理函数的入口地址_receiverMethod。

DefaultStrategy类：

template <class TArgs, class TDelegate> 
class DefaultStrategy: public NotificationStrategy<TArgs, TDelegate>
	/// Default notification strategy.
	///
	/// Internally, a std::vector<> is used to store
	/// delegate objects. Delegates are invoked in the
	/// order in which they have been registered.
{
public:
	typedef TDelegate*                   DelegateHandle;
	typedef SharedPtr<TDelegate>         DelegatePtr;
	typedef std::vector<DelegatePtr>     Delegates;
	typedef typename Delegates::iterator Iterator;

public:
	DefaultStrategy()
	{
	}

	DefaultStrategy(const DefaultStrategy& s):
		_delegates(s._delegates)
	{
	}

	~DefaultStrategy()
	{
	}

	void notify(const void* sender, TArgs& arguments)
	{
		for (Iterator it = _delegates.begin(); it != _delegates.end(); ++it)
		{
			(*it)->notify(sender, arguments);
		}
	}

	DelegateHandle add(const TDelegate& delegate)
	{
		DelegatePtr pDelegate(static_cast<TDelegate*>(delegate.clone()));
		_delegates.push_back(pDelegate);
		return pDelegate.get();
	}

	void remove(const TDelegate& delegate)
	{
		for (Iterator it = _delegates.begin(); it != _delegates.end(); ++it)
		{
			if (delegate.equals(**it))
			{
				(*it)->disable();
				_delegates.erase(it);
				return;
			}
		}
	}
	
	void remove(DelegateHandle delegateHandle)
	{
		for (Iterator it = _delegates.begin(); it != _delegates.end(); ++it)
		{
			if (*it == delegateHandle)
			{
				(*it)->disable();
				_delegates.erase(it);
				return;
			}
		}
	}

	DefaultStrategy& operator = (const DefaultStrategy& s)
	{
		if (this != &s)
		{
			_delegates = s._delegates;
		}
		return *this;
	}

	void clear()
	{
		for (Iterator it = _delegates.begin(); it != _delegates.end(); ++it)
		{
			(*it)->disable();
		}
		_delegates.clear();
	}

	bool empty() const
	{
		return _delegates.empty();
	}

protected:
	Delegates _delegates;
};

DefaultStrategy 是一组委托的集合。

BasicEvent 类：

template <class TArgs, class TMutex = FastMutex> 
class BasicEvent: public AbstractEvent < 
	TArgs, DefaultStrategy<TArgs, AbstractDelegate<TArgs> >,
	AbstractDelegate<TArgs>,
	TMutex
>
	/// A BasicEvent uses the DefaultStrategy which 
	/// invokes delegates in the order they have been registered.
	///
	/// Please see the AbstractEvent class template documentation
	/// for more information.
{
public:
	BasicEvent()
	{
	}

	~BasicEvent()
	{
	}

private:
	BasicEvent(const BasicEvent& e);
	BasicEvent& operator = (const BasicEvent& e);
};

AbstractEvent类：

template <class TArgs, class TStrategy, class TDelegate, class TMutex = FastMutex> 
class AbstractEvent
{
public:
	typedef TDelegate* DelegateHandle;
	typedef TArgs Args;

	AbstractEvent(): 
		_executeAsync(this, &AbstractEvent::executeAsyncImpl),
		_enabled(true)
	{
	}

	AbstractEvent(const TStrategy& strat): 
		_executeAsync(this, &AbstractEvent::executeAsyncImpl),
		_strategy(strat),
		_enabled(true)
	{	
	}

	virtual ~AbstractEvent()
	{
	}

	void operator += (const TDelegate& aDelegate)
		/// Adds a delegate to the event. 
		///
		/// Exact behavior is determined by the TStrategy.
	{
		typename TMutex::ScopedLock lock(_mutex);
		_strategy.add(aDelegate);
	}
	
	void operator -= (const TDelegate& aDelegate)
		/// Removes a delegate from the event.
		///
		/// If the delegate is not found, this function does nothing.
	{
		typename TMutex::ScopedLock lock(_mutex);
		_strategy.remove(aDelegate);
	}
	
	DelegateHandle add(const TDelegate& aDelegate)
		/// Adds a delegate to the event. 
		///
		/// Exact behavior is determined by the TStrategy.
		///
		/// Returns a DelegateHandle which can be used in call to
		/// remove() to remove the delegate.
	{
		typename TMutex::ScopedLock lock(_mutex);
		return _strategy.add(aDelegate);
	}
	
	void remove(DelegateHandle delegateHandle)
		/// Removes a delegate from the event using a DelegateHandle
		/// returned by add().
		///
		/// If the delegate is not found, this function does nothing.
	{
		typename TMutex::ScopedLock lock(_mutex);
		_strategy.remove(delegateHandle);
	}
		
	void operator () (const void* pSender, TArgs& args)
		/// Shortcut for notify(pSender, args);
	{
		notify(pSender, args);
	}
	
	void operator () (TArgs& args)
		/// Shortcut for notify(args).
	{
		notify(0, args);
	}

	void notify(const void* pSender, TArgs& args)
		/// Sends a notification to all registered delegates. The order is 
		/// determined by the TStrategy. This method is blocking. While executing,
		/// the list of delegates may be modified. These changes don't
		/// influence the current active notifications but are activated with
		/// the next notify. If a delegate is removed during a notify(), the
		/// delegate will no longer be invoked (unless it has already been
		/// invoked prior to removal). If one of the delegates throws an exception, 
		/// the notify method is immediately aborted and the exception is propagated
		/// to the caller.
	{
		Poco::ScopedLockWithUnlock<TMutex> lock(_mutex);
		
		if (!_enabled) return;
		
		// thread-safeness: 
		// copy should be faster and safer than blocking until
		// execution ends
		TStrategy strategy(_strategy);
		lock.unlock();
		strategy.notify(pSender, args);
	}

	bool hasDelegates() const {
		return !(_strategy.empty());
	}

	ActiveResult<TArgs> notifyAsync(const void* pSender, const TArgs& args)
		/// Sends a notification to all registered delegates. The order is 
		/// determined by the TStrategy. This method is not blocking and will
		/// immediately return. The delegates are invoked in a seperate thread.
		/// Call activeResult.wait() to wait until the notification has ended.
		/// While executing, other objects can change the delegate list. These changes don't
		/// influence the current active notifications but are activated with
		/// the next notify. If a delegate is removed during a notify(), the
		/// delegate will no longer be invoked (unless it has already been
		/// invoked prior to removal). If one of the delegates throws an exception, 
		/// the execution is aborted and the exception is propagated to the caller.
	{
		NotifyAsyncParams params(pSender, args);
		{
			typename TMutex::ScopedLock lock(_mutex);

			// thread-safeness: 
			// copy should be faster and safer than blocking until
			// execution ends
			// make a copy of the strategy here to guarantee that
			// between notifyAsync and the execution of the method no changes can occur
				
			params.ptrStrat = SharedPtr<TStrategy>(new TStrategy(_strategy));
			params.enabled  = _enabled;
		}
		ActiveResult<TArgs> result = _executeAsync(params);
		return result;
	}
	
	void enable()
		/// Enables the event.
	{
		typename TMutex::ScopedLock lock(_mutex);
		_enabled = true;
	}

	void disable()
		/// Disables the event. notify and notifyAsnyc will be ignored,
		/// but adding/removing delegates is still allowed.
	{
		typename TMutex::ScopedLock lock(_mutex);
		_enabled = false;
	}

	bool isEnabled() const
	{
		typename TMutex::ScopedLock lock(_mutex);
		return _enabled;
	}

	void clear()
		/// Removes all delegates.
	{
		typename TMutex::ScopedLock lock(_mutex);
		_strategy.clear();
	}
	
	bool empty() const
		/// Checks if any delegates are registered at the delegate.
	{
		typename TMutex::ScopedLock lock(_mutex);
		return _strategy.empty();
	}

protected:
	struct NotifyAsyncParams
	{
		SharedPtr<TStrategy> ptrStrat;
		const void* pSender;
		TArgs       args;
		bool        enabled;
		
		NotifyAsyncParams(const void* pSend, const TArgs& a):ptrStrat(), pSender(pSend), args(a), enabled(true)
			/// Default constructor reduces the need for TArgs to have an empty constructor, only copy constructor is needed.
		{
		}
	};

	ActiveMethod<TArgs, NotifyAsyncParams, AbstractEvent> _executeAsync;

	TArgs executeAsyncImpl(const NotifyAsyncParams& par)
	{
		if (!par.enabled)
		{
			return par.args;
		}

		NotifyAsyncParams params = par;
		TArgs retArgs(params.args);
		params.ptrStrat->notify(params.pSender, retArgs);
		return retArgs;
	}

	TStrategy _strategy; /// The strategy used to notify observers.
	bool      _enabled;  /// Stores if an event is enabled. Notfies on disabled events have no effect
	                     /// but it is possible to change the observers.
	mutable TMutex _mutex;

private:
	AbstractEvent(const AbstractEvent& other);
	AbstractEvent& operator = (const AbstractEvent& other);
};

4.总结

（1）source.theEvent += Poco::delegate(&target, &Target::onEvent)时， Poco::delegate会创建Delegate<TObj, void, true>(pObj, NotifyMethod)；
（2） += 会 _strategy.add(aDelegate); 将代理保存在_strategy， _strategy默认是 DefaultStrategy
（3） DefaultStrategy类中将_delegates.push_back(pDelegate)， _delegates 是 std::vector
（4）当有新事件时， theEvent(this, n)，将调用BasicEvent 的void operator () (const void* pSender, TArgs& args)，从而调用void notify(const void* pSender, TArgs& args)
（5）strategy.notify(pSender, args)， 将_delegates遍历通知每个Delegate
（6）Delegate将调用事件回调。

所以，通过Delegate执行应用程序具体的回调，DefaultStrategy中保存了所有代理。source -->basicEvent --> DefaultStrategy --> Delegate --> Target

以上是同步事件，异步事件如下使用例子

#include "Poco/BasicEvent.h"
#include "Poco/Delegate.h"
#include "Poco/ActiveResult.h"
#include <iostream>

using Poco::BasicEvent;
using Poco::Delegate;
using Poco::ActiveResult;

class TargetAsync
{
public:
	void onAsyncEvent(const void* pSender, int& arg)
	{
		std::cout << "onAsyncEvent: " << arg <<  " Current Thread Id is :" 
		<< GetCurrentThreadId() << " "<< std::endl;
		return;
	}
};
template<typename RT> class Source
{
public:
	BasicEvent<int> theEvent;
	ActiveResult<RT> AsyncFireEvent(int n)
	{
		return ActiveResult<RT> (theEvent.notifyAsync(this, n));
	}
};

int main(int argc, char** argv)
{
	Source<int> source;
	TargetAsync target;
	std::cout <<  "Main Thread Id is :" << GetCurrentThreadId() << " " << std::endl;
	source.theEvent += Poco::delegate(&target, &TargetAsync::onAsyncEvent);
	ActiveResult<int> Targs = source.AsyncFireEvent(43);
	Targs.wait();
	std::cout << "onEventAsync: " << Targs.data() << std::endl;
	source.theEvent -= Poco::delegate(&target, &TargetAsync::onAsyncEvent);
	return 0;
}

5. 参考链接

https://blog.csdn.net/arau_sh/article/details/8673557
https://pocoproject.org/slides/090-NotificationsEvents.pdf