distribute the object among listeners.

@param obj specific obj
@param isSync is sync or not
429523
public void distribute(ProcessorListener.Notification<ApiType> obj, boolean isSync) {
    lock.readLock().lock();
    try {
      if (isSync) {
        for (ProcessorListener<ApiType> listener : syncingListeners) {
          listener.add(obj);
        }
      } else {
        for (ProcessorListener<ApiType> listener : listeners) {
          listener.add(obj);
        }
      }
    } finally {
      lock.readLock().unlock();
    }
  }
1007834
public static void runIntoJITSync(final JRebirthRunnable runnable, final long... timeout) {
        final SyncRunnable sync = new SyncRunnable(runnable);

        if (JRebirth.isJIT()) {
            // We are into a JIT so just run it synchronously
            sync.run();
            // Be careful in this case no timeout protection is achieved
        } else {
            // The runnable will be run into the JIT during the next round according to existing runnable priorities
            JRebirthThread.getThread().runLater(sync);
            // Wait the end of the runnable execution
            sync.waitEnd(timeout);
        }
    }
637768
function _resyncList() {
  var listKey = this.listKey;
  var parentPath = this.parentPath;
  if (!listKey || !parentPath) return;

  var newContainer = parentPath.node[listKey];
  if (this.container === newContainer) return;

  // container is out of sync. this is likely the result of it being reassigned

  if (newContainer) {
    this.container = newContainer;
  } else {
    this.container = null;
  }
}
301621
@Override
	public void externalizeCacheObjectAsync(final String fileNamePrefix, final Object object,
			final CreateCompletionCallback callback) {
		_aggregator.getExecutors().getFileCreateExecutor().submit(new Runnable() {
			File file = null;
			public void run() {
				try {
					file = File.createTempFile(fileNamePrefix, ".cache", _directory); //$NON-NLS-1$
					ObjectOutputStream os;
					os = new ObjectOutputStream(new FileOutputStream(file));
					try {
						os.writeObject(object);
					} finally {
						try { os.close(); } catch (Exception ignore) {}
					}
					if (callback != null) {
						callback.completed(file.getName(), null);
					}
				} catch (IOException e) {
					if (log.isLoggable(Level.WARNING))
						log.log(Level.WARNING, MessageFormat.format(
								Messages.CacheManagerImpl_4, new Object[]{file.getPath()}), e);
					if (callback != null) {
						callback.completed(file != null ? file.getName() : null, e);
					}
				}
			}
		});
	}
460636
function (obj) {
                shared.utils.dassert(shared.tracker.isTracked(obj));
                if(obj instanceof Array) {
                    this.arrayChanges(obj);
                } else {
                    this.objectChanges(obj);
                }
            }
425719
public boolean syncClonedListener(BaseField field, FieldListener listener, boolean bInitCalled)
    {
        if (!bInitCalled)
            ((CheckRangeHandler)listener).init(null, m_dStartRange, m_dEndRange);
        return super.syncClonedListener(field, listener, true);
    }
830741
function publish(prop, obj, callback) {
    callback = callback || function() {};
    if (typeof callback !== 'function') {
      throw new Error('invalid callback');
    }
    if (!prop || typeof prop !== 'string' || prop.length < 1) {
      throw new Error('invalid property path');
    }

    // Worker for publishing queue.
    function _publish(obj, callback) {
      // Write the object to the separate file.
      var shortname = prop + '.json';
      var data = JSON.stringify(obj);
      var hash = _checkNewHash(shortname, data);
      if (!hash) {
        // No need to publish if the data is the same.
        return callback();
      }
      published[shortname] = true;
      // Update new object and call notifications.
      _updateObject(shortname, obj, hash);
      var fullname = path.join(folder, shortname);
      fs.writeFile(fullname, data, function (err) {
        if (err) {
          console.error('Failed to write file %s err:', fullname, err);
        }
        return callback(err);
      });
    }

    // If there is no queue for this object, create one.
    if (!publishQueues[prop]) {
      // Only one task should be executed by worker at any time.
      publishQueues[prop] = async.queue(_publish, 1);
    }
    // Publishing of the same object should be queued and executed one after
    // completion of another.
    publishQueues[prop].push(obj, callback);
  }
