static <A> O vmBus_onMessage_withLivelinessCheck(bool verbose default false, IF0<Bool> livelinessCheck, S msg, IVF1<A> onMessage) {
  // Do this BEFORE synchronizing on the map!
  VMBusListenerWithLivelinessCheck listener = new(null, livelinessCheck, (_msg, arg) -> onMessage.get((A) arg));
  listener.verbose = verbose;
  
  Map<S, Set> map = vm_busListenersByMessage_live();
  synchronized(map) {
    Set listeners = map.get(msg);
    if (listeners == null) map.put(msg, listeners = syncIdentityHashSet());
    listener.listeners = listeners;
    // We're technically violating the one-synchronized-object-per-thread rule,
    // but it should be OK here.
    listeners.add(listener);
    ret listener;
  }
}

Began life as a copy of #1031570