Notify the management listeners on a messageReceived.
<p/>
NOTE: this starts on the IO thread, but runs a task OFF the thread.
510753
@Override
    public void doFilterWriteListeners(final WriteRequest writeRequest) {

        final Object message = writeRequest.getMessage();
        WriteRequest originalRequest = writeRequest.getOriginalRequest();
        final Object originalMessage = originalRequest != null ? originalRequest.getMessage() : null;

        runManagementTask(new Runnable() {
            @Override
            public void run() {
                try {
                    List<SessionManagementListener> sessionListeners = getManagementListeners();
                    for (final SessionManagementListener listener : sessionListeners) {
                        listener.doFilterWrite(SessionManagementBeanImpl.this, message, originalMessage);
                    }

                    markChanged();
                } catch (Exception ex) {
                    logger.warn("Error during doFilterWrite session listener notifications:", ex);
                }
            }
        });
    }
777997
@Override
    public void onReceive(Object message) {
        if (message instanceof NewMessage) {
            NewMessage newMessage = (NewMessage) message;
            onNewMessage(newMessage.getPeer(), newMessage.getSender(), newMessage.getSortDate(),
                    newMessage.getContentDescription(), newMessage.getHasCurrentUserMention(),
                    newMessage.getUnreadMessagesCount(), newMessage.getUnreadDialogsCount());
        } else if (message instanceof MessagesRead) {
            MessagesRead read = (MessagesRead) message;
            onMessagesRead(read.getPeer(), read.getFromDate());
        } else if (message instanceof PauseNotifications) {
            onNotificationsPaused();
        } else if (message instanceof ResumeNotifications) {
            onNotificationsResumed();
        } else {
            super.onReceive(message);
        }
    }
694193
public void enableReceiveNotifyMsg(boolean enable, EnableReceiveNotifyMsgHandler handler) {
        HMSAgentLog.i("enableReceiveNotifyMsg:enable=" + enable + " handler=" + StrUtils.objDesc(handler));
        this.enable = enable;
        this.handler = handler;
        connect();
    }
1284867
@Override
    public void messageReceived(IoSession session, Object message)
            throws Exception {
        chain.execute(null, session, message);
    }
825805
protected final void notifyStarted() {
    monitor.enter();
    try {
      // We have to examine the internal state of the snapshot here to properly handle the stop
      // while starting case.
      if (snapshot.state != STARTING) {
        IllegalStateException failure =
            new IllegalStateException(
                "Cannot notifyStarted() when the service is " + snapshot.state);
        notifyFailed(failure);
        throw failure;
      }

      if (snapshot.shutdownWhenStartupFinishes) {
        snapshot = new StateSnapshot(STOPPING);
        // We don't call listeners here because we already did that when we set the
        // shutdownWhenStartupFinishes flag.
        doStop();
      } else {
        snapshot = new StateSnapshot(RUNNING);
        running();
      }
    } finally {
      monitor.leave();
      executeListeners();
    }
  }
767973
def _notify_listeners(self, sender, message):
        
        uid = message['uid']
        msg_topic = message['topic']


        self._ack(sender, uid, 'fire')

        all_listeners = set()
        for lst_topic, listeners in self.__listeners.items():
            if fnmatch.fnmatch(msg_topic, lst_topic):
                all_listeners.update(listeners)

        self._ack(sender, uid, 'notice', 'ok' if all_listeners else 'none')

        try:
            results = []
            for listener in all_listeners:
                result = listener.handle_message(sender,
                                                 message['topic'],
                                                 message['content'])
                if result:
                    results.append(result)

            self._ack(sender, uid, 'send', json.dumps(results))
        except:
            self._ack(sender, uid, 'send', "Error")