776184
@Override
    public IPromise then(Callback resultCB) {
        // FIXME: this can be implemented more efficient
        while( !lock.compareAndSet(false,true) ) {}
        try {
            if (resultReceiver != null)
                throw new RuntimeException("Double register of promise listener");
            resultReceiver = resultCB;
            if (hadResult) {
                hasFired = true;
                if (nextFuture == null) {
                    nextFuture = new Promise(result, error);
                    lock.set(false);
                    resultCB.complete(result, error);
                } else {
                    lock.set(false);
                    resultCB.complete(result, error);
                    nextFuture.complete(result, error);
                    return nextFuture;
                }
            }
            if (resultCB instanceof IPromise) {
                return (IPromise) resultCB;
            }
            lock.set(false);
            while( !lock.compareAndSet(false,true) ) {}
            if (nextFuture == null) {
                return nextFuture = new Promise();
            } else {
                return nextFuture;
            }
        } finally {
            lock.set(false);
        }
    }
46730
public boolean start(final FileDownloadListener listener, final boolean isSerial) {
        if (listener == null) {
            Util.w(TAG, "Tasks with the listener can't start, because the listener "
                    + "provided is null: [null, " + isSerial + "]");
            return false;
        }

        List<DownloadTaskAdapter> originalTasks =
                FileDownloadList.getImpl().assembleTasksToStart(listener);
        if (originalTasks.isEmpty()) {
            Util.w(TAG, "no task for listener: " + listener + " to start");
            return false;
        }

        ArrayList<DownloadTask> downloadTasks = new ArrayList<>();
        for (DownloadTaskAdapter task : originalTasks) {
            downloadTasks.add(task.getDownloadTask());
        }
        final DownloadContext downloadContext =
                new DownloadContext.Builder(new DownloadContext.QueueSet(), downloadTasks)
                        .setListener(new DownloadContextListener() {
                            @Override
                            public void taskEnd(@NonNull DownloadContext context,
                                                @NonNull DownloadTask task,
                                                @NonNull EndCause cause,
                                                @Nullable Exception realCause,
                                                int remainCount) {
                                Util.d(TAG, "task " + task.getId() + "end");
                                final DownloadTaskAdapter downloadTaskAdapter =
                                        FileDownloadUtils.findDownloadTaskAdapter(task);
                                if (downloadTaskAdapter != null) {
                                    FileDownloadList.getImpl().remove(downloadTaskAdapter);
                                }
                            }

                            @Override
                            public void queueEnd(@NonNull DownloadContext context) {
                                Util.d(TAG, "queue end");
                            }
                        })
                        .build();

        final CompatListenerAdapter compatListenerAdapter = CompatListenerAdapter.create(listener);
        downloadContext.start(compatListenerAdapter, isSerial);
        return true;
    }
128503
static <T> boolean syncInstantiation(T objectType) {
        List<Template> templates = new ArrayList<>();
        Templates tr = TEMP_FINDER.findAnnotations(templates, objectType);
        if (tr == null) {
            /* Default to synchronous instantiation */
            return true;
        } else {
            return tr.syncInstantiation();
        }
    }
