This method is called by the worker via the context. Successfully
 processed msgs are passed here.
556068
def on_msg(msg)

      return if @max_size < 1

      msg = Ruote.fulldup(msg)
      msg['seen_at'] = Ruote.now_to_utc_s

      @history << msg

      while (@history.size > @max_size) do
        @history.shift
      end

    rescue => e

      $stderr.puts '>' + '-' * 79
      $stderr.puts "#{self.class} issue, skipping"
      $stderr.puts e.inspect
      $stderr.puts e.backtrace[0, 2]
      $stderr.puts '<' + '-' * 79
    end
812900
void processAck(Message msg) {
        PubAck pa;
        Exception ex = null;
        try {
            pa = PubAck.parseFrom(msg.getData());
        } catch (InvalidProtocolBufferException e) {
            // If we are speaking to a server we don't understand, let the
            // user know.
            System.err.println("Protocol error: " +  e.getStackTrace());
            return;
        }

        // Remove
        AckClosure ackClosure = removeAck(pa.getGuid());
        if (ackClosure != null) {
            // Capture error if it exists.
            String ackError = pa.getError();

            if (ackClosure.ah != null) {
                if (!ackError.isEmpty()) {
                    ex = new IOException(ackError);
                }
                // Perform the ackHandler callback
                ackClosure.ah.onAck(pa.getGuid(), ex);
            } else if (ackClosure.ch != null) {
                try {
                    ackClosure.ch.put(ackError);
                } catch (InterruptedException e) {
                    // ignore
                }
            }
        }
    }
899202
public void handleSuccess ()
    {
        if (_listener instanceof InvocationService.ConfirmListener) {
            reportRequestProcessed();
        } else if (_listener instanceof InvocationService.ResultListener && _resultSet) {
            reportRequestProcessed(_result);
        }
    }
1205594
def _process_messages(self, messages, ignore_unknown_message_types=False):
        
        with self.before_after_send_handling():
            for message_id, message_data in messages.items():
                message_model, dispatch_models = message_data
                try:
                    message_cls = get_registered_message_type(message_model.cls)
                except UnknownMessageTypeError:
                    if ignore_unknown_message_types:
                        continue
                    raise

                message_type_cache = None
                for dispatch in dispatch_models:
                    if not dispatch.message_cache:  # Create actual message text for further usage.
                        try:
                            if message_type_cache is None and not message_cls.has_dynamic_context:
                                # If a message class doesn't depend upon a dispatch data for message compilation,
                                # we'd compile a message just once.
                                message_type_cache = message_cls.compile(message_model, self, dispatch=dispatch)

                            dispatch.message_cache = message_type_cache or message_cls.compile(
                                message_model, self, dispatch=dispatch)

                        except Exception as e:
                            self.mark_error(dispatch, e, message_cls)

                self.send(message_cls, message_model, dispatch_models)
279971
@Override
	public void processMessage( Message message ) {

		if( message instanceof MsgNotifMachineDown )
			processMsgNotifMachineDown((MsgNotifMachineDown) message );

		else if( message instanceof MsgNotifInstanceChanged )
			processMsgNotifInstanceChanged((MsgNotifInstanceChanged) message );

		else if( message instanceof MsgNotifInstanceRemoved )
			processMsgNotifInstanceRemoved((MsgNotifInstanceRemoved) message );

		else if( message instanceof MsgNotifHeartbeat )
			processMsgNotifHeartbeat((MsgNotifHeartbeat) message );

		else if(message instanceof MsgNotifAutonomic)
			processMsgMonitoringEvent((MsgNotifAutonomic) message );

		else if( message instanceof MsgEcho )
			this.manager.debugMngr().notifyMsgEchoReceived((MsgEcho) message );

		else if( message instanceof MsgNotifLogs )
			processMsgNotifLogs((MsgNotifLogs) message );

		else
			this.logger.warning( "The DM got an undetermined message to process: " + message.getClass().getName());
	}
262827
@Override
	public final void run() {

		this.running.set( true );
		while( this.running.get()) {
			try {
				Message message = this.messageQueue.take();
				if( this.running.get())
					processMessage( message );

			} catch( InterruptedException e ) {
				break;
			}
		}

		Logger.getLogger( getClass().getName()).fine( "Roboconf's message processing thread is stopping." );
		this.running.set( false );
	}
740527
protected function prepareProcessContext()
    {
        if (null !== $this->context) {
            $this->context->setData($this->data);
            $this->context->setProcessor($this);
            return;
        }

        $class = $this->processContextClass;

        $this->context = new $class($this->data, $this->settings, $this);

    }
1181334
public final void handleMessageReceived(HttpRequestContext context, long bytes) {
    checkNotNull(context, "context");
    context.receiveMessageSize.addAndGet(bytes);
    if (context.span.getOptions().contains(Options.RECORD_EVENTS)) {
      // record compressed size
      recordMessageEvent(
          context.span, context.receviedSeqId.addAndGet(1L), Type.RECEIVED, bytes, 0L);
    }
  }
984003
@Override
    public void handleMessage(Message msg) {
        if (msg.what == ASYNC) {
            mAsyncDispatcher.dispatch();
        } else if (msg.what == SYNC) {
            mSyncDispatcher.dispatch();
        } else super.handleMessage(msg);
    }