1064183
@Override
    public void onReceive(Object message) {
        try {
            // Start all workers
            if (message instanceof InitialRequestToManager) {
                director = getSender();
                logger.info("parallec task state : "
                        + ParallelTaskState.IN_PROGRESS.toString());
                task.setState(ParallelTaskState.IN_PROGRESS);
                task.setExecuteStartTime(startTime);

                taskId = task.getTaskId();

                // assumption: at least REDUCE_LEN
                taskIdTrim = taskId.length() <= REDUCE_LEN ? taskId : taskId
                        .substring(taskId.length() - REDUCE_LEN,
                                taskId.length());

                httpMeta = task.getHttpMeta();
                targetHostMeta = task.getTargetHostMeta();
                final RequestProtocol requestProtocol = task
                        .getRequestProtocol();

                // Get request parameters to construct a REST CALL
                final String requestUrlPrefixOrig = httpMeta
                        .getRequestUrlPostfix();
                final HttpMethod httpMethod = httpMeta.getHttpMethod();

                String requestPortStrOrig = httpMeta.getRequestPort();

                final boolean pollable = httpMeta.isPollable();

                final int maxConcurrency = task.getConcurrency();

                Map<String, NodeReqResponse> nodeDataMapValid = task
                        .getParallelTaskResult();

                logger.info("Before Safety Check: total entry count: "
                        + nodeDataMapValid.size());

                Map<String, NodeReqResponse> nodeDataMapValidSafe = new HashMap<String, NodeReqResponse>();

                InternalDataProvider.getInstance()
                        .filterUnsafeOrUnnecessaryRequest(nodeDataMapValid,
                                nodeDataMapValidSafe);

                logger.info(
                        "After Safety Check: total entry count in nodeDataMapValidSafe: {}",
                        nodeDataMapValidSafe.size());

                logger.debug("maxConcurrency : {}", maxConcurrency);

                // after filter, no duplicated target hosts for this count.
                requestCount = nodeDataMapValidSafe.size();
                task.setRequestNumActual(requestCount);

                logger.info("!Obtain command request for target host meta id "
                        + targetHostMeta.getTargetHostId() + "  with count: "
                        + requestCount);
                // If there are no target hosts in the incoming message, send a
                // message back
                if (requestCount <= 0) {
                    getSender().tell(new ResponseFromManager(requestCount),
                            getSelf());
                    logger.info("req count <=0. return");
                    // not really error. just no task to run
                    reply(ParallelTaskState.COMPLETED_WITHOUT_ERROR,
                            new RuntimeException(
                                    "ReqCount after trim is 0. Return."));
                    return;
                }

                int sentRequestCounter = 0;

                asyncHttpClientGlobal = task.getAsyncHttpClient();

                final AsyncHttpClient asyncHttpClient = asyncHttpClientGlobal;

                // always send with valid safe data.
                for (Entry<String, NodeReqResponse> entry : nodeDataMapValidSafe
                        .entrySet()) {

                    final String targetHost = entry.getKey();
                    NodeReqResponse nodeReqResponse = entry.getValue();

                    final String requestContentOrig = nodeReqResponse
                            .getRequestParameters().get(
                                    PcConstants.REQUEST_BODY_PLACE_HOLDER);

                    final String requestContent = NodeReqResponse
                            .replaceStrByMap(
                                    nodeReqResponse.getRequestParameters(),
                                    requestContentOrig);
                    final String resourcePath = NodeReqResponse
                            .replaceStrByMap(
                                    nodeReqResponse.getRequestParameters(),
                                    requestUrlPrefixOrig);
                    // add support for port replacement
                    final String requestPortStr = NodeReqResponse
                            .replaceStrByMap(
                                    nodeReqResponse.getRequestParameters(),
                                    requestPortStrOrig);
                    int requestPort = 80;
                    try {
                        requestPort = Integer.parseInt(requestPortStr);
                    } catch (NumberFormatException nfe) {
                        logger.error(
                                "Error parsing replacable port with NumberFormatException. "
                                        + "No valid port for host {}. Now use default port 80",
                                targetHost);
                    }
                    // only pass when it is not in manager
                    final ParallecResponseHandler handler = task.getConfig()
                            .getHandlerExecutionLocation() == HandlerExecutionLocation.MANAGER ? null
                            : task.getHandler();
                    final Map<String, Object> responseContext = task
                            .getConfig().getHandlerExecutionLocation() == HandlerExecutionLocation.MANAGER ? null
                            : task.getResponseContext();
                    Map<String, String> httpHeaderMapLocal = new HashMap<String, String>();
                    httpHeaderMapLocal.putAll(httpMeta.getHeaderMetadata()
                            .getHeaderMap());

                    // 3rd, add the dynamic part ; generic var based
                    // replacement.
                    PcHttpUtils.replaceHttpHeaderMapNodeSpecific(
                            httpHeaderMapLocal,
                            nodeReqResponse.getRequestParameters());

                    /**
                     * If want to print to check
                     */
                    if (task.getConfig().isPrintHttpTrueHeaderMap()) {

                        for (Entry<String, String> headerEntry : httpHeaderMapLocal
                                .entrySet()) {
                            String headerKey = headerEntry.getKey();
                            String headerValue = headerEntry.getValue();

                            nodeReqResponse
                                    .getRequestParameters()
                                    .put(PcConstants.REQUEST_PARAMETER_HTTP_HEADER_PREFIX
                                            + headerKey, headerValue);
                        }

                    }

                    if (task.getConfig().isPrintPoller()) {
                        // put the one before encoding
                        nodeReqResponse.getRequestParameters().put(
                                PcConstants.NODE_REQUEST_NEED_POLLER,
                                Boolean.toString(pollable));
                    }

                    /**
                     * 20140310: END add pass HTTP header into Operation worker
                     * as part of the
                     */

                    String targetHostNew = nodeReqResponse
                            .getRequestParameters()
                            .get(PcConstants.UNIFORM_TARGET_HOST_VAR_WHEN_CHECK);
                    if (targetHostNew != null) {
                        nodeReqResponse.getRequestParameters().put(
                                PcConstants.NODE_REQUEST_TRUE_TARGET_NODE,
                                targetHostNew);
                    }

                    final String hostUniform = (targetHostNew == null) ? null
                            : targetHostNew;

                    nodeReqResponse.getRequestParameters().put(
                            PcConstants.NODE_REQUEST_STATUS,
                            SingleTargetTaskStatus.IN_PROGRESS.toString());

                    nodeReqResponse.getRequestParameters().put(
                            PcConstants.NODE_REQUEST_TRUE_CONTENT,
                            requestContent);

                    // put the one before encoding
                    nodeReqResponse.getRequestParameters().put(
                            PcConstants.NODE_REQUEST_TRUE_URL, resourcePath);

                    nodeReqResponse.getRequestParameters().put(
                            PcConstants.NODE_REQUEST_TRUE_PORT,
                            Integer.toString(requestPort));

                    nodeReqResponse.getRequestParameters().put(
                            PcConstants.NODE_REQUEST_HTTP_HEADER_META,
                            httpMeta.getHeaderMetadata().getHeaderStr());

                    long prepareRequestTime = System.currentTimeMillis();

                    String prepareRequestTimeStr = PcDateUtils
                            .getDateTimeStrStandard(new Date(prepareRequestTime));
                    nodeReqResponse.getRequestParameters().put(
                            PcConstants.NODE_REQUEST_PREPARE_TIME,
                            prepareRequestTimeStr);

                    final SshMeta sshMeta = task.getSshMeta();
                    final TcpMeta tcpMeta = task.getTcpMeta();
                    final UdpMeta udpMeta = task.getUdpMeta();

                    final PingMeta pingMeta = task.getPingMeta();
                    final ResponseHeaderMeta responseHeaderMeta = task
                            .getHttpMeta().getResponseHeaderMeta();

                    logger.debug("REQUEST GENERATED: "
                            + (sentRequestCounter + 1)
                            + " / "
                            + requestCount
                            + " after "
                            + Double.toString((prepareRequestTime - startTime) / 1000.0)
                            + " secs" + ":  (NOT SEND YET) " + targetHost
                            + " at " + prepareRequestTimeStr);

                    ActorRef worker = getContext().system().actorOf(
                            Props.create(OperationWorker.class,
                                    new TaskRequest(task.getConfig()
                                            .getActorMaxOperationTimeoutSec(),
                                            requestProtocol, targetHost,
                                            hostUniform, requestPort,
                                            resourcePath, requestContent,
                                            httpMethod, pollable,
                                            httpHeaderMapLocal, handler,
                                            responseContext, sshMeta, tcpMeta,
                                            udpMeta, pingMeta,
                                            responseHeaderMeta),
                                    asyncHttpClient, task.getHttpMeta()
                                            .getHttpPollerProcessor()

                            ));

                    workers.put(targetHost, worker);

                    ++sentRequestCounter;

                }// end for loop

                final RequestToBatchSenderAsstManager requestToBatchSenderAsstManager = new RequestToBatchSenderAsstManager(
                        taskId, task.getConfig()
                                .getAsstManagerRetryIntervalMillis(),
                        new ArrayList<ActorRef>(workers.values()), getSelf(),
                        maxConcurrency);

                batchSenderAsstManager = getContext().system().actorOf(
                        Props.create(AssistantExecutionManager.class),
                        "RequestToBatchSenderAsstManager-"
                                + UUID.randomUUID().toString());

                batchSenderAsstManager.tell(requestToBatchSenderAsstManager,
                        getSelf());

                final FiniteDuration timeOutDuration = Duration
                        .create(task.getConfig().getTimeoutInManagerSec(),
                                TimeUnit.SECONDS);
                timeoutMessageCancellable = getContext()
                        .system()
                        .scheduler()
                        .scheduleOnce(timeOutDuration, getSelf(),
                                ExecutionManagerMsgType.OPERATION_TIMEOUT,
                                getContext().system().dispatcher(), getSelf());

                logger.debug(
                        "Scheduled TIMEOUT_IN_MANAGER_SCONDS OPERATION_TIMEOUT after SEC {} ",
                        task.getConfig().getTimeoutInManagerSec());
            } else if (message instanceof ResponseOnSingleTask) {

                ResponseOnSingleTask taskResponse = (ResponseOnSingleTask) message;

                this.responseCount += 1;
                task.setResponsedNum(responseCount);

                /**
                 * add feedback of current responseCount to asst manager
                 * ResponseCountToBatchSenderAsstManager
                 */
                final ResponseCountToBatchSenderAsstManager responseCountToBatchSenderAsstManager = new ResponseCountToBatchSenderAsstManager(
                        this.responseCount);

                batchSenderAsstManager.tell(
                        responseCountToBatchSenderAsstManager, getSelf());

                logger.debug("Send batchSenderAsstManager to responseCountToBatchSenderAsstManager: "
                        + this.responseCount);

                String hostName = taskResponse.getRequest().getHost();
                if (responseMap.containsKey(hostName)) {
                    logger.error("ERROR: duplicate response received {}", hostName);
                }
                responseMap.put(hostName, taskResponse);

                String responseSummary = taskResponse.isError() ? "FAIL_GET_RESPONSE: "
                        + taskResponse.getErrorMessage()
                        : taskResponse.getStatusCode();
                Map<String, LinkedHashSet<String>> resultMap = task
                        .getAggregateResultMap();
                if (resultMap.containsKey(responseSummary)) {
                    resultMap.get(responseSummary).add(hostName);
                } else {
                    LinkedHashSet<String> set = new LinkedHashSet<String>();
                    set.add(hostName);
                    resultMap.put(responseSummary, set);
                }

                // save response to result map
                NodeReqResponse nrr = task.getParallelTaskResult()
                        .get(hostName);
                nrr.setSingleTaskResponse(taskResponse);

                String responseTrim = taskResponse.getResponseContent() == null ? null
                        : taskResponse.getResponseContent().trim();
                String displayResponse = (Strings.isNullOrEmpty(responseTrim)) ? "EMPTY"
                        : responseTrim
                                .substring(
                                        0,
                                        Math.min(
                                                PcConstants.AGNET_RESPONSE_MAX_RESPONSE_DISPLAY_BYTE,
                                                responseTrim.length()));

                long responseReceiveTime = System.currentTimeMillis();
                // %.5g%n
                double progressPercent = (double) (responseCount)
                        / (double) (requestCount) * 100.0;
                String responseReceiveTimeStr = PcDateUtils
                        .getDateTimeStrStandard(new Date(responseReceiveTime));
                String secondElapsedStr = Double
                        .toString((responseReceiveTime - startTime) / 1000.0);

                // log the first/ last 5 percent; then sample the middle
                if (requestCount < ParallecGlobalConfig.logAllResponseIfTotalLessThan
                        || responseCount <= ParallecGlobalConfig.logAllResponseBeforeInitCount
                        || progressPercent < ParallecGlobalConfig.logAllResponseBeforePercent
                        || progressPercent > ParallecGlobalConfig.logAllResponseAfterPercent
                        || responseCount
                                % ParallecGlobalConfig.logResponseInterval == 0) {
                    // percent is escaped using percent sign; hostName
                    logger.info(String
                            .format("\n[%d]__RESP_RECV_IN_MGR %d (+%d) / %d (%.5g%%)  "
                                    + "AFT %s S @ %s @ %s , TaskID : %s , CODE: %s, RESP_BRIEF: %s %s",
                                    responseCount,
                                    responseCount,
                                    requestCount - responseCount,
                                    requestCount,
                                    progressPercent,
                                    secondElapsedStr,
                                    hostName,
                                    responseReceiveTimeStr,
                                    taskIdTrim,
                                    taskResponse.getStatusCode(),
                                    displayResponse,
                                    taskResponse.getErrorMessage() == null ? ""
                                            : ", ERR: "
                                                    + taskResponse
                                                            .getErrorMessage()));
                }

                nrr.getRequestParameters().put(PcConstants.NODE_REQUEST_STATUS,
                        SingleTargetTaskStatus.COMPLETED.toString());

                if (task.getConfig().getHandlerExecutionLocation() == HandlerExecutionLocation.MANAGER
                        && task != null && task.getHandler() != null) {
                    try {
                        // logger.debug("HANDLE In manager: " +
                        // taskResponse.getHost());
                        task.getHandler().onCompleted(taskResponse,
                                task.getResponseContext());
                    } catch (Exception t) {
                        logger.error(
                                "Error handling onCompleted in manager for response: {} Error {}",
                                taskResponse.toString(),
                                t.getLocalizedMessage());
                    }
                }

                if (!task.getConfig().isSaveResponseToTask()) {
                    taskResponse.setResponseContent(PcConstants.NOT_SAVED);
                    taskResponse.setResponseHeaders(null);
                    logger.debug("Erased single task response content and response headers to save space.");
                }

                if (this.responseCount == this.requestCount) {
                    if (wasIssuedCancel) {
                        ExecutionManagerExecutionException exCanceled = new ExecutionManagerExecutionException(
                                "ExecutionManager: task was canceled by user",
                                ManagerExceptionType.CANCEL);
                        reply(ParallelTaskState.COMPLETED_WITH_ERROR,
                                exCanceled);
                    } else {
                        reply(ParallelTaskState.COMPLETED_WITHOUT_ERROR, null);
                    }

                }// end when all requests have resonponse

            } else if (message instanceof CancelTaskOnHostRequest) {
                CancelTaskOnHostRequest msg = (CancelTaskOnHostRequest) message;
                cancelRequestAndWorkerOnHost(msg.getTargetHosts());

            } else if (message instanceof ExecutionManagerMsgType) {
                switch ((ExecutionManagerMsgType) message) {

                // this will immediately return. not waiting for the op workers
                // reply
                case OPERATION_TIMEOUT:
                    cancelRequestAndWorkers();
                    String msg = "Execution manager timeout on whole Parallel Task.";
                    ExecutionManagerExecutionException ex = new ExecutionManagerExecutionException(
                            msg,
                            ManagerExceptionType.TIMEOUT);
                    logger.error(msg);
                    reply(ParallelTaskState.COMPLETED_WITH_ERROR, ex);
                    
                    break;
                // this will wait for the works to reply.
                case CANCEL:
                    cancelRequestAndWorkers();
                    wasIssuedCancel = true;
                    break;
                default:
                    break;
                }
            } else {
                logger.error("Unhandled: " + message);
                unhandled(message);
            }

        } catch (Exception t) {
            logger.error("Command Manager error: " + t + " trace: ", t);

            // not to terminate: but add error details. Will exit after timeout
            // or cancel
            task.getTaskErrorMetas().add(
                    new TaskErrorMeta(TaskErrorType.COMMAND_MANAGER_ERROR,
                            t == null ? "NA" : t.getLocalizedMessage()));
            reply(ParallelTaskState.COMPLETED_WITH_ERROR, t);

        }

    }
