// listener: voidfunc(S, O) or func(S, O)
static <A> AutoCloseable vmBus_addListener_basic(S msg, O listener) {
  Map<S, Set> map = vm_busListenersByMessage_live();
  synchronized(map) {
    Set listeners = map.get(msg);
    if (listeners == null) map.put(msg, listeners = syncIdentityHashSet());
    // We're technically violating the one-synchronized-object-per-thread rule,
    // but it should be OK here.
    // TODO: remove empty sets from map
    ret tempAdd(listeners, listener);
  }
}