Synchronously publish a message to all registered listeners (this includes listeners defined for super types)
The call blocks until every messageHandler has processed the message.

@param message
332211
public IMessagePublication publish(T message) {
        IMessagePublication publication = createMessagePublication(message);
        try {
            publication.execute();
        } catch (Throwable e) {
            handlePublicationError(new PublicationError()
                    .setMessage("Error during publication of message")
                    .setCause(e)
                    .setPublication(publication));
        }
        finally{
            return publication;
        }
    }
1251223
public void send(Type type, final String message) {

        final CmsNotificationMessage notificationMessage = new CmsNotificationMessage(Mode.NORMAL, type, message);
        m_messages.add(notificationMessage);
        if (hasWidget()) {
            m_widget.addMessage(notificationMessage);
        }
        Timer timer = new Timer() {

            @Override
            public void run() {

                removeMessage(notificationMessage);
            }
        };
        timer.schedule(4000 * (type == Type.NORMAL ? 1 : 2));

    }
890511
@Override
  public void publish(final LogRecord record) {
    if (record == null) {
      return;
    }

    if (!isLoggable(record)) {
      return;
    }

    synchronized (this) {
      this.logs.add(record);
      if (!this.driverInitialized || this.logs.size() < BUFFER_LEN) {
        return;
      }
    }

    logAll();
  }
1119929
private void send( String message ) {

		if( ! this.enabled.get()) {
			this.logger.finest( "Notifications were disabled by the DM." );

		} else if( message == null ) {
			this.logger.finest( "No message to send to web socket clients." );

		} else synchronized( SESSIONS ) {
			for( Session session : SESSIONS ) {
				try {
					this.logger.finest( "Sending a message to a web socket client..." );
					session.getRemote().sendString( message );

				} catch( IOException e ) {
					StringBuilder sb = new StringBuilder( "A notification could not be propagated for session " );
					sb.append( session.getRemoteAddress());
					sb.append( "." );
					if( ! Utils.isEmptyOrWhitespaces( e.getMessage()))
						sb.append( " " + e.getMessage());

					this.logger.severe( sb.toString());
					Utils.logException( this.logger, e );
				}
			}
		}
	}
624829
public void sendMessage(Message message)
    {
        try   {
            if (((RemoteMessageQueue)this.getMessageQueue()).isSendRemoteMessage((BaseMessage)message) == false)
                return;
            if (((BaseMessage)message).isProcessedByServer())
                return;     // Don't send back down (the server already processed it).
            if (((BaseMessage)message).isConsumed())
                return;     // Don't send if already handled.
            if (MessageConstants.LOCAL_QUEUE.equalsIgnoreCase(((BaseMessage)message).getMessageHeader().getQueueType()))
                return;     // Local queue only
            synchronized (m_sendQueue)
            {   // In case this is called from another task
                m_sendQueue.sendMessage(message);
            }
        } catch (RemoteException ex)    {
            ex.printStackTrace();
        }
    }
1345257
@Override
  public void write(final T message) {
    LOG.log(Level.FINEST, "write {0} :: {1}", new Object[] {channel, message});
    final ChannelFuture future = channel.writeAndFlush(Unpooled.wrappedBuffer(encoder.encode(message)));
    if (listener !=  null) {
      future.addListener(new NettyChannelFutureListener<>(message, listener));
    }
  }
501165
private void onMessageSent(MessageSentEvent event) {
        handler.post(() -> listener.onMessageSent(event));
        log("Event published " + event.toString());
    }
887414
@Override
    public void doMessageReceivedListeners(final Object message) {
        runManagementTask(new Runnable() {
            @Override
            public void run() {
                try {
                    List<SessionManagementListener> sessionListeners = getManagementListeners();
                    for (final SessionManagementListener listener : sessionListeners) {
                        listener.doMessageReceived(SessionManagementBeanImpl.this, message);
                    }

                    markChanged();
                } catch (Exception ex) {
                    logger.warn("Error during doMessageReceived session listener notifications:", ex);
                }
            }
        });
    }
2476
public void updateMessage(final Wave wave, final String message) {

        // Update the current task message
        JRebirth.runIntoJAT("Service Task Mesage => " + message,
                            () -> wave.get(JRebirthWaves.SERVICE_TASK).updateMessage(message));
    }