789386
public void notifyChanged() {
        if (listeners != null) {
            synchronized (notifyLock) {
                if (listeners != null) {
                    if (notifyThread == null) {
                        notifyThread = new NotifyThread(this);
                        notifyThread.setDaemon(true);
                        notifyThread.start();
                    }
                    notifyThread.queue(listeners.toArray(new EventListener[listeners.size()]));
                }
            }
        }
    }
1326377
private <T extends Rdp4jListener> void notifyListeners(Class<T> listenerType, Notifier<T> notifier)
            throws InterruptedException {
        for (Rdp4jListener listener : listeners) {
            if (isInstanceOf(listener, listenerType)) {
                /*
                 * This cast is correct, since we check if listener
                 * is instance of listenerType.
                 */
                @SuppressWarnings("unchecked")
                T listener2 = (T) listener;
                try {
                    notifier.notify(listener2);
                } catch (InterruptedException e) {
                    throw e;
                } catch (Throwable e) {
                    logErrorMessage(e);
                }
            }
        }
    }
782263
private void notifyConsumer( AbstractMessage message )
    {
    	consumersLock.readLock().lock();
    	try
    	{
    		switch (localConsumers.size())
    		{
    			case 0 : return; // Nobody's listening
    			case 1 : 
    				notifySingleConsumer(localConsumers.get(0),message);
    				break;
    			default : // Multiple consumers
    				notifyNextConsumer(localConsumers,message);
    				break;
    		}
    	}
    	finally
    	{
    		consumersLock.readLock().unlock();
    	}
    }
