MEL
Microthread & Execution library
CallbackSubscriptor_Impl.h
1 /*
2  * SPDX-FileCopyrightText: 2005,2022 Daniel Barrientos
3  * <danivillamanin@gmail.com>
4  *
5  * SPDX-License-Identifier: MIT
6  */
7 namespace mel
8 {
9  namespace core
10  {
14 #if VARIABLE_NUM_ARGS == VARIABLE_MAX_ARGS
16  namespace _private
17  {
18  template <bool> struct _CriticalSectionWrapper
19  {
20  std::mutex mSC;
21  };
22  template <bool enabled> struct _Lock
23  {
24  _Lock( _CriticalSectionWrapper<enabled>& cs ) : mLck( cs.mSC )
25  {
26  }
27 
28  private:
29  std::scoped_lock<std::mutex> mLck;
30  };
31  template <> struct _CriticalSectionWrapper<false>
32  {
33  private:
34  };
35  template <> struct _Lock<false>
36  {
37  _Lock( _CriticalSectionWrapper<false>& cs ) {}
38  };
39  } // namespace _private
41  template <class ThreadingPolicy, VARIABLE_ARGS>
42  class CallbackSubscriptor_Base
43  {
44  public:
45  typedef Callback<ECallbackResult, VARIABLE_ARGS_DECL> CallbackType;
46 
47  protected:
48  struct CallbackInfo
49  {
50  std::shared_ptr<CallbackType> cb;
51  int id; // callback id to use on unsubscriptionr when
52  // needed/neccesary
53  CallbackInfo( std::shared_ptr<CallbackType> acb, int aId )
54  : cb( acb ), id( aId )
55  {
56  }
57  CallbackInfo() : cb( nullptr ), id( -1 ) {}
58  };
59  static int mCurrId;
60  bool mTriggering; // to detect subscription/unsubscription while
61  // triggering
62  struct PendingOperation
63  {
64  enum class EOperation
65  {
66  O_SUBSCRIPTION,
67  O_UNSUBSCRIPTION
68  } op;
69  SubscriptionEmplacement emplacement; // only for subscription
70  CallbackInfo info;
71  };
72  std::deque<PendingOperation> mPendingOperation;
73 
74  public:
75  CallbackSubscriptor_Base() : /*mCurrId(0),*/ mTriggering( false ){};
76  typedef list<CallbackInfo> CallbackListType;
77  virtual ~CallbackSubscriptor_Base() { removeCallbacks(); }
78 
79  inline size_t getNumCallbacks() const { return mCallbacks.size(); }
80  void removeCallbacks()
81  {
82  _private::_Lock<mpl::isSame<
83  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
84  lck( mSC );
85  mCallbacks.clear();
86  }
87  template <class U>
88  int subscribeCallback( U&& callback,
89  SubscriptionEmplacement se = SE_BACK )
90  {
91  return subscribeCreatedCallback(
92  new CallbackType( std::forward<U>( callback ),
93  ::mel::core::use_functor ),
94  se );
95  }
96  int subscribeCallback(
97  std::function<ECallbackResult( VARIABLE_ARGS_DECL )>&& callback,
98  SubscriptionEmplacement se = SE_BACK )
99  {
100  return subscribeCreatedCallback(
101  new CallbackType( std::move( callback ),
102  ::mel::core::use_function ),
103  se );
104  }
105  int subscribeCallback(
106  const std::function<ECallbackResult( VARIABLE_ARGS_DECL )>&
107  callback,
108  SubscriptionEmplacement se = SE_BACK )
109  {
110  return subscribeCreatedCallback(
111  new CallbackType( callback, ::mel::core::use_function ),
112  se );
113  }
114  int subscribeCallback(
115  std::function<ECallbackResult( VARIABLE_ARGS_DECL )>& callback,
116  SubscriptionEmplacement se = SE_BACK )
117  {
118  return subscribeCreatedCallback(
119  new CallbackType( callback, ::mel::core::use_function ),
120  se );
121  }
127  bool unsubscribeCallback(
128  std::function<ECallbackResult( VARIABLE_ARGS_DECL )>&&
129  callback ) = delete;
130  template <class U> bool unsubscribeCallback( U&& callback )
131  {
132  _private::_Lock<mpl::isSame<
133  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
134  lck( mSC );
135  CallbackType* auxiliar = new CallbackType(
136  ::std::forward<U>( callback ), ::mel::core::use_functor );
137  return unsubscribeCreatedCallback(
138  std::shared_ptr<CallbackType>( auxiliar ) );
139  }
140  bool unsubscribeCallback( int id )
141  {
142  bool result = false;
143  //@todo revisar locks
144  _private::_Lock<mpl::isSame<
145  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
146  lck( mSC );
147  if ( mTriggering )
148  {
149  // spdlog::debug("CallbackSubscriptor
150  //Callbacks are being triggered while unsubscribing!!");
151  for ( typename CallbackListType::iterator
152  i = mCallbacks.begin(),
153  j = mCallbacks.end();
154  i != j; ++i )
155  {
156  if ( i->id == id )
157  {
158  PendingOperation po;
159  po.op =
160  PendingOperation::EOperation::O_UNSUBSCRIPTION;
161  po.info = std::move( CallbackInfo( nullptr, id ) );
162  mPendingOperation.push_back( std::move( po ) );
163  result = true;
164  break;
165  }
166  }
167  }
168  else
169  {
170  for ( typename CallbackListType::iterator
171  i = mCallbacks.begin(),
172  j = mCallbacks.end();
173  i != j; ++i )
174  {
175  if ( i->id == id )
176  {
177  mCallbacks.erase( i );
178  result = true;
179  break;
180  }
181  }
182  }
183  return result;
184  }
189  int subscribeCreatedCallback( CallbackType* cb,
190  SubscriptionEmplacement se = SE_BACK )
191  {
192  int result;
193  _private::_Lock<mpl::isSame<
194  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
195  lck( mSC );
196  result = ++mCurrId;
197  if ( mTriggering )
198  {
199  // spdlog::debug("CallbackSubscriptor
200  //Callbacks are being triggered while subscribing!!");
201  PendingOperation po;
202  po.op = PendingOperation::EOperation::O_SUBSCRIPTION;
203  po.info = std::move( CallbackInfo(
204  std::shared_ptr<CallbackType>( cb ), result ) );
205  po.emplacement = se;
206  mPendingOperation.push_back( std::move( po ) );
207  }
208  else
209  {
210  if ( se == SE_BACK )
211  {
212  mCallbacks.push_back( CallbackInfo(
213  std::shared_ptr<CallbackType>( cb ), result ) );
214  }
215  else
216  {
217  mCallbacks.push_front( CallbackInfo(
218  std::shared_ptr<CallbackType>( cb ), result ) );
219  }
220  }
221  return result;
222  }
229  bool unsubscribeCreatedCallback( std::shared_ptr<CallbackType> cb )
230  {
231  bool result = false;
232  _private::_Lock<mpl::isSame<
233  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
234  lck( mSC );
235  if ( mTriggering )
236  {
237 
238  // spdlog::debug("CallbackSubscriptor
239  //Callbacks are being triggered while unsubscribing!!");
240  typename CallbackListType::iterator i = mCallbacks.begin();
241  while ( i != mCallbacks.end() )
242  {
243  if ( *i->cb == *cb )
244  {
245  PendingOperation po;
246  po.op =
247  PendingOperation::EOperation::O_UNSUBSCRIPTION;
248  po.info = std::move( CallbackInfo(
249  std::shared_ptr<CallbackType>( cb ), -1 ) );
250  mPendingOperation.push_back( std::move( po ) );
251  result = true;
252  break;
253  }
254  else
255  {
256  ++i;
257  }
258  }
259  }
260  else
261  {
262  typename CallbackListType::iterator i = mCallbacks.begin();
263  while ( i != mCallbacks.end() )
264  {
265  if ( *i->cb == *cb )
266  {
267  mCallbacks.erase( i );
268  i = mCallbacks.end();
269  result = true;
270  break;
271  }
272  else
273  {
274  ++i;
275  }
276  }
277  }
278  return result;
279  }
280  void
281  append( const CallbackSubscriptor_Base<ThreadingPolicy,
282  VARIABLE_ARGS_DECL>& ob2 )
283  {
284  for ( auto& cb : ob2.mCallbacks )
285  mCallbacks.push_back( cb );
286  }
287  void append( CallbackSubscriptor_Base<ThreadingPolicy,
288  VARIABLE_ARGS_DECL>&& ob2 )
289  {
290  for ( auto& cb : ob2.mCallbacks )
291  mCallbacks.push_back( std::move( cb ) );
292  }
293  inline CallbackListType& getCallbacks() { return mCallbacks; }
294  inline const CallbackListType& getCallbacks() const
295  {
296  return mCallbacks;
297  }
298 
299  protected:
300  CallbackListType mCallbacks;
301  _private::_CriticalSectionWrapper<mpl::isSame<
302  ThreadingPolicy, mel::core::CSMultithreadPolicy>::result>
303  mSC;
304 
305  CallbackSubscriptor_Base( const CallbackSubscriptor_Base& o2 )
306  : mTriggering( false )
307  {
308  _private::_Lock<mpl::isSame<
309  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
310  lck( mSC );
311  for ( typename CallbackListType::iterator
312  i = o2.mCallbacks.begin(),
313  j = o2.mCallbacks.end();
314  i != j; ++i )
315  mCallbacks.push_back(
316  CallbackInfo( ( *i ).cb->clone(), ++mCurrId ) );
317  }
318  void _applyPendingOperations()
319  {
320  for ( auto& p : mPendingOperation )
321  {
322  switch ( p.op )
323  {
324  case PendingOperation::EOperation::O_SUBSCRIPTION:
325  if ( p.emplacement == SE_BACK )
326  {
327  mCallbacks.push_back( p.info );
328  }
329  else
330  {
331  mCallbacks.push_front( p.info );
332  }
333  break;
334  case PendingOperation::EOperation::O_UNSUBSCRIPTION:
335  if ( p.info.cb == nullptr ) // use id for unsubscription
336  {
337  unsubscribeCallback( p.info.id );
338  }
339  else
340  {
341  unsubscribeCreatedCallback( p.info.cb );
342  }
343  }
344  }
345  mPendingOperation.clear();
346  }
347  };
348  template <class ThreadingPolicy, VARIABLE_ARGS_NODEFAULT>
349  int CallbackSubscriptor_Base<ThreadingPolicy,
350  VARIABLE_ARGS_DECL>::mCurrId = 0;
351 
352  template <class ThreadingPolicy, VARIABLE_ARGS>
353  class CallbackSubscriptorNotTyped
354  : public CallbackSubscriptor_Base<ThreadingPolicy,
355  VARIABLE_ARGS_DECL>
356  {
357  typedef CallbackSubscriptor_Base<ThreadingPolicy,
358  VARIABLE_ARGS_DECL>
359  BaseType;
360 
361  public:
362  void triggerCallbacks( VARIABLE_ARGS_IMPL )
363  {
364  if ( BaseType::mTriggering )
365  {
366  // spdlog::debug("CallbackSubscriptor
367  //Callbacks are being triggered while triggering again!!");
368  return;
369  }
370  _private::_Lock<mpl::isSame<
371  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
372  lck( BaseType::mSC );
373  BaseType::mTriggering = true;
374  typename BaseType::CallbackListType::iterator i =
375  BaseType::mCallbacks.begin();
376  auto j = BaseType::mCallbacks.end();
377  while ( i != j )
378  {
379  if ( ( *i->cb )( VARIABLE_ARGS_USE ) ==
380  ECallbackResult::UNSUBSCRIBE )
381  {
382  i = BaseType::mCallbacks.erase( i );
383  }
384  else
385  {
386  ++i;
387  }
388  }
389  BaseType::mTriggering = false;
390  if ( !BaseType::mPendingOperation.empty() )
391  BaseType::_applyPendingOperations();
392  }
393 
394  protected:
395  CallbackSubscriptorNotTyped() : BaseType() {}
396  CallbackSubscriptorNotTyped( const CallbackSubscriptorNotTyped& o2 )
397  : BaseType( o2 )
398  {
399  }
400  };
401 
403  template <class ThreadingPolicy>
404  class CallbackSubscriptorNotTyped<ThreadingPolicy, void>
405  : public CallbackSubscriptor_Base<ThreadingPolicy, void>
406  {
407  typedef CallbackSubscriptor_Base<ThreadingPolicy, void> BaseType;
408 
409  public:
410  void triggerCallbacks()
411  {
412  _private::_Lock<mpl::isSame<
413  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
414  lck( BaseType::mSC );
415  if ( BaseType::mTriggering )
416  {
417  // spdlog::debug("CallbackSubscriptor Callbacks are being
418  // triggered while triggering again!!");
419  return;
420  }
421  BaseType::mTriggering = true;
422 
423  typename BaseType::CallbackListType::iterator i =
424  BaseType::mCallbacks.begin();
425  auto j = BaseType::mCallbacks.end();
426  while ( i != j )
427  {
428  if ( ( *i->cb )() == ECallbackResult::UNSUBSCRIBE )
429  {
430  i = BaseType::mCallbacks.erase( i );
431  }
432  else
433  {
434  ++i;
435  }
436  }
437  BaseType::mTriggering = false;
438  if ( !BaseType::mPendingOperation.empty() )
439  BaseType::_applyPendingOperations();
440  }
441 
442  protected:
443  CallbackSubscriptorNotTyped() : BaseType() {}
444  CallbackSubscriptorNotTyped( const CallbackSubscriptorNotTyped& o2 )
445  : BaseType( o2 )
446  {
447  }
448  };
449 
450  template <class ThreadingPolicy, VARIABLE_ARGS>
451  class CallbackSubscriptor
452  : public CallbackSubscriptorNotTyped<ThreadingPolicy,
453  VARIABLE_ARGS_DECL>
454  {
455  typedef CallbackSubscriptorNotTyped<ThreadingPolicy,
456  VARIABLE_ARGS_DECL>
457  BaseType;
458 
459  public:
460  CallbackSubscriptor() : BaseType() {}
461  CallbackSubscriptor( const CallbackSubscriptor& o2 )
462  : BaseType( o2 )
463  {
464  }
465  };
466  // specialization for void arguments
467  template <class ThreadingPolicy>
468  class CallbackSubscriptor<ThreadingPolicy, void>
469  : public CallbackSubscriptorNotTyped<ThreadingPolicy, void>
470  {
471  typedef CallbackSubscriptorNotTyped<ThreadingPolicy, void> BaseType;
472 
473  public:
474  CallbackSubscriptor() : BaseType() {}
475  CallbackSubscriptor( const CallbackSubscriptor& o2 )
476  : BaseType( o2 )
477  {
478  }
479  int subscribeCallback( std::function<ECallbackResult()>&& callback,
480  SubscriptionEmplacement se = SE_BACK )
481  {
482  return BaseType::subscribeCreatedCallback(
483  new typename BaseType::CallbackType(
484  std::move( callback ), ::mel::core::use_function ),
485  se );
486  }
487  int
488  subscribeCallback( const std::function<ECallbackResult()>& callback,
489  SubscriptionEmplacement se = SE_BACK )
490  {
491  return BaseType::subscribeCreatedCallback(
492  new typename BaseType::CallbackType(
493  callback, ::mel::core::use_function ),
494  se );
495  }
496  int subscribeCallback( std::function<ECallbackResult()>& callback,
497  SubscriptionEmplacement se )
498  {
499  return BaseType::subscribeCreatedCallback(
500  new typename BaseType::CallbackType(
501  callback, ::mel::core::use_function ),
502  se );
503  }
504  template <class U>
505  int subscribeCallback( U&& callback,
506  SubscriptionEmplacement se = SE_BACK )
507  {
508  return BaseType::subscribeCallback(
509  ::std::forward<U>( callback ), se );
510  }
516  bool unsubscribeCallback(
517  std::function<ECallbackResult()>&& callback ) = delete;
518  template <class U> bool unsubscribeCallback( U&& callback )
519  {
520  return BaseType::unsubscribeCallback(
521  ::std::forward<U>( callback ) );
522  }
523  bool unsubscribeCallback( int id )
524  {
525  return BaseType::unsubscribeCallback( id );
526  }
527  };
528 
529 #else
530  template <class ThreadingPolicy, VARIABLE_ARGS>
531  class CallbackSubscriptor_Base<ThreadingPolicy, VARIABLE_ARGS_DECL,
532  void>
533  {
534  public:
535  typedef Callback<ECallbackResult, VARIABLE_ARGS_DECL> CallbackType;
536 
537  protected:
538  struct CallbackInfo
539  {
540  std::shared_ptr<CallbackType> cb;
541  int id; // callback id to use on unsubscriptionr when
542  // needed/neccesary
543  CallbackInfo( std::shared_ptr<CallbackType> acb, int aId )
544  : cb( acb ), id( aId )
545  {
546  }
547  CallbackInfo() : cb( nullptr ), id( -1 ) {}
548  };
549  static int mCurrId;
550  bool mTriggering; // to detect subscription/unsubscription while
551  // triggering
552  struct PendingOperation
553  {
554  enum class EOperation
555  {
556  O_SUBSCRIPTION,
557  O_UNSUBSCRIPTION
558  } op;
559  SubscriptionEmplacement emplacement; // only for subscription
560  CallbackInfo info;
561  };
562  std::deque<PendingOperation> mPendingOperation;
563 
564  public:
565  typedef list<CallbackInfo> CallbackListType;
566  CallbackSubscriptor_Base() : /*mCurrId(0),*/ mTriggering( false ){};
567  virtual ~CallbackSubscriptor_Base() { removeCallbacks(); }
568  inline size_t getNumCallbacks() const { return mCallbacks.size(); }
569  void removeCallbacks()
570  {
571  _private::_Lock<mpl::isSame<
572  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
573  lck( mSC );
574  mCallbacks.clear();
575  }
576 
577  template <class U>
578  int subscribeCallback( U&& callback,
579  SubscriptionEmplacement se = SE_BACK )
580  {
581  return subscribeCreatedCallback(
582  new CallbackType( ::std::forward<U>( callback ),
583  ::mel::core::use_functor ),
584  se );
585  }
586 
592  int subscribeCreatedCallback( CallbackType* cb,
593  SubscriptionEmplacement se = SE_BACK )
594  {
595  int result;
596  _private::_Lock<mpl::isSame<
597  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
598  lck( mSC );
599  result = ++mCurrId;
600  if ( mTriggering )
601  {
602  // spdlog::debug("CallbackSubscriptor Callbacks are being
603  // triggered
604  // while triggering subscribing!!");
605  PendingOperation po;
606  po.op = PendingOperation::EOperation::O_SUBSCRIPTION;
607  po.info = std::move( CallbackInfo(
608  std::shared_ptr<CallbackType>( cb ), result ) );
609  po.emplacement = se;
610  mPendingOperation.push_back( std::move( po ) );
611  }
612  else
613  {
614  if ( se == SE_BACK )
615  {
616  mCallbacks.push_back( CallbackInfo(
617  std::shared_ptr<CallbackType>( cb ), result ) );
618  }
619  else
620  {
621  mCallbacks.push_front( CallbackInfo(
622  std::shared_ptr<CallbackType>( cb ), result ) );
623  }
624  }
625  return result;
626  }
627 
628  template <class U> bool unsubscribeCallback( U&& callback )
629  {
630  bool result = false;
631  _private::_Lock<mpl::isSame<
632  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
633  lck( mSC );
634  CallbackType* auxiliar = new CallbackType(
635  std::forward<U>( callback ), ::mel::core::use_functor );
636  result = unsubscribeCreatedCallback(
637  std::shared_ptr<CallbackType>( auxiliar ) );
638  return result;
639  }
640  bool unsubscribeCallback( int id )
641  {
642  bool result = false;
643  _private::_Lock<mpl::isSame<
644  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
645  lck( mSC );
646 
647  if ( mTriggering )
648  {
649  // spdlog::debug("CallbackSubscriptor Callbacks are being
650  // triggered while triggering unsubscribing!!");
651  for ( typename CallbackListType::iterator
652  i = mCallbacks.begin(),
653  j = mCallbacks.end();
654  i != j; ++i )
655  {
656  if ( i->id == id )
657  {
658  PendingOperation po;
659  po.op =
660  PendingOperation::EOperation::O_UNSUBSCRIPTION;
661  po.info = std::move( CallbackInfo( nullptr, id ) );
662  mPendingOperation.push_back( std::move( po ) );
663  result = true;
664  break;
665  }
666  }
667  }
668  else
669  {
670  for ( typename CallbackListType::iterator
671  i = mCallbacks.begin(),
672  j = mCallbacks.end();
673  i != j; ++i )
674  {
675  if ( i->id == id )
676  {
677  mCallbacks.erase( i );
678  result = true;
679  break;
680  }
681  }
682  }
683  return result;
684  }
685 
690  bool unsubscribeCreatedCallback( std::shared_ptr<CallbackType> cb )
691  {
692  bool result = false;
693  _private::_Lock<mpl::isSame<
694  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
695  lck( mSC );
696  if ( mTriggering )
697  {
698  // spdlog::debug("CallbackSubscriptor Callbacks are being
699  // triggered while triggering unsubscribing!!");
700  typename CallbackListType::iterator i = mCallbacks.begin();
701  while ( i != mCallbacks.end() )
702  {
703  if ( *i->cb == *cb )
704  {
705  PendingOperation po;
706  po.op =
707  PendingOperation::EOperation::O_UNSUBSCRIPTION;
708  po.info = std::move( CallbackInfo(
709  std::shared_ptr<CallbackType>( cb ), -1 ) );
710  mPendingOperation.push_back( std::move( po ) );
711  result = true;
712  break;
713  }
714  else
715  {
716  ++i;
717  }
718  }
719  }
720  else
721  {
722  typename CallbackListType::iterator i = mCallbacks.begin();
723  while ( i != mCallbacks.end() )
724  {
725  if ( *i->cb == *cb )
726  {
727  mCallbacks.erase( i );
728  i = mCallbacks.end();
729  result = true;
730  break;
731  }
732  else
733  {
734  ++i;
735  }
736  }
737  }
738  return result;
739  }
740  void
741  append( const CallbackSubscriptor_Base<ThreadingPolicy,
742  VARIABLE_ARGS_DECL>& ob2 )
743  {
744  for ( auto& cb : ob2.mCallbacks )
745  mCallbacks.push_back( cb );
746  }
747  void append( CallbackSubscriptor_Base<ThreadingPolicy,
748  VARIABLE_ARGS_DECL>&& ob2 )
749  {
750  for ( auto& cb : ob2.mCallbacks )
751  mCallbacks.push_back( std::move( cb ) );
752  }
753  inline CallbackListType& getCallbacks() { return mCallbacks; }
754  inline const CallbackListType& getCallbacks() const
755  {
756  return mCallbacks;
757  }
758 
759  protected:
760  CallbackListType mCallbacks;
761  _private::_CriticalSectionWrapper<mpl::isSame<
762  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
763  mSC;
764 
765  CallbackSubscriptor_Base( const CallbackSubscriptor_Base& o2 )
766  : mTriggering( false )
767  {
768  _private::_Lock<mpl::isSame<
769  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
770  lck( mSC );
771  typename CallbackListType::const_iterator i, j;
772  for ( i = o2.mCallbacks.begin(), j = o2.mCallbacks.end();
773  i != j; ++i )
774  mCallbacks.push_back( CallbackInfo(
775  std::shared_ptr<CallbackType>( ( *i ).cb->clone() ),
776  ++mCurrId ) );
777  }
778  void _applyPendingOperations()
779  {
780  for ( auto& p : mPendingOperation )
781  {
782  switch ( p.op )
783  {
784  case PendingOperation::EOperation::O_SUBSCRIPTION:
785  if ( p.emplacement == SE_BACK )
786  {
787  mCallbacks.push_back( p.info );
788  }
789  else
790  {
791  mCallbacks.push_front( p.info );
792  }
793  break;
794  case PendingOperation::EOperation::O_UNSUBSCRIPTION:
795  if ( p.info.cb == nullptr ) // use id for unsubscription
796  {
797  unsubscribeCallback( p.info.id );
798  }
799  else
800  {
801  unsubscribeCreatedCallback( p.info.cb );
802  }
803  }
804  }
805  mPendingOperation.clear();
806  }
807  };
808  template <class ThreadingPolicy, VARIABLE_ARGS>
809  int CallbackSubscriptor_Base<ThreadingPolicy,
810  VARIABLE_ARGS_DECL>::mCurrId = 0;
811 
812  template <class ThreadingPolicy, VARIABLE_ARGS>
813  class CallbackSubscriptorNotTyped<ThreadingPolicy, VARIABLE_ARGS_DECL>
814  : public CallbackSubscriptor_Base<ThreadingPolicy,
815  VARIABLE_ARGS_DECL>
816  {
817  typedef CallbackSubscriptor_Base<ThreadingPolicy,
818  VARIABLE_ARGS_DECL>
819  BaseType;
820 
821  public:
822  void triggerCallbacks( VARIABLE_ARGS_IMPL )
823  {
824  if ( BaseType::mTriggering )
825  {
826  mel::text::error( "CallbackSubscriptor Callbacks are being "
827  "triggered while triggering again!!" );
828  return;
829  }
830  _private::_Lock<mpl::isSame<
831  ThreadingPolicy, ::mel::core::CSMultithreadPolicy>::result>
832  lck( BaseType::mSC );
833  BaseType::mTriggering = true;
834  typename BaseType::CallbackListType::iterator i =
835  BaseType::mCallbacks.begin();
836  while ( i != BaseType::mCallbacks.end() )
837  {
838  if ( ( *i->cb )( VARIABLE_ARGS_USE ) ==
839  ECallbackResult::UNSUBSCRIBE )
840  {
841  i = BaseType::mCallbacks.erase( i );
842  }
843  else
844  {
845  ++i;
846  }
847  }
848  BaseType::mTriggering = false;
849  if ( !BaseType::mPendingOperation.empty() )
850  BaseType::_applyPendingOperations();
851  }
852 
853  protected:
854  CallbackSubscriptorNotTyped() : BaseType() {}
855  CallbackSubscriptorNotTyped( const CallbackSubscriptorNotTyped& o2 )
856  : BaseType( o2 )
857  {
858  }
859  };
860 
861  template <class ThreadingPolicy, VARIABLE_ARGS>
862  class CallbackSubscriptor<ThreadingPolicy, VARIABLE_ARGS_DECL, void>
863  : public CallbackSubscriptorNotTyped<ThreadingPolicy,
864  VARIABLE_ARGS_DECL, void>
865  {
866  typedef CallbackSubscriptorNotTyped<ThreadingPolicy,
867  VARIABLE_ARGS_DECL, void>
868  BaseType;
869 
870  public:
871  CallbackSubscriptor() : BaseType() {}
872  CallbackSubscriptor( const CallbackSubscriptor& o2 )
873  : BaseType( o2 )
874  {
875  }
876  int subscribeCallback(
877  std::function<ECallbackResult( VARIABLE_ARGS_DECL )>&& callback,
878  SubscriptionEmplacement se = SE_BACK )
879  {
880  return BaseType::subscribeCreatedCallback(
881  new typename BaseType::CallbackType(
882  std::move( callback ), ::mel::core::use_function ),
883  se );
884  }
885  int subscribeCallback(
886  const std::function<ECallbackResult( VARIABLE_ARGS_DECL )>&
887  callback,
888  SubscriptionEmplacement se = SE_BACK )
889  {
890  return BaseType::subscribeCreatedCallback(
891  new typename BaseType::CallbackType(
892  callback, ::mel::core::use_function ),
893  se );
894  }
895  int subscribeCallback(
896  std::function<ECallbackResult( VARIABLE_ARGS_DECL )>& callback,
897  SubscriptionEmplacement se = SE_BACK )
898  {
899  return BaseType::subscribeCreatedCallback(
900  new typename BaseType::CallbackType(
901  callback, ::mel::core::use_function ),
902  se );
903  }
904 
905  template <class U>
906  int subscribeCallback( U&& functor,
907  SubscriptionEmplacement se = SE_BACK )
908  {
909  return BaseType::subscribeCallback( std::forward<U>( functor ),
910  se );
911  }
917  bool unsubscribeCallback(
918  std::function<ECallbackResult( VARIABLE_ARGS_DECL )>&&
919  callback ) = delete;
920  // {
921  // //MPL_STATIC_ASSERT(true, PEPE)
922  // //static_assert(false,"can't unsubscribe std::function<>.
923  // unsubscribe by
924  // id or use mpl function utilities") return false; //can't
925  // unsubscribe function
926  // }
927  template <class U> bool unsubscribeCallback( U&& callback )
928  {
929  return BaseType::unsubscribeCallback( callback );
930  }
931  bool unsubscribeCallback( int id )
932  {
933  return BaseType::unsubscribeCallback( id );
934  }
935  };
936 #endif
937  } // namespace core
938 } // namespace mel
mel::core::Callback
create Callback from callable
Definition: Callback_Impl.h:142
mel::core::CallbackSubscriptor< ThreadingPolicy, void >::unsubscribeCallback
bool unsubscribeCallback(std::function< ECallbackResult()> &&callback)=delete
mel::core::CallbackSubscriptor_Base
callback subscription functionalit
Definition: CallbackSubscriptor_Impl.h:43
mel::core::CallbackSubscriptor_Base::unsubscribeCallback
bool unsubscribeCallback(std::function< ECallbackResult(VARIABLE_ARGS_DECL)> &&callback)=delete
mel::core::CallbackSubscriptor_Base::unsubscribeCreatedCallback
bool unsubscribeCreatedCallback(std::shared_ptr< CallbackType > cb)
Definition: CallbackSubscriptor_Impl.h:229
mel::core::CallbackSubscriptor_Base::subscribeCreatedCallback
int subscribeCreatedCallback(CallbackType *cb, SubscriptionEmplacement se=SE_BACK)
Definition: CallbackSubscriptor_Impl.h:189
mel::core::CallbackSubscriptor
Definition: CallbackSubscriptor_Impl.h:454
mel::core::CallbackSubscriptorNotTyped< ThreadingPolicy, void >
Definition: CallbackSubscriptor_Impl.h:406
mel::core::CallbackSubscriptorNotTyped
Definition: CallbackSubscriptor_Impl.h:356
mel::core::ECallbackResult
ECallbackResult
Type resturned by callbacks subscribed to CallbackSubscriptors.
Definition: CallbackSubscriptor.h:32
mel
Definition: Callback_Impl.h:11
mel::core::CSMultithreadPolicy
Definition: CallbackSubscriptor.h:37
mel::core::CallbackSubscriptor_Base::CallbackInfo
Definition: CallbackSubscriptor_Impl.h:49
mel::core::CallbackSubscriptor_Base::PendingOperation
Definition: CallbackSubscriptor_Impl.h:63
mel::mpl::isSame
Definition: IsSame.h:12