eventbus.d

   1 /* Invisible Vector Library
   2  *
   3  * This program is free software: you can redistribute it and/or modify
   4  * it under the terms of the GNU General Public License as published by
   5  * the Free Software Foundation, version 3 of the License ONLY.
   6  *
   7  * This program is distributed in the hope that it will be useful,
   8  * but WITHOUT ANY WARRANTY; without even the implied warranty of
   9  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10  * GNU General Public License for more details.
  11  *
  12  * You should have received a copy of the GNU General Public License
  13  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  14  */
  15 module iv.eventbus /*is aliced*/;
  16 private:
  17
  18 import core.time : MonoTime;
  19 import std.datetime : Clock, SysTime;
  20
  21 import iv.alice;
  22 import iv.weakref;
  23
  24
  25 // ////////////////////////////////////////////////////////////////////////// //
  26 ///WARNING! changing this is NOT thread-safe!
  27 ///         you should set this up at program startup and don't touch after that
  28 __gshared void delegate (const(char)[] msg) ebusLogError;
  29
  30
  31 /// objects should implement this interface to make sinking/bubbling work.
  32 /// if this interface is not implemented, targeted message will be dropped on the floor.
  33 public interface EventTarget {
  34   /// this should return parent object or null
  35   abstract @property Object eventbusParent ();
  36   /// this will be called on sinking and bubbling
  37   abstract void eventbusOnEvent (Event evt);
  38 }
  39
  40
  41 // ////////////////////////////////////////////////////////////////////////// //
  42 /** note that when event reached it's destination, it will be switched to bubbling
  43   * phase before calling the appropriate event handler. */
  44 public class Event {
  45 public:
  46   /// propagation flags
  47   enum PFlags : ubyte {
  48     Eaten     = 1U<<0, /// event is processed, but not cancelled
  49     Cancelled = 1U<<1, /// event is cancelled (it may be *both* processed and cancelled!)
  50     Bubbling  = 1U<<2, /// event is in bubbling phase
  51     Posted    = 1U<<7, /// event is posted
  52   }
  53   protected Object osource; /// event source, can be null
  54   protected Object odest; /// event destination, can be null for broadcast events
  55   protected ubyte flags; /// see PFlags
  56   protected SysTime etime; /// emiting time
  57
  58   this () { etime = Clock.currTime; } ///
  59   this (Object asrc) { etime = Clock.currTime; osource = asrc; } ///
  60   this (Object asrc, Object adest) { etime = Clock.currTime; osource = asrc; odest = adest; } ///
  61
  62   /// post event: put it into queue to process at next iteration.
  63   final void post () { this.postpone(0); }
  64
  65   /// schedule event to be processed later (after `msecs` msecs passed)
  66   final void later (int msecs) { this.postpone(msecs); }
  67
  68 final pure nothrow @safe @nogc:
  69   void eat () { pragma(inline, true); flags |= PFlags.Eaten; } /// "eat" event (set "eaten" flag)
  70   void cancel () { pragma(inline, true); flags |= PFlags.Cancelled; } /// "cancel" event (set "cancelled" flag)
  71
  72   inout(Object) source () inout { pragma(inline, true); return osource; } /// source object
  73   inout(Object) dest () inout { pragma(inline, true); return odest; } /// destination object
  74   inout(SysTime) time () inout { pragma(inline, true); return etime; } /// emitting time
  75
  76 const @property:
  77   bool eaten () { pragma(inline, true); return ((flags&PFlags.Eaten) != 0); } ///
  78   bool cancelled () { pragma(inline, true); return ((flags&PFlags.Cancelled) != 0); } ///
  79   bool processed () { pragma(inline, true); return ((flags&(PFlags.Eaten|PFlags.Cancelled)) != 0); } ///
  80   bool posted () { pragma(inline, true); return ((flags&PFlags.Posted) != 0); } ///
  81   bool sinking () { pragma(inline, true); return ((flags&PFlags.Bubbling) == 0); } ///
  82   bool bubbling () { pragma(inline, true); return ((flags&PFlags.Bubbling) != 0); } ///
  83 }
  84
  85
  86 // ////////////////////////////////////////////////////////////////////////// //
  87 /** register listener for all events (both targeted and not, and for anonymous too).
  88  * returns event id that can be used to remove listener. */
  89 public uint addEventListener(E:Event) (void delegate (E evt) dg, bool oneshot=false) {
  90   return addEventListener!E(null, dg, oneshot);
  91 }
  92
  93 /** register event listener for *source* object `obj`.
  94  * returns event id that can be used to remove listener. */
  95 public uint addEventListener(E:Event) (Object obj, void delegate (E evt) dg, bool oneshot=false) {
  96   if (dg is null) return 0;
  97   synchronized (Event.classinfo) {
  98     if (!oneshot) {
  99       foreach (ref EventListenerInfo eli; llist) {
 100         if (typeid(E) == eli.ti && eli.dg is cast(EventListenerInfo.DgType)dg) return eli.id;
 101       }
 102     }
 103     llist ~= EventListenerInfo(obj, typeid(E), cast(EventListenerInfo.DgType)dg, oneshot);
 104     return lastid;
 105   }
 106 }
 107
 108
 109 // ////////////////////////////////////////////////////////////////////////// //
 110 /// remove event listener with the given id. returns `true` if some listener was removed.
 111 public bool removeEventListener (uint id) {
 112   synchronized (Event.classinfo) {
 113     if (id !in usedIds) return false;
 114     usedIds.remove(id);
 115     foreach (ref EventListenerInfo eli; llist) {
 116       if (eli.id == id) {
 117         needListenerCleanup = true;
 118         eli.id = 0;
 119         eli.dg = null;
 120         eli.dobj = null;
 121         eli.xsrc = null;
 122         return true;
 123       }
 124     }
 125     return false;
 126   }
 127 }
 128
 129
 130 // ////////////////////////////////////////////////////////////////////////// //
 131 enum SBDispatchMixin = q{
 132   static if (is(typeof(&__traits(getMember, this, "onMyEvent")) == delegate) ||
 133              is(typeof(&__traits(getMember, this, "onMyEvent")) == function))
 134   {
 135     {
 136       //if (this !is evt.odest) return; // not our
 137       void doit (TypeInfo_Class ti) {
 138         import std.traits;
 139         if (typeid(Event) != ti) doit(ti.base);
 140         if (evt.processed) return;
 141         foreach (immutable oidx, /*auto*/ over; __traits(getOverloads, this, "onMyEvent")) {
 142           alias tt = typeof(over);
 143           static if (is(ReturnType!tt == void)) {
 144             alias pars = Parameters!tt;
 145             static if (pars.length == 1) {
 146               static if (is(pars[0] : Event)) {
 147                 if (typeid(pars[0]) == ti) {
 148                   if (auto ev = cast(pars[0])evt) {
 149                     __traits(getOverloads, this, "onMyEvent")[oidx](ev);
 150                     return;
 151                   }
 152                 }
 153               }
 154             }
 155           }
 156         }
 157       }
 158       if (evt !is null && !evt.processed && evt.bubbling && evt.odest is this) {
 159         doit(typeid(evt));
 160       }
 161     }
 162   }
 163 };
 164
 165
 166 private class ObjDispatcher {
 167   enum Type { Broadcast, My, Sink, Bubble }
 168   abstract bool alive ();
 169   abstract Type type ();
 170   abstract bool isObject (Object o);
 171   abstract void dispatch (Event ev);
 172 }
 173
 174
 175 /** this should be called in object ctor to automatically connect event listening methods.
 176  *
 177  * valid handler methods:
 178  *
 179  * ---------------
 180  *   void onEvent (EventXXX ev); // will receive any events
 181  *   void onMyEvent (EventXXX ev); // will receive only events targeted to this object (dest is this), when they sinked
 182  *   void onSinkEvent (EventXXX ev); // will receive sinking events (but not "my")
 183  *   void onBubbleEvent (EventXXX ev); // will receive bubbling events (but not "my")
 184  * ---------------
 185  */
 186 public void connectListeners(O:Object) (O obj) {
 187   if (obj is null) return;
 188
 189   void addListener(ObjDispatcher.Type tp) () {
 190          static if (tp == ObjDispatcher.Type.Broadcast) enum EvtName = "onEvent";
 191     else static if (tp == ObjDispatcher.Type.My) enum EvtName = "onMyEvent";
 192     else static if (tp == ObjDispatcher.Type.Sink) enum EvtName = "onSinkEvent";
 193     else static if (tp == ObjDispatcher.Type.Bubble) enum EvtName = "onBubbleEvent";
 194     else static assert(0, "invalid type");
 195
 196     static if (is(typeof(&__traits(getMember, obj, EvtName)) == delegate) ||
 197                is(typeof(&__traits(getMember, obj, EvtName)) == function))
 198     {
 199       static class ObjDispatcherX(OT, ObjDispatcher.Type tp) : ObjDispatcher {
 200              static if (tp == ObjDispatcher.Type.Broadcast) enum EvtName = "onEvent";
 201         else static if (tp == ObjDispatcher.Type.My) enum EvtName = "onMyEvent";
 202         else static if (tp == ObjDispatcher.Type.Sink) enum EvtName = "onSinkEvent";
 203         else static if (tp == ObjDispatcher.Type.Bubble) enum EvtName = "onBubbleEvent";
 204         else static assert(0, "invalid type");
 205         Weak!OT obj;
 206         this (OT aobj) { obj = new Weak!OT(aobj); /*obj.onDead = &ebusOnDeadCB;*/ }
 207         override bool alive () { return !obj.empty; }
 208         override Type type () { return tp; }
 209         override bool isObject (Object o) {
 210           if (o is null || obj.empty) return false;
 211           return (obj.object is o);
 212         }
 213         override void dispatch (Event evt) {
 214           if (obj.empty) return;
 215           auto o = obj.object;
 216           scope(exit) o = null; // for GC
 217           void doit (TypeInfo_Class ti) {
 218             import std.traits;
 219             if (typeid(Event) != ti) doit(ti.base);
 220             if (evt.processed) return;
 221             foreach (immutable oidx, /*auto*/ over; __traits(getOverloads, OT, EvtName)) {
 222               alias tt = typeof(over);
 223               static if (is(ReturnType!tt == void)) {
 224                 alias pars = Parameters!tt;
 225                 static if (pars.length == 1) {
 226                   static if (is(pars[0] : Event)) {
 227                     if (typeid(pars[0]) == ti) {
 228                       if (auto ev = cast(pars[0])evt) {
 229                         debug(tuing_eventbus) { import iv.vfs.io; VFile("zevtlog.log", "a").writefln("calling %s for %s 0x%08x (%s)", tp, o.classinfo.name, *cast(void**)&o, evt.classinfo.name); }
 230                         __traits(getOverloads, o, EvtName)[oidx](ev);
 231                         return;
 232                       }
 233                     }
 234                   }
 235                 }
 236               }
 237             }
 238           }
 239           doit(typeid(evt));
 240         }
 241       }
 242       synchronized (Event.classinfo) {
 243         bool found = false;
 244         static if (tp == ObjDispatcher.Type.Broadcast) {
 245           foreach (ref EventListenerInfo eli; llist) {
 246             if (eli.dobj !is null && eli.dobj.type == tp && eli.dobj.isObject(obj)) { found = true; break; }
 247           }
 248         }
 249         if (!found) {
 250           debug(tuing_eventbus) { import iv.vfs.io; VFile("zevtlog.log", "a").writefln("added %s for %s 0x%08x", tp, obj.classinfo.name, *cast(void**)&obj); }
 251           auto dp = new ObjDispatcherX!(O, tp)(obj);
 252           static if (tp == ObjDispatcher.Type.Broadcast) {
 253             llist ~= EventListenerInfo(dp);
 254           } else {
 255             llistsb ~= EventListenerInfo(dp);
 256           }
 257         }
 258       }
 259     }
 260   }
 261
 262   addListener!(ObjDispatcher.Type.Broadcast)();
 263   addListener!(ObjDispatcher.Type.My)();
 264   addListener!(ObjDispatcher.Type.Sink)();
 265   addListener!(ObjDispatcher.Type.Bubble)();
 266 }
 267
 268
 269 // ////////////////////////////////////////////////////////////////////////// //
 270 private import core.thread : Thread, ThreadID;
 271
 272 private __gshared bool nowProcessing = false;
 273 private __gshared ThreadID peLastTID = 0;
 274
 275
 276 /** process queued events. it is safe to post new events in event handlers.
 277  *
 278  * WARNING! this MUST be called only in main processing thread!
 279  */
 280 public void processEvents () {
 281   usize left;
 282   synchronized (Event.classinfo) {
 283     if (nowProcessing) throw new Exception("recursive calls to `processEvents()` aren't allowed");
 284     if (peLastTID == 0) {
 285       peLastTID = Thread.getThis.id;
 286     } else {
 287       if (peLastTID != Thread.getThis.id) throw new Exception("calling `processEvents()` from different threads aren't allowed");
 288     }
 289     nowProcessing = true;
 290     cleanupListeners();
 291     // process only so much events to avoid endless loop
 292     left = ppevents.length;
 293   }
 294   Event evt;
 295   scope(exit) evt = null;
 296   if (left > 0) {
 297     auto tm = MonoTime.currTime;
 298     while (left-- > 0) {
 299       synchronized (Event.classinfo) {
 300         if (tm < ppevents[0].hitTime) break;
 301         evt = ppevents.ptr[0].evt;
 302         foreach (immutable c; 1..ppevents.length) ppevents.ptr[c-1] = ppevents.ptr[c];
 303         ppevents[$-1] = ppevents[0].init;
 304         ppevents.length -= 1;
 305         ppevents.assumeSafeAppend;
 306       }
 307       try {
 308         callEventListeners(evt);
 309       } catch (Exception e) {
 310         if (ebusLogError !is null) {
 311           ebusLogError("******** EVENT PROCESSING ERROR: "~e.msg);
 312           ebusLogError("******** EVENT PROCESSING ERROR: "~e.toString);
 313         }
 314       }
 315     }
 316   }
 317   // process only so much events to avoid endless loop
 318   synchronized (Event.classinfo) left = events.length;
 319   while (left-- > 0) {
 320     synchronized (Event.classinfo) {
 321       evt = events.ptr[0];
 322       foreach (immutable c; 1..events.length) events.ptr[c-1] = events.ptr[c];
 323       events[$-1] = null;
 324       events.length -= 1;
 325       events.assumeSafeAppend;
 326     }
 327     try {
 328       callEventListeners(evt);
 329     } catch (Exception e) {
 330       if (ebusLogError !is null) {
 331         ebusLogError("******** EVENT PROCESSING ERROR: "~e.msg);
 332         ebusLogError("******** EVENT PROCESSING ERROR: "~e.toString);
 333       }
 334     }
 335   }
 336   synchronized (Event.classinfo) {
 337     cleanupListeners();
 338     nowProcessing = false;
 339   }
 340 }
 341
 342
 343 // ////////////////////////////////////////////////////////////////////////// //
 344 /// returns milliseconds until next postponed event or -1 if there are none
 345 public int ebusSafeDelay () {
 346   synchronized (Event.classinfo) {
 347     if (events.length > 0) return 0; // we have something to process
 348     if (ppevents.length == 0) return -1;
 349     auto tm = MonoTime.currTime;
 350     if (tm >= ppevents[0].hitTime) return 0; // first scheduled event should already be fired
 351     auto res = cast(int)((ppevents[0].hitTime-tm).total!"msecs");
 352     return res;
 353   }
 354 }
 355
 356
 357 // ////////////////////////////////////////////////////////////////////////// //
 358 private:
 359 __gshared bool[uint] usedIds;
 360 __gshared uint lastid;
 361 __gshared bool wrapped = false;
 362
 363
 364 uint getId () {
 365   if (!wrapped) {
 366     auto res = ++lastid;
 367     if (res != 0) { usedIds[res] = true; return res; }
 368     wrapped = true;
 369     lastid = 1;
 370   }
 371   // wrapped
 372   foreach (long count; 0..cast(long)uint.max+1) {
 373     if (lastid !in usedIds) { usedIds[lastid] = true; return lastid; }
 374     if (++lastid == 0) lastid = 1;
 375   }
 376   assert(0, "too many listeners!");
 377 }
 378
 379
 380 struct PostponedEvent {
 381   Event evt;
 382   MonoTime hitTime;
 383 }
 384
 385 __gshared Event[] events;
 386 __gshared PostponedEvent[] ppevents;
 387
 388
 389 void postpone (Event evt, int msecs) {
 390   import core.time : dur;
 391   if (evt is null) return;
 392   synchronized (Event.classinfo) {
 393     if (evt.posted) throw new Exception("can't post already posted event");
 394     evt.flags |= Event.PFlags.Posted;
 395     if (msecs <= 0) { events ~= evt; return; }
 396     auto tm = MonoTime.currTime+dur!"msecs"(msecs);
 397     ppevents ~= PostponedEvent(evt, tm);
 398     if (ppevents.length > 1) {
 399       import std.algorithm : sort;
 400       ppevents.sort!((in ref i0, in ref i1) => i0.hitTime < i1.hitTime);
 401     }
 402   }
 403 }
 404
 405
 406 // ////////////////////////////////////////////////////////////////////////// //
 407 // get druntime dynamic cast function
 408 private extern(C) void* _d_dynamic_cast (Object o, ClassInfo c);
 409
 410
 411 struct EventListenerInfo {
 412   alias DgType = void delegate (/*Event*/void* e); // actually, `e` is any `Event` subclass; cheater!
 413   TypeInfo_Class ti;
 414   DgType dg;
 415   uint id;
 416   ObjDispatcher dobj;
 417   Weak!Object xsrc;
 418   bool oneshot;
 419   this (TypeInfo_Class ati, DgType adg, bool aoneshot) {
 420     id = getId;
 421     ti = ati;
 422     dg = adg;
 423     oneshot = aoneshot;
 424   }
 425   this (Object asrc, TypeInfo_Class ati, DgType adg, bool aoneshot) {
 426     id = getId;
 427     ti = ati;
 428     dg = adg;
 429     if (asrc !is null) {
 430       xsrc = new Weak!Object(asrc);
 431       //xsrc.onDead = &ebusOnDeadCB;
 432     }
 433     oneshot = aoneshot;
 434   }
 435   this (ObjDispatcher adobj) {
 436     id = getId;
 437     dobj = adobj;
 438   }
 439 }
 440
 441 __gshared EventListenerInfo[] llist;
 442 __gshared EventListenerInfo[] llistsb; // sink/bubble
 443 __gshared bool needListenerCleanup = false;
 444 shared bool needListenerCleanupSB = false;
 445
 446
 447 void cleanupListeners () {
 448   if (needListenerCleanup) {
 449     needListenerCleanup = false;
 450     int pos = 0;
 451     while (pos < llist.length) {
 452       auto ell = llist.ptr+pos;
 453       if (ell.id != 0) {
 454         if (ell.dobj !is null && !ell.dobj.alive) ell.id = 0;
 455         if (ell.xsrc !is null && ell.xsrc.empty) ell.id = 0;
 456       }
 457       if (ell.id == 0) {
 458         foreach (immutable c; pos+1..llist.length) llist.ptr[c-1] = llist.ptr[c];
 459         llist[$-1] = EventListenerInfo.init;
 460         llist.length -= 1;
 461         llist.assumeSafeAppend;
 462       } else {
 463         ++pos;
 464       }
 465     }
 466   }
 467   import core.atomic : cas;
 468   if (cas(&needListenerCleanupSB, true, false)) {
 469     int pos = 0;
 470     while (pos < llistsb.length) {
 471       auto ell = llistsb.ptr+pos;
 472       if (ell.id != 0) {
 473         if (ell.dobj !is null && !ell.dobj.alive) ell.id = 0;
 474         if (ell.xsrc !is null && ell.xsrc.empty) ell.id = 0;
 475       }
 476       if (ell.id == 0) {
 477         foreach (immutable c; pos+1..llistsb.length) llistsb.ptr[c-1] = llistsb.ptr[c];
 478         llistsb[$-1] = EventListenerInfo.init;
 479         llistsb.length -= 1;
 480         llistsb.assumeSafeAppend;
 481       } else {
 482         ++pos;
 483       }
 484     }
 485   }
 486 }
 487
 488
 489 __gshared Object[] evchain;
 490
 491 void buildEvChain (Object obj) {
 492   if (evchain.length) { evchain[] = null; evchain.length = 0; evchain.assumeSafeAppend; }
 493   while (obj !is null) {
 494     evchain ~= obj;
 495     if (auto itf = cast(EventTarget)(obj)) {
 496       obj = itf.eventbusParent();
 497     } else {
 498       break;
 499     }
 500   }
 501 }
 502
 503
 504 // with evchain
 505 void dispatchSinkBubble (Event evt) {
 506   void callOnTypeFor (ObjDispatcher.Type type, Object obj) {
 507     usize llen;
 508     synchronized (Event.classinfo) llen = llistsb.length;
 509     foreach (usize idx; 0..llen) {
 510       ObjDispatcher dobj = null;
 511       synchronized (Event.classinfo) {
 512         auto eli = &llistsb[idx];
 513         if (eli.id == 0) continue;
 514         if (eli.dobj !is null) {
 515           if (!eli.dobj.alive) { import core.atomic; atomicStore(needListenerCleanupSB, true); eli.id = 0; eli.dobj = null; eli.xsrc = null; continue; }
 516           dobj = eli.dobj;
 517           if (dobj.type != type || !dobj.isObject(obj)) dobj = null;
 518         }
 519       }
 520       if (dobj !is null) {
 521         dobj.dispatch(evt);
 522         if (evt.processed) break;
 523       }
 524     }
 525   }
 526
 527   if (evchain.length != 0) {
 528     // sinking phase, backward
 529     foreach_reverse (Object o; evchain[1..$]) {
 530       callOnTypeFor(ObjDispatcher.Type.Sink, o);
 531       if (evt.processed) return;
 532       if (auto itf = cast(EventTarget)o) {
 533         itf.eventbusOnEvent(evt);
 534         if (evt.processed) return;
 535       }
 536     }
 537     evt.flags |= Event.PFlags.Bubbling;
 538     callOnTypeFor(ObjDispatcher.Type.My, evchain[0]);
 539     if (evt.processed) return;
 540     if (auto itf = cast(EventTarget)evchain[0]) {
 541       itf.eventbusOnEvent(evt);
 542       if (evt.processed) return;
 543     }
 544     // bubbling phase, forward
 545     foreach (Object o; evchain[1..$]) {
 546       callOnTypeFor(ObjDispatcher.Type.Bubble, o);
 547       if (evt.processed) return;
 548       if (auto itf = cast(EventTarget)o) {
 549         itf.eventbusOnEvent(evt);
 550         if (evt.processed) return;
 551       }
 552     }
 553   } else {
 554     if (evt.processed) return;
 555     evt.flags |= Event.PFlags.Bubbling;
 556     callOnTypeFor(ObjDispatcher.Type.My, evt.odest);
 557   }
 558 }
 559
 560
 561 void callEventListeners (Event evt) {
 562   if (evt is null) return;
 563   scope(exit) { evt.osource = null; evt.odest = null; }
 564   if (evt.processed) return;
 565   debug(tuing_eventbus) {{
 566     import iv.vfs.io;
 567     auto fo = VFile("zevtlog.log", "a");
 568     fo.writef("dispatching %s", evt.classinfo.name);
 569     if (evt.osource !is null) fo.writef("; src=%s", evt.osource.classinfo.name);
 570     if (evt.odest !is null) fo.writef("; dest=%s", evt.odest.classinfo.name);
 571     fo.writeln;
 572   }}
 573   // broadcast listeners (they will also catch directed events w/o doing sink/bubble)
 574   usize llen;
 575   synchronized (Event.classinfo) llen = llist.length;
 576   foreach (usize idx; 0..llen) {
 577     ObjDispatcher dobj = null;
 578     EventListenerInfo.DgType dg = null;
 579     TypeInfo_Class ti = null;
 580     Weak!Object xsrc = null;
 581     bool oneshot = false;
 582     synchronized (Event.classinfo) {
 583       auto eli = &llist[idx];
 584       if (eli.id == 0) continue;
 585       xsrc = eli.xsrc;
 586       if (xsrc !is null && xsrc.empty) { needListenerCleanup = true; eli.id = 0; eli.dobj = null; eli.xsrc = null; continue; }
 587       if (eli.dobj !is null) {
 588         if (!eli.dobj.alive) { needListenerCleanup = true; eli.id = 0; eli.dobj = null; eli.xsrc = null; continue; }
 589         dobj = eli.dobj;
 590         if (dobj.type != ObjDispatcher.Type.Broadcast) dobj = null;
 591       } else {
 592         dg = eli.dg;
 593         ti = eli.ti;
 594       }
 595       oneshot = eli.oneshot;
 596     }
 597     if (xsrc !is null) {
 598       if (xsrc.empty) continue;
 599       if (xsrc.object !is evt.osource) continue;
 600     }
 601     if (dobj !is null) {
 602       if (oneshot) synchronized (Event.classinfo) { auto eli = &llist[idx]; needListenerCleanup = true; eli.id = 0; eli.dg = null; eli.dobj = null; eli.xsrc = null; }
 603       dobj.dispatch(evt);
 604       if (evt.processed) break;
 605     } else if (dg !is null && ti !is null) {
 606       auto ecc = _d_dynamic_cast(evt, ti);
 607       if (ecc !is null) {
 608         if (oneshot) synchronized (Event.classinfo) { auto eli = &llist[idx]; needListenerCleanup = true; eli.id = 0; eli.dg = null; eli.dobj = null; eli.xsrc = null; }
 609         dg(ecc);
 610         if (evt.processed) break;
 611       }
 612     }
 613   }
 614   if (evt.processed) return;
 615   // targeted message?
 616   if (evt.odest !is null) {
 617     scope(exit) if (evchain.length) { evchain[] = null; evchain.length = 0; evchain.assumeSafeAppend; }
 618     buildEvChain(evt.odest);
 619     dispatchSinkBubble(evt);
 620   }
 621 }
 622
 623
 624 // ////////////////////////////////////////////////////////////////////////// //
 625 version(tuing_ebus_test) {
 626 import iv.vfs.io;
 627
 628 class EventEx : Event { this (Object s, Object d) { super(s, d); } }
 629
 630 class Obj : EventTarget {
 631   Object parent;
 632
 633   // this should return parent object or null
 634   override @property Object eventbusParent () { return parent; }
 635
 636   // this will be called on sinking and bubbling
 637   override void eventbusOnEvent (Event evt) {
 638     writefln("eventbusOnEvent: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 639     //mixin(SBDispatchMixin);
 640   }
 641
 642   this () {
 643     this.connectListeners();
 644     writefln("ctor: 0x%08x", cast(uint)cast(void*)this);
 645   }
 646
 647   void onEvent (Event evt) {
 648     writefln("onEvent: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 649   }
 650
 651   void onMyEvent (Event evt) {
 652     writefln("onMy: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 653   }
 654
 655   void onMyEvent (EventEx evt) {
 656     writefln("onMyEx: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 657   }
 658
 659   void onSinkEvent (Event evt) {
 660     writefln("onSink: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 661   }
 662
 663   void onBubbleEvent (EventEx evt) {
 664     writefln("onBubble: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 665   }
 666 }
 667
 668 class Obj1 : Obj {
 669   alias onMyEvent = super.onMyEvent;
 670
 671   this () {
 672     this.connectListeners();
 673     super();
 674     //writefln("ctor: 0x%08x", cast(uint)cast(void*)this);
 675   }
 676
 677   override void onMyEvent (EventEx evt) {
 678     writefln("OVERRIDEN onMyEx: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 679     super.onMyEvent(evt);
 680     writefln("EXITING OVERRIDEN onMyEx: 0x%08x 0x%08x %s", cast(uint)cast(void*)this, cast(uint)cast(void*)evt.odest, evt.bubbling);
 681   }
 682 }
 683
 684
 685 void main () {
 686   Object o0 = new Obj;
 687   Obj o1 = new Obj;
 688   Obj o2 = new Obj1;
 689   o1.parent = o0;
 690   o2.parent = o1;
 691   addEventListener((Event evt) { writeln("!!! main handler!"); });
 692   (new Event()).post;
 693   (new Event(null, o0)).post;
 694   (new Event(null, o1)).post;
 695   (new Event(null, o2)).post;
 696   processEvents();
 697   writeln("====");
 698   (new EventEx(null, o2)).post;
 699   processEvents();
 700 }
 701 }