Take any actions necessary to quiesce client interaction in preparation
 for agent termination but allow message receipt to continue

 === Return
 (Array):: Number of pending non-push requests and age of youngest request
961263
def terminate
      if @offline_handler
        @offline_handler.terminate
        @connectivity_checker.terminate if @connectivity_checker
        pending = @pending_requests.kind(:send_request)
        [pending.size, PendingRequests.youngest_age(pending)]
      else
        [0, nil]
      end
    end
1268643
def handle_delayed_messages(self):
        
        for eta, message in iter_queue(self.delay_queue):
            if eta > current_millis():
                self.delay_queue.put((eta, message))
                self.delay_queue.task_done()
                break

            queue_name = q_name(message.queue_name)
            new_message = message.copy(queue_name=queue_name)
            del new_message.options["eta"]

            self.broker.enqueue(new_message)
            self.post_process_message(message)
            self.delay_queue.task_done()
1252950
private static List<DiscoveryIncomingMessage> collectIncomingMessages(int pTimeout, List<Future<List<DiscoveryIncomingMessage>>> pFutures, LogHandler pLogHandler) throws UnknownHostException {
        List<DiscoveryIncomingMessage> ret = new ArrayList<DiscoveryIncomingMessage>();
        Set<String> seen = new HashSet<String>();
        int nrCouldntSend = 0;
        for (Future<List<DiscoveryIncomingMessage>> future : pFutures) {
            try {
                List<DiscoveryIncomingMessage> inMsgs = future.get(pTimeout + 500 /* some additional buffer */, TimeUnit.MILLISECONDS);
                for (DiscoveryIncomingMessage inMsg : inMsgs) {
                    AgentDetails details = inMsg.getAgentDetails();
                    String id = details.getAgentId();
                    // There can be multiples answers with the same message id
                    if (!seen.contains(id)) {
                        ret.add(inMsg);
                        seen.add(id);
                    }
                }
            } catch (InterruptedException exp) {
                // Try next one ...
            } catch (ExecutionException e) {
                Throwable exp = e.getCause();
                if (exp instanceof CouldntSendDiscoveryPacketException) {
                    nrCouldntSend++;
                    pLogHandler.debug("--> Couldnt send discovery message from " +
                                      ((CouldntSendDiscoveryPacketException) exp).getAddress() + ": " + exp.getCause());
                }
                // Didn't worked a given address, which can happen e.g. when multicast is not routed or in other cases
                // throw new IOException("Error while performing a discovery call " + e,e);
                pLogHandler.debug("--> Exception during lookup: " + e);
            } catch (TimeoutException e) {
                // Timeout occurred while waiting for the results. So we go to the next one ...
            }
        }
        if (nrCouldntSend == pFutures.size()) {
            // No a single discovery message could be send out
            throw new UnknownHostException("Cannot send a single multicast recovery request on any multicast enabled interface");
        }
        return ret;
    }
1110816
async def serviceQueues(self, limit=None) -> int:
        
        # do not start any view changes until catch-up is finished!
        if not Mode.is_done_syncing(self.provider.node_mode()):
            return 0
        return await self.inBoxRouter.handleAll(self.inBox, limit)
1326453
def stop
      return if @state != :started
      # stop receiving new incoming messages
      @service_queue.unsubscribe
      # only stop the service if all incoming and outgoing messages are complete
      decisecond_timeout = @options[:timeout]/100
      waited_deciseconds = 0 # guarantee that this loop will stop
      while (@transactions.length > 0 || @processing_messages > 0) && waited_deciseconds < decisecond_timeout
        sleep(0.1) # wait a decisecond to check the incoming and outgoing messages again
        waited_deciseconds += 1
      end

      @channel.close
      @state = :stopped
    end
434614
def on_request(self, ch, method_frame, props, body):
        
        LOGGER.debug("rabbitmq.Service.on_request - request " + str(props) + " received")
        try:
            properties = props.headers.copy()
            properties['app_id'] = props.app_id
            properties['cluster_id'] = props.cluster_id
            properties['content_encoding'] = props.content_encoding
            properties['content_type'] = props.content_type
            properties['correlation_id'] = props.correlation_id
            properties['delivery_mode'] = props.delivery_mode
            properties['expiration'] = props.expiration
            properties['message_id'] = props.message_id
            properties['priority'] = props.priority
            properties['reply_to'] = props.reply_to
            properties['timestamp'] = props.timestamp
            properties['type'] = props.type
            properties['user_id'] = props.user_id
            self.cb(properties, body)
        except Exception as e:
            LOGGER.warn("rabbitmq.Service.on_request - Exception raised while treating msg {" +
                        str(props) + "," + str(body) + "}")
        LOGGER.debug("rabbitmq.Service.on_request - request " + str(props) + " treated")
        ch.basic_ack(delivery_tag=method_frame.delivery_tag)
        LOGGER.debug("rabbitmq.Service.on_request - request " + str(props) + " acked")
687788
def process_request(self, request, client_address):
        

        self.finish_request(request, client_address)
        self.shutdown_request(request)
494608
def on_message(self, message_protocol_entity):
        
        logger.info("Message id %s received" % message_protocol_entity.getId())
        # answer with receipt
        self.toLower(message_protocol_entity.ack())
589582
def batch_message(cls, batch, request_ids):
        ''''''
        assert isinstance(batch, Batch)
        if not cls.allow_batches:
            raise ProtocolError.invalid_request(
                'protocol does not permit batches')
        id_iter = iter(request_ids)
        rm = cls.request_message
        nm = cls.notification_message
        parts = (rm(request, next(id_iter)) if isinstance(request, Request)
                 else nm(request) for request in batch)
        return cls.batch_message_from_parts(parts)
536919
def next_message(self):
        
        if self._is_running:
            raise Exception('Event loop already running')
        if self._pending_messages:
            return self._pending_messages.popleft()
        self._async_session.run(self._enqueue_request_and_stop,
                                self._enqueue_notification_and_stop)
        if self._pending_messages:
            return self._pending_messages.popleft()
694561
async def messages(self):
        
        # Messages may be posted on the RMB while the generator is running.
        oldest_id_seen = float('inf')
        for offset in count(step=100):
            posts_bunch = await self._get_messages(offset=offset)
            for post in reversed(posts_bunch):
                if post.id < oldest_id_seen:
                    yield post
            oldest_id_seen = posts_bunch[0].id
            if len(posts_bunch) < 100:
                break
558369
def _handle_cancel_notification(self, msg_id):
        """"""
        request_future = self._client_request_futures.pop(msg_id, None)

        if not request_future:
            log.warn("Received cancel notification for unknown message id %s", msg_id)
            return

        # Will only work if the request hasn't started executing
        if request_future.cancel():
            log.debug("Cancelled request with id %s", msg_id)
621678
def get_next_action(self, request, application, roles):
        
        # Check for serious errors in submission.
        # Should only happen in rare circumstances.
        errors = application.check_valid()
        if len(errors) > 0:
            for error in errors:
                messages.error(request, error)
            return 'error'

        # approve application
        approved_by = request.user
        created_person, created_account = application.approve(approved_by)

        # send email
        application.extend()
        link, is_secret = base.get_email_link(application)
        emails.send_approved_email(
            application, created_person, created_account, link, is_secret)

        if created_person or created_account:
            return 'password_needed'
        else:
            return 'password_ok'
804446
def maybe_get_next_message(self):
        
        if self.pending:
            self.ioloop.add_callback(
                self.invoke_consumer, self.pending.popleft())
265003
def request(self, method, args, response_cb):
        
        request_id = self._next_request_id
        self._next_request_id = request_id + 1
        self._msgpack_stream.send([0, request_id, method, args])
        self._pending_requests[request_id] = response_cb
611832
def _pop_states(self, msg_id):
        
        state = self._pending_state.pop(msg_id, None)
        if state:
            return [state]

        to_pop = []
        for state in self._pending_state.values():
            if state.container_id == msg_id:
                to_pop.append(state.msg_id)

        if to_pop:
            return [self._pending_state.pop(x) for x in to_pop]

        for ack in self._last_acks:
            if ack.msg_id == msg_id:
                return [ack]

        return []
786680
def check_for_expire_acknowledge(self):
        
        if (self.acknowledgement and
                self.acknowledgement.end_time != 0 and
                self.acknowledgement.end_time < time.time()):
            self.unacknowledge_problem()