507079
@Override
  public void notifyManager() {
    logger.debug(String.format("Notifying Manager for %s", this.getClass().getName()));
    try {
      this.metricManager.reportMetric(this);
    } catch (final Throwable ex) {
      logger.error(
          String.format("Metric Manager is not set for %s metric", this.getClass().getName()), ex);
    }
  }
907984
void deliverTaskMessage(final byte[] message) {
    synchronized (this.contextLifeCycle) {
      if (!this.task.isPresent()) {
        LOG.log(Level.WARNING, "Received a task message while there was no task running. Ignoring.");
      } else {
        this.task.get().deliver(message);
      }
    }
  }
720861
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        byte[] rpcResponse = (byte[]) e.getMessage();
        // remove marking
        Xdr x = RecordMarkingUtil.removeRecordMarking(rpcResponse);
        // remove the request from timeout manager map
        int xid = x.getXid();
        _connection.notifySender(Integer.valueOf(xid), x);
    }
948974
@Override
    public void doSessionCreatedListeners(final long sessionId, final ManagementSessionType managementSessionType) {
        runManagementTask(new Runnable() {
            @Override
            public void run() {
                try {
                    // The particular management listeners change on strategy, so get them here.
                    for (final GatewayManagementListener listener : getManagementListeners()) {
                        listener.doSessionCreated(GatewayManagementBeanImpl.this, sessionId);
                    }

                    markChanged();  // mark ourselves as changed, possibly tell listeners
                } catch (Exception ex) {
                    logger.warn("Error during sessionCreated gateway listener notifications:", ex);
                }
            }
        });
    }