24734
async function emitAsync (eventName, ...args) {
  // using a copy to avoid error when listener array changed
  let listeners = this.listeners(eventName)
  for (let i = 0; i < listeners.length; i++) {
    let fn = listeners[i]
    let obj = fn(...args)
    if (!!obj && (typeof obj === 'object' || typeof obj === 'function') && typeof obj.then === 'function') obj = await obj
    if (obj !== undefined) return obj
  }
}
1103630
@Override
    public void addObject(DigitalObject obj) {
        // CURRENT TIME:
        // Get the current time to use for created dates on object
        // and object components (if they are not already there).
        Date nowUTC = new Date();

        // DOReplicator replicator=(DOReplicator)
        // m_server.getModule("org.fcrepo.server.storage.replication.DOReplicator");
        DOManager manager =
                (DOManager) m_server
                        .getModule("org.fcrepo.server.storage.DOManager");
        FieldSearch fieldSearch =
                (FieldSearch) m_server
                        .getModule("org.fcrepo.server.search.FieldSearch");
        PIDGenerator pidGenerator =
                (PIDGenerator) m_server
                        .getModule("org.fcrepo.server.management.PIDGenerator");

        // SET OBJECT PROPERTIES:
        logger.debug("Rebuild: Setting object/component states and create dates if unset...");
        // set object state to "A" (Active) if not already set
        if (obj.getState() == null || obj.getState().isEmpty()) {
            obj.setState("A");
        }
        // set object create date to UTC if not already set
        if (obj.getCreateDate() == null) {
            obj.setCreateDate(nowUTC);
        }
        // set object last modified date to UTC
        obj.setLastModDate(nowUTC);

        // SET OBJECT PROPERTIES:
        logger.debug("Rebuild: Setting object/component states and create dates if unset...");
        // set object state to "A" (Active) if not already set
        if (obj.getState() == null || obj.getState().isEmpty()) {
            obj.setState("A");
        }
        // set object create date to UTC if not already set
        if (obj.getCreateDate() == null) {
            obj.setCreateDate(nowUTC);
        }
        // set object last modified date to UTC
        obj.setLastModDate(nowUTC);

        // SET DATASTREAM PROPERTIES...
        Iterator<String> dsIter = obj.datastreamIdIterator();
        while (dsIter.hasNext()) {
            for (Datastream ds : obj.datastreams(dsIter.next())) {
                // Set create date to UTC if not already set
                if (ds.DSCreateDT == null) {
                    ds.DSCreateDT = nowUTC;
                }
                // Set state to "A" (Active) if not already set
                if (ds.DSState == null || ds.DSState.isEmpty()) {
                    ds.DSState = "A";
                }
            }
        }

        // PID GENERATION:
        // have the system generate a PID if one was not provided
        logger.debug("INGEST: Stream contained PID with retainable namespace-id... will use PID from stream.");
        try {
            pidGenerator.neverGeneratePID(obj.getPid());
        } catch (IOException e) {
            throw new RuntimeException("Error calling pidGenerator.neverGeneratePID(): "
                                               + e.getMessage(),
                                       e);
        }

        // REGISTRY:
        // at this point the object is valid, so make a record
        // of it in the digital object registry
        try {
            registerObject(obj);
        } catch (StorageDeviceException e) {
            // continue past individual errors
            logger.error(e.getMessage());
        }

        try {
            logger.info("COMMIT: Attempting replication: {}", obj.getPid());
            DOReader reader =
                    manager.getReader(Server.USE_DEFINITIVE_STORE,
                                      m_context,
                                      obj.getPid());
            logger.info("COMMIT: Updating FieldSearch indexes...");
            fieldSearch.update(reader);

        } catch (ServerException se) {
            System.out.println("Error while replicating: "
                    + se.getClass().getName() + ": " + se.getMessage());
            se.printStackTrace();
        } catch (Throwable th) {
            System.out.println("Error while replicating: "
                    + th.getClass().getName() + ": " + th.getMessage());
            th.printStackTrace();
        }
    }
309885
public void sync() throws ClosedJournalException, IOException {
        try {
            appender.sync().get();
            if (appender.getAsyncException() != null) {
                throw new IOException(appender.getAsyncException());
            }
        } catch (Exception ex) {
            throw new IllegalStateException(ex.getMessage(), ex);
        }
    }
566707
void push(Object object) {
        callStack.push(object);

        if (object != null && !SIMPLE_TYPES.contains(object.getClass())) {
            objectsInCallStack.add(object);
        }
        objectCount++;
    }