778508
@Override
    public void onNDArrayMessage(NDArrayMessage message) {
        try (AeronNDArrayPublisher publisher = AeronNDArrayPublisher.builder().streamId(streamId).ctx(aeronContext)
                        .channel(masterUrl).build()) {
            publisher.publish(message);
            log.debug("NDArray PublishingListener publishing to channel " + masterUrl + ":" + streamId);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
1209346
public void publishEvent(final ApplicationEvent event) {
        if (this.eventPublisher != null) {
            LOGGER.trace("Publishing event [{}]", event);
            this.eventPublisher.publishEvent(event);
        }
    }
564865
public void onMessage(MutateRowsResponse message) {
    for (Entry entry : message.getEntriesList()) {
      int index = (int) entry.getIndex();

      // Sanity check to make sure that the index returned from the server is valid.
      if (index >= mapToOriginalIndex.length || index < 0) {
        messageIsInvalid = true;
        break;
      }

      // Set the result.
      results[mapToOriginalIndex[index]] = entry.getStatus();
    }
  }
476045
public void publish(TaskEvent event) {
		synchronized (listeners) {
			for (EventListener<TaskEvent> listener : listeners.get(event.getClass())) {
				listener.onEvent(event);
			}
		}
	}
70953
public void onReceive(Object message) {

        // Start all workers
        if (message instanceof RequestToBatchSenderAsstManager) {

            // clear responseMap

            RequestToBatchSenderAsstManager request = (RequestToBatchSenderAsstManager) message;
            originalManager = getSender();
            taskId = request.getTaskId();
            asstManagerRetryIntervalMillis = request
                    .getAsstManagerRetryIntervalMillis();
            // assumption: at least 12
            taskIdTrim = taskId.length() <= 12 ? taskId
                    : taskId.substring(taskId.length() - 12,
                            taskId.length());
            workers = request.getWorkers();
            maxConcurrencyAdjusted = request.getMaxConcurrency();

            requestTotalCount = workers.size();

            sendMessageUntilStopCount(maxConcurrencyAdjusted);

            // if not completed; will schedule a continue send msg
            if (processedWorkerCount < requestTotalCount) {

                waitAndRetry();
                return;
            } else {
                logger.info("Now finished sending all needed messages. Done job of ASST Manager at "
                        + PcDateUtils.getNowDateTimeStrStandard());
                return;
            }
        } else if (message instanceof ContinueToSendToBatchSenderAsstManager) {

            // now reaching the end; have processed all of them, just waiting
            // the response to come back
            int notProcessedNodeCount = requestTotalCount
                    - processedWorkerCount;
            if (notProcessedNodeCount <= 0) {
                logger.info("!Finished sending all msg in ASST MANAGER at "
                        + PcDateUtils.getNowDateTimeStrStandard()

                        + " STOP doing wait and retry.");
                return;
            }

            int extraSendCount = maxConcurrencyAdjusted
                    - (processedWorkerCount - responseCount);

            if (extraSendCount > 0) {
                logger.info("HAVE ROOM to send extra of : " + extraSendCount
                        + " MSG. now Send at "
                        + PcDateUtils.getNowDateTimeStrStandard());

                sendMessageUntilStopCount(processedWorkerCount + extraSendCount);
                waitAndRetry();
            } else {
                logger.info("NO ROOM to send extra. Windowns is full. extraSendCount is negative: "
                        + extraSendCount
                        + " reschedule now at "
                        + PcDateUtils.getNowDateTimeStrStandard());
                waitAndRetry();

            }
        } else if (message instanceof ResponseCountToBatchSenderAsstManager) {

            responseCount = ((ResponseCountToBatchSenderAsstManager) message)
                    .getResponseCount();

            logger.debug("RECV IN batchSenderAsstManager FROM ExecutionManager responseCount: "
                    + responseCount);

        }
    }
574370
@Override
	public void callback(final AbstractMessage msg) {

		synchronized (lock) {

			hasTimeout = false;
			message = msg;

			lock.notify();
		}
	}
288692
public void onMessage(WSMessage message) {
        log.trace("Listeners: {}", listeners.size());
        for (IWebSocketDataListener listener : listeners) {
            try {
                listener.onWSMessage(message);
            } catch (Exception e) {
                log.warn("onMessage exception", e);
            }
        }
    }
1263786
public void processReceivedMessage( Message message, Session session ) throws IOException {
		this.logger.fine( "The DM's HTTP client is about to process a message (" + message + ") received through a web socket." );

		// HttpMessage
		if( message instanceof HttpMessage ) {
			HttpMessage httpMsg = (HttpMessage) message;

			// Store the session
			registerSession( httpMsg.getOwnerId(), session );

			// Publish the message
			publish( httpMsg.getCtx(), httpMsg.getMessage());
		}

		// Subscription message
		else if( message instanceof SubscriptionMessage ) {
			SubscriptionMessage sub = (SubscriptionMessage) message;

			// Store the session
			registerSession( sub.getOwnerId(), session );

			// Update the subscriptions
			if( sub.isSubscribe())
				subscribe( sub.getOwnerId(), sub.getCtx());
			else
				unsubscribe( sub.getOwnerId(), sub.getCtx());
		}
	}