391799
def on_msg(msg)

      puts(fancy_print(msg, @noisy)) if @noisy

      return if msg['action'] == 'noop'

      @mutex.synchronize do

        @seen << msg
        @log << msg

        while @log.size > @log_max; @log.shift; end
        while @seen.size > @log_max; @seen.shift; end
      end
    end
667343
protected final void safePostMessage (final DownstreamMessage msg)
    {
        _pcmgr._omgr.postRunnable(new Runnable() {
            public void run () {
                postMessage(msg);
            }
        });
    }
284017
private synchronized void onContextStatusMessage(final ContextStatusPOJO contextStatus,
                                                   final boolean notifyClientOnNewActiveContext) {

    LOG.log(Level.FINER, "Processing context status message for context {0}", contextStatus.getContextId());
    switch (contextStatus.getContextState()) {
    case READY:
      this.onContextReady(contextStatus, notifyClientOnNewActiveContext);
      break;
    case FAIL:
      this.onContextFailed(contextStatus);
      break;
    case DONE:
      this.onContextDone(contextStatus);
      break;
    default:
      this.onUnknownContextStatus(contextStatus);
      break;
    }
    LOG.log(Level.FINER, "Done processing context status message for context {0}", contextStatus.getContextId());

  }
1223543
def consume_messages(self, batchsize):
        
        if not self._reservoir:
            self.finished = True
            return
        for msg in self._reservoir[:batchsize]:
            yield msg
        self._reservoir = self._reservoir[batchsize:]
230478
def post_send_process(context):
    
    if "error" in context:
        return context

    [deserialized_subscription] = serializers.deserialize(
        "json", context["subscription"]
    )
    subscription = deserialized_subscription.object
    [messageset] = serializers.deserialize("json", context["messageset"])
    messageset = messageset.object

    # Get set max
    set_max = messageset.messages.filter(lang=subscription.lang).count()
    logger.debug("set_max calculated - %s" % set_max)

    # Compare user position to max
    if subscription.next_sequence_number == set_max:
        with transaction.atomic():
            # Mark current as completed
            logger.debug("marking current subscription as complete")
            subscription.completed = True
            subscription.active = False
            subscription.process_status = 2  # Completed
            deserialized_subscription.save(
                update_fields=("completed", "active", "process_status")
            )
            # If next set defined create new subscription
            if messageset.next_set:
                logger.info("Creating new subscription for next set")
                newsub = Subscription.objects.create(
                    identity=subscription.identity,
                    lang=subscription.lang,
                    messageset=messageset.next_set,
                    schedule=messageset.next_set.default_schedule,
                )
                logger.debug("Created Subscription <%s>" % newsub.id)
    else:
        # More in this set so increment by one
        logger.debug("incrementing next_sequence_number")
        subscription.next_sequence_number = F("next_sequence_number") + 1
        logger.debug("setting process status back to 0")
        subscription.process_status = 0
        logger.debug("saving subscription")
        deserialized_subscription.save(
            update_fields=("next_sequence_number", "process_status")
        )
    # return response
    return "Subscription for %s updated" % str(subscription.id)
796043
@Override
	public void processMessage(WebSocketMessage webSocketData) throws FrameworkException {

		setDoTransactionNotifications(true);

		final String mode                     = webSocketData.getNodeDataStringValue("mode");		// default: list    start | pause | resume | cancel | abort
		final Long jobId                      = webSocketData.getNodeDataLongValue("jobId");

		final JobQueueManager mgr = JobQueueManager.getInstance();

		final List<GraphObject> result = new LinkedList<>();

		switch (mode) {

			case "start":
				mgr.startJob(jobId);
				break;

			case "pause":
				mgr.pauseRunningJob(jobId);
				break;

			case "resume":
				mgr.resumePausedJob(jobId);
				break;

			case "abort":
				mgr.abortActiveJob(jobId);
				break;

			case "cancel":
				mgr.cancelQueuedJob(jobId);
				break;

			case "list":
			default:
				final GraphObjectMap importsContainer = new GraphObjectMap();

				importsContainer.put(importJobsProperty, mgr.listJobs());
				result.add(importsContainer);
		}

			webSocketData.setResult(result);
			webSocketData.setRawResultCount(1);
			getWebSocket().send(webSocketData, true);

	}
705007
@Override
    public void doExecute(final TestContext context) {
        final Message message = createMessage(context, messageType);
        finished = new CompletableFuture<>();

        // extract variables from before sending message so we can save dynamic message ids
        for (VariableExtractor variableExtractor : variableExtractors) {
            variableExtractor.extractVariables(message, context);
        }

        final Endpoint messageEndpoint = getOrCreateEndpoint(context);

        if (StringUtils.hasText(message.getName())) {
            context.getMessageStore().storeMessage(message.getName(), message);
        } else {
            context.getMessageStore().storeMessage(context.getMessageStore().constructMessageName(this, messageEndpoint), message);
        }

        if (forkMode) {
            log.debug("Forking message sending action ...");

            SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
            taskExecutor.execute(() -> {
                try {
                    messageEndpoint.createProducer().send(message, context);
                } catch (Exception e) {
                    if (e instanceof CitrusRuntimeException) {
                        context.addException((CitrusRuntimeException) e);
                    } else {
                        context.addException(new CitrusRuntimeException(e));
                    }
                } finally {
                    finished.complete(null);
                }
            });
        } else {
            try {
                messageEndpoint.createProducer().send(message, context);
            } finally {
                finished.complete(null);
            }
        }
    }
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);

        }
    }