1261698
public void registerListenersOnEverything(final Object object) {
        try {
            // the removeListener() call ensures that the listener is not added more than once
            new PropertyVisitor(object) {
                @Override
                protected boolean visitSingleValueProperty(final Property<?> fieldValue) {
                    fieldValue.removeListener(propertyListener);
                    fieldValue.addListener(propertyListener);
                    return true;
                }

                @Override
                protected boolean visitCollectionProperty(final ListProperty<?> fieldValue) {
                    fieldValue.removeListener(listListener);
                    fieldValue.addListener(listListener);
                    return true;
                }

                @Override
                protected boolean visitCollectionProperty(final MapProperty<?, ?> fieldValue) {
                    fieldValue.removeListener(mapListener);
                    fieldValue.addListener(mapListener);
                    return true;
                }

                @Override
                protected boolean visitCollectionProperty(final SetProperty<?> fieldValue) {
                    fieldValue.removeListener(setListener);
                    fieldValue.addListener(setListener);
                    return false;
                }
            };
        } catch (final IllegalAccessException e) {
            topology.onError(new SynchronizeFXException(e));
        } catch (final SecurityException e) {
            topology.onError(new SynchronizeFXException(
                    "Maybe you're JVM doesn't allow reflection for this application?", e));
        }
    }
341383
public boolean syncClonedListener(BaseField field, FieldListener listener, boolean bInitCalled)
    {
        if (!bInitCalled)
            ((ChangeTitleOnChangeHandler)listener).init(null, m_bSetTitleToThisString);
        return super.syncClonedListener(field, listener, true);
    }
164434
synchronized void sync()
    {
        try
        {
            // check we have more work to do
            if (allocatePosition.get() <= lastSyncedOffset + SYNC_MARKER_SIZE)
                return;

            // allocate a new sync marker; this is both necessary in itself, but also serves to demarcate
            // the point at which we can safely consider records to have been completely written to
            int nextMarker;
            nextMarker = allocate(SYNC_MARKER_SIZE);
            boolean close = false;
            if (nextMarker < 0)
            {
                // ensure no more of this CLS is writeable, and mark ourselves for closing
                discardUnusedTail();
                close = true;

                // wait for modifications guards both discardedTailFrom, and any outstanding appends
                waitForModifications();

                if (discardedTailFrom < buffer.capacity() - SYNC_MARKER_SIZE)
                {
                    // if there's room in the discard section to write an empty header, use that as the nextMarker
                    nextMarker = discardedTailFrom;
                }
                else
                {
                    // not enough space left in the buffer, so mark the next sync marker as the EOF position
                    nextMarker = buffer.capacity();
                }
            }
            else
            {
                waitForModifications();
            }

            assert nextMarker > lastSyncedOffset;

            // write previous sync marker to point to next sync marker
            // we don't chain the crcs here to ensure this method is idempotent if it fails
            int offset = lastSyncedOffset;
            final PureJavaCrc32 crc = new PureJavaCrc32();
            crc.updateInt((int) (id & 0xFFFFFFFFL));
            crc.updateInt((int) (id >>> 32));
            crc.updateInt(offset);
            buffer.putInt(offset, nextMarker);
            buffer.putInt(offset + 4, crc.getCrc());

            // zero out the next sync marker so replayer can cleanly exit
            if (nextMarker < buffer.capacity())
            {
                buffer.putInt(nextMarker, 0);
                buffer.putInt(nextMarker + 4, 0);
            }

            // actually perform the sync and signal those waiting for it
            buffer.force();

            if (close)
                nextMarker = buffer.capacity();

            lastSyncedOffset = nextMarker;
            syncComplete.signalAll();

            CLibrary.trySkipCache(fd, offset, nextMarker);
            if (close)
                internalClose();
        }
        catch (Exception e) // MappedByteBuffer.force() does not declare IOException but can actually throw it
        {
            throw new FSWriteError(e, getPath());
        }
    }