1291772
private void notifyListenersAfterWrite(ModbusMessage msg) {
        synchronized (listeners) {
            for (AbstractSerialTransportListener listener : listeners) {
                listener.afterMessageWrite(commPort, msg);
            }
        }
    }
420176
public void process(final boolean asynch)
          throws NotificationException {
    if (asynch) {
      try {
        consumer.setMessageListener(this);
        return;
      } catch (final JMSException je) {
        throw new NotificationException(je);
      }
    }

    while (running) {
      final Message m = conn.receive();

      if (m == null) {
        running = false;
        return;
      }
      onMessage(m);
    }
  }
1192105
void notifyObservers(ApptentiveNotification notification) {
		boolean hasLostReferences = false;

		// create a temporary list of observers to avoid concurrent modification errors
		List<ApptentiveNotificationObserver> temp = new ArrayList<>(observers.size());
		for (int i = 0; i < observers.size(); ++i) {
			ApptentiveNotificationObserver observer = observers.get(i);
			ObserverWeakReference observerRef = ObjectUtils.as(observer, ObserverWeakReference.class);
			if (observerRef == null || !observerRef.isReferenceLost()) {
				temp.add(observer);
			} else {
				hasLostReferences = true;
			}
		}

		// notify observers
		for (int i = 0; i < temp.size(); ++i) {
			try {
				temp.get(i).onReceiveNotification(notification);
			} catch (Exception e) {
				ApptentiveLog.e(e, "Exception while posting notification: %s", notification);
				logException(e); // TODO: add more context info
			}
		}

		// clean lost references
		if (hasLostReferences) {
			for (int i = observers.size() - 1; i >= 0; --i) {
				final ObserverWeakReference observerRef = ObjectUtils.as(observers.get(i), ObserverWeakReference.class);
				if (observerRef != null && observerRef.isReferenceLost()) {
					observers.remove(i);
				}
			}
		}
	}