message consumed treatment through provided callback and basic ack
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")
1134526
public void ack(String requestId, String resultMessage) throws TException
    {
        int index = this.requestedMap.remove(requestId);
        DRPCInvocationsClient client = this.clients.get(index);
        client.result(requestId, resultMessage);
    }
291535
void registerCallback(AsynchConsumerCallback callback)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "registerCallback", callback);

    asynchConsumerCallback = callback;

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "registerCallback");
  }
268989
def ack(self):
        
        if self.acknowledged:
            raise self.MessageStateError(
                "Message already acknowledged with state: %s" % self._state)
        self.backend.ack(self.delivery_tag)
        self._state = "ACK"
684056
async def _handle_ack(self, sid, namespace, id, data):
        """"""
        namespace = namespace or '/'
        self.logger.info('received ack from %s [%s]', sid, namespace)
        await self.manager.trigger_callback(sid, namespace, id, data)
574370
@Override
	public void callback(final AbstractMessage msg) {

		synchronized (lock) {

			hasTimeout = false;
			message = msg;

			lock.notify();
		}
	}
763005
def __recv_cb(self, msg):
        
        try:
            self.__msg_callback(msg)
        except:
            logger.exception("AmqpLink.__recv_cb exception calling msg_callback")
        finally:
            # only works if all messages handled in series
            self.__last_id = msg.delivery_tag
            self.__unacked += 1
1055845
function onMessage(obj) {
    var id = obj.id;
    if (!self._matchesPrefix(id)) {
      debug('discarding unscoped message %s', id);
      return;
    }
    var cb = self._registry.get(id);
    if (cb) {
      process.nextTick(function () {
        debug('received message %s', id);
        // Ensure that the initial callback gets called asynchronously, even
        // for completely synchronous transports (otherwise the number of
        // pending requests will sometimes be inconsistent between stateful and
        // stateless transports).
        cb(null, Buffer.concat(obj.payload), self._adapter);
      });
    }
  }
197543
def dispatch(self, receiver):
        ''' 

        '''
        super(SessionCallbackAdded, self).dispatch(receiver)
        if hasattr(receiver, '_session_callback_added'):
            receiver._session_callback_added(self)
680476
def __dispatch_msg(self, message):
        
        msg = self.__validate_decode_msg(message)
        if msg:
            msg, seqnum = msg
        else:
            self.__fire_callback(_CB_DEBUG_BAD, message.body, message.content_type)
            return

        if DEBUG_ENABLED:
            logger.debug(decode_rcvd_msg('decode_rcvd_msg', msg, seqnum))
        self.__fire_callback(_CB_DEBUG_RCVD, msg)

        # no reference, or set by client (not container)
        if msg[M_TYPE] not in _RSP_CONTAINER_REF:
            # solicitied
            if msg[M_CLIENTREF]:
                if not self.__handle_known_solicited(msg):
                    logger.debug('Ignoring response for unknown request %s of type %s', msg[M_CLIENTREF], msg[M_TYPE])
            # unsolicitied
            else:
                self.__perform_unsolicited_callbacks(msg)

        # unsolicited but can have reference set by container
        elif msg[M_TYPE] == E_CONTROLREQ:
            self.__handle_controlreq(msg[M_PAYLOAD], msg[M_CLIENTREF])

        else:
            logger.error('Unhandled unsolicited message of type %s', msg[M_TYPE])
750268
def _recv_ack(self, method_frame):
        ''''''
        if self._ack_listener:
            delivery_tag = method_frame.args.read_longlong()
            multiple = method_frame.args.read_bit()
            if multiple:
                while self._last_ack_id < delivery_tag:
                    self._last_ack_id += 1
                    self._ack_listener(self._last_ack_id)
            else:
                self._last_ack_id = delivery_tag
                self._ack_listener(self._last_ack_id)
499969
def _publisher_callback(self, publish_ack):
        
        logging.debug("ack received: " + str(publish_ack).replace('\n', ' '))
        self._rx_queue.append(publish_ack)
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
                }
            }
        }
    }
1123735
def ack_message(self, message):
        
        if message.channel.is_closed:
            LOGGER.warning('Can not ack message, channel is closed')
            self.counters[self.CLOSED_ON_COMPLETE] += 1
            return
        message.channel.basic_ack(delivery_tag=message.delivery_tag)
        self.counters[self.ACKED] += 1
        self.measurement.set_tag(self.ACKED, True)
431321
function handleIt(cb) {
    callbacks = cb;
    broadcast = callbacks.broadcast;

    let msg = {
      cmd: MessageType.RockPaperScissors,
      handlerId: id
    };
    broadcast("action", msg);

    // Return a promise for the result
    deferred = q.defer();
    return deferred.promise;
  }
29313
def _call(self, msg, **metadata):
        
        if msg.msg_type:
            for callback in self._get_callbacks(msg.msg_type):
                try:
                    callback(msg, **metadata)
                except Handler._DeadCallbackException:
                    # The callback was an upstream iterator that has been garbage
                    # collected.  Remove it from our internal structures.
                    self.remove_callback(callback)
                    self._gc_dead_sinks()
                except SystemExit:
                    raise
                except:
                    import traceback
                    traceback.print_exc()
1098048
public void updateInfoInBackground(final AVIMConversationCallback callback) {
    if (!pendingAttributes.isEmpty() || !pendingInstanceData.isEmpty()) {
      Map<String, Object> attributesMap = new HashMap<>();
      if (!pendingAttributes.isEmpty()) {
        final Map<String, Object> pendingAttrMap = processAttributes(pendingAttributes, false);
        if (null != pendingAttrMap) {
          attributesMap.putAll(pendingAttrMap);
        }
      }

      if (!pendingInstanceData.isEmpty()) {
        attributesMap.putAll(pendingInstanceData);
      }

      Map<String, Object> params = new HashMap<String, Object>();
      if (!attributesMap.isEmpty()) {
        params.put(Conversation.PARAM_CONVERSATION_ATTRIBUTE, attributesMap);
      }
      final AVIMCommonJsonCallback tmpCallback = new AVIMCommonJsonCallback() {
        @Override
        public void done(Map<String, Object> result, AVIMException e) {
          if (null == e) {
            attributes.putAll(pendingAttributes);
            pendingAttributes.clear();
            instanceData.putAll(pendingInstanceData);
            pendingInstanceData.clear();
            storage.insertConversations(Arrays.asList(AVIMConversation.this));
          }
          if (null != callback) {
            callback.internalDone(e);
          }
        }
      };
      InternalConfiguration.getOperationTube().updateConversation(this.client.getClientId(), this.conversationId, this.getType(),
              params, tmpCallback);
    } else {
      if (null != callback) {
        callback.internalDone(null);
      }
    }
  }