处理 v2 版本中 message 的 rcp 消息
@param msgId
@param conversationId
@param timestamp
351063
private void processMessageReceipt(String msgId, String conversationId, int convType, long timestamp) {
    Object messageCache =
            MessageReceiptCache.get(session.getSelfPeerId(), msgId);
    if (messageCache == null) {
      return;
    }
    Message m = (Message) messageCache;
    AVIMMessage msg =
            new AVIMMessage(conversationId, session.getSelfPeerId(), m.timestamp, timestamp);
    msg.setMessageId(m.id);
    msg.setContent(m.msg);
    msg.setMessageStatus(AVIMMessage.AVIMMessageStatus.AVIMMessageStatusReceipt);
    AVConversationHolder conversation = session.getConversationHolder(conversationId, convType);
    conversation.onMessageReceipt(msg);
  }
193956
private void processDestinationListenerCallback (CommsByteBuffer buffer, Conversation conversation) {
     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "processDestinationListenerCallback", new Object[]{buffer, conversation});

     final ClientConversationState convState = (ClientConversationState) conversation.getAttachment();
     final SICoreConnection connection = convState.getSICoreConnection();

     final short connectionObjectId                = buffer.getShort();
     final short destinationListenerId             = buffer.getShort();
     final SIDestinationAddress destinationAddress = buffer.getSIDestinationAddress(conversation.getHandshakeProperties().getFapLevel());
     final short destinationAvailabilityShort      = buffer.getShort();

     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "connectionObjectId="+connectionObjectId);
     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "destinationListenerId="+destinationListenerId);
     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "destinationAddress="+destinationAddress);

     DestinationAvailability destinationAvailability = null;
     if (destinationAvailabilityShort != CommsConstants.NO_DEST_AVAIL) {
       destinationAvailability = DestinationAvailability.getDestinationAvailability(destinationAvailabilityShort);
     }
     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "destinationAvailability="+destinationAvailability);

     // Look up the real DestinationListener in the local DestinationListenerCache

     final DestinationListenerCache destinationListenerCache = convState.getDestinationListenerCache();
     final DestinationListener destinationListener = destinationListenerCache.get(destinationListenerId);

     if (destinationListener != null) {
        //Call the listener on a seperate thread.
        ClientAsynchEventThreadPool.getInstance().dispatchDestinationListenerEvent(connection, destinationAddress, destinationAvailability, destinationListener);
     } else {
       if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "DestinationListener id="+destinationListenerId+" not found in DestinationListenerCache");
       if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, destinationListenerCache.toString());

       SIErrorException e = new SIErrorException(nls.getFormattedMessage("UNABLE_TO_FIND_DESTINATION_LISTENER_SICO8019", new Object[] {Short.valueOf(destinationListenerId)}, null));
       FFDCFilter.processException(e, CLASS_NAME + ".processDestinationListenerCallback", CommsConstants.PROXYRECEIVELISTENER_DESTLIST_CALLBACK_02, this);

       SibTr.error(tc, "UNABLE_TO_FIND_DESTINATION_LISTENER_SICO8019", e);
       throw e;
     }

     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "processDestinationListenerCallback");
   }
248868
public QuerySmResult queryShortMessage(String messageId,
            TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
            String sourceAddr) throws PDUException, ResponseTimeoutException,
            InvalidResponseException, NegativeResponseException, IOException {
        
        ensureTransmittable("queryShortMessage");
        
        QuerySmCommandTask task = new QuerySmCommandTask(pduSender(),
                messageId, sourceAddrTon, sourceAddrNpi, sourceAddr);
        
        QuerySmResp resp = (QuerySmResp)executeSendCommand(task,
                getTransactionTimer());

        if (resp.getMessageId().equals(messageId)) {
            return new QuerySmResult(resp.getFinalDate(), resp
                    .getMessageState(), resp.getErrorCode());
        } else {
            // message id requested not same as the returned
            throw new InvalidResponseException(
                    "Requested message_id doesn't match with the result");
        }
    }
606959
private void processAsyncSessionStoppedCallback (CommsByteBuffer buffer, Conversation conversation) {
     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "processAsyncSessionStoppedCallback", new Object[]{buffer, conversation});

     final short connectionObjectId = buffer.getShort();
     final short clientSessionId    = buffer.getShort();

     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "connectionObjectId="+connectionObjectId);
     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "clientSessionId="+clientSessionId);

     // Obtain the proxy queue group
     final ClientConversationState convState = (ClientConversationState) conversation.getAttachment();
     final ProxyQueueConversationGroup pqcg = convState.getProxyQueueConversationGroup();
     if (pqcg == null) {
       if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "ProxyQueueConversationGroup=null");
       SIErrorException e = new SIErrorException(nls.getFormattedMessage("NULL_PROXY_QUEUE_CONV_GROUP_CWSICO8020", new Object[] {}, null));
       FFDCFilter.processException(e, CLASS_NAME + ".processAsyncSessionStoppedCallback", CommsConstants.PROXYRECEIVELISTENER_SESSION_STOPPED_01, this);
       SibTr.error(tc, "NULL_PROXY_QUEUE_CONV_GROUP_CWSICO8020", e);
       throw e;
     }

     // Obtain the required proxy queue from the proxy queue group and ensure its of the right class (ie not read ahead)
     final ProxyQueue proxyQueue = pqcg.find(clientSessionId);
     if (proxyQueue instanceof AsynchConsumerProxyQueue) {
       final ConsumerSessionProxy consumerSessionProxy = ((AsynchConsumerProxyQueue)proxyQueue).getConsumerSessionProxy();

       //Drive the ConsumerSessionProxy.stoppableConsumerSessionStopped method on a different thread.
       ClientAsynchEventThreadPool.getInstance().dispatchStoppableConsumerSessionStopped(consumerSessionProxy);
     } else {
       if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "proxyQueue not an instance of AsynchConsumerProxyQueue is an instance of "+ proxyQueue.getClass().getName());
       SIErrorException e = new SIErrorException(nls.getFormattedMessage("WRONG_CLASS_CWSICO8021", new Object[] {proxyQueue.getClass().getName()}, null));
       FFDCFilter.processException(e, CLASS_NAME + ".processAsyncSessionStoppedCallback", CommsConstants.PROXYRECEIVELISTENER_SESSION_STOPPED_02, this);
       SibTr.error(tc, "WRONG_CLASS_CWSICO8021", e);
       throw e;
     }

     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "processAsyncSessionStoppedCallback");
   }
807944
@Override
  public void update(final GroupCommunicationMessage msg) {
    if (msg.hasSrcVersion()) {
      final String srcId = msg.getSrcid();
      final int srcVersion = msg.getSrcVersion();
      LOG.finest(getQualifiedName() + "Updating " + msg.getType() + " msg from " + srcId);
      LOG.finest(getQualifiedName() + "Before update: parent=" + ((parent != null) ? parent.getId() : "NULL"));
      LOG.finest(getQualifiedName() + "Before update: children=" + children);
      switch (msg.getType()) {
      case ParentAdd:
        updateParentAdd(srcId, srcVersion);
        break;
      case ParentDead:
        updateParentDead(srcId, srcVersion);
        break;
      case ChildAdd:
        updateChildAdd(srcId, srcVersion);
        break;
      case ChildDead:
        updateChildDead(srcId, srcVersion);
        break;
      default:
        throw new RuntimeException("Received a non control message in update");
      }
      LOG.finest(getQualifiedName() + "After update: parent=" + ((parent != null) ? parent.getId() : "NULL"));
      LOG.finest(getQualifiedName() + "After update: children=" + children);
    } else {
      throw new RuntimeException(getQualifiedName() + "can only deal with msgs that have src version set");
    }
  }
1067149
public String getConversationId() {
        return (data != null && data.payload != null) ? data.payload.conversationId : null;
    }
117756
public void onMessageEvent(javax.sip.RequestEvent event,
			ActivityContextInterface aci) {

		final Request request = event.getRequest();
		try {
			// message body should be *FIRST_TOKEN<timer value in
			// seconds>MIDDLE_TOKEN<msg to send back to UA>LAST_TOKEN*
			final String body = new String(request.getRawContent());
			final int firstTokenStart = body.indexOf(FIRST_TOKEN);
			final int timerDurationStart = firstTokenStart + FIRST_TOKEN_LENGTH;
			final int middleTokenStart = body.indexOf(MIDDLE_TOKEN,
					timerDurationStart);
			final int bodyMessageStart = middleTokenStart + MIDDLE_TOKEN_LENGTH;
			final int lastTokenStart = body.indexOf(LAST_TOKEN,
					bodyMessageStart);
			if (firstTokenStart > -1 && middleTokenStart > -1
					&& lastTokenStart > -1) {
				// extract the timer duration
				final int timerDuration = Integer.parseInt(body.substring(
						timerDurationStart, middleTokenStart));
				// create a null AC and attach the sbb local object
				final ActivityContextInterface timerACI = this.nullACIFactory
						.getActivityContextInterface(this.nullActivityFactory
								.createNullActivity());
				timerACI.attach(sbbContext.getSbbLocalObject());
				// set the timer on the null AC, because the one from this event
				// will end as soon as we send back the 200 ok
				this.timerFacility.setTimer(timerACI, null, System
						.currentTimeMillis()
						+ (timerDuration * 1000), new TimerOptions());
				// extract the body message
				final String bodyMessage = body.substring(bodyMessageStart,
						lastTokenStart);
				// store it in a cmp field
				setBody(bodyMessage);
				// do the same for the call id
				setCallId((CallIdHeader) request.getHeader(CallIdHeader.NAME));
				// also store the sender's address, so we can send the wake up
				// message
				final FromHeader fromHeader = (FromHeader) request
						.getHeader(FromHeader.NAME);
				if (tracer.isInfoEnabled()) {
					tracer.info("Received a valid message from "
							+ fromHeader.getAddress()
							+ " requesting a reply containing '" + bodyMessage
							+ "' after " + timerDuration + "s");
				}
				setSender(fromHeader.getAddress());
				// finally reply to the SIP message request
				sendResponse(event, Response.OK);
			} else {
				// parsing failed
				tracer.warning("Invalid msg '" + body + "' received");
				sendResponse(event, Response.BAD_REQUEST);
			}
		} catch (Throwable e) {
			// oh oh something wrong happened
			tracer.severe("Exception while processing MESSAGE", e);
			try {
				sendResponse(event, Response.SERVER_INTERNAL_ERROR);
			} catch (Exception f) {
				tracer.severe("Exception while sending SERVER INTERNAL ERROR",
						f);
			}
		}
	}
469063
static void rcvRequestMsgs(CommsByteBuffer request, Conversation conversation, int requestNumber,
                               boolean allocatedFromBufferPool, boolean partOfExchange)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "rcvRequestMsgs",
                        new Object[]
                        {
                         request,
                         conversation,
                         "" + requestNumber,
                         "" + allocatedFromBufferPool
                        });

        short connectionObjectID = request.getShort(); // BIT16 ConnectionObjectId
        short consumerObjectID = request.getShort(); // BIT16 SyncConsumerSessionId
        int receivedBytes = request.getInt(); // BIT32 ReceivedBytes
        int requestedBytes = request.getInt(); // BIT32 RequestedBytes

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        {
            SibTr.debug(tc, "connectionObjectID", connectionObjectID);
            SibTr.debug(tc, "consumerObjectID", consumerObjectID);
            SibTr.debug(tc, "receivedBytes", receivedBytes);
            SibTr.debug(tc, "requestedBytes", requestedBytes);
        }

        CATMainConsumer mainConsumer =
                        ((CATMainConsumer) ((ConversationState) conversation.getAttachment()).getObject(consumerObjectID));

        mainConsumer.requestMsgs(requestNumber, receivedBytes, requestedBytes);

        request.release(allocatedFromBufferPool);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "rcvRequestMsgs");
    }
865495
function(params, callbacks, recursiveCallback) {
                /**
                 * + ChatMessage        {object}
                 *    - token           {string}
                 *    - tokenIssuer     {string}
                 *    - type            {int}
                 *    - typeCode        {string}
                 *    - messageType     {int}
                 *    - subjectId       {long}
                 *    - uniqueId        {string}
                 *    - content         {string}
                 *    - time            {long}
                 *    - medadata        {string}
                 *    - systemMedadata  {string}
                 *    - repliedTo       {long}
                 */

                var asyncPriority = (params.asyncPriority > 0)
                    ? params.asyncPriority
                    : msgPriority;

                var messageVO = {
                    type: params.chatMessageVOType,
                    token: token,
                    tokenIssuer: 1
                };

                var threadId = params.subjectId;

                if (params.typeCode) {
                    messageVO.typeCode = params.typeCode;
                }
                else if (generalTypeCode) {
                    messageVO.typeCode = generalTypeCode;
                }

                if (params.messageType) {
                    messageVO.messageType = params.messageType;
                }

                if (params.subjectId) {
                    messageVO.subjectId = params.subjectId;
                }

                if (params.content) {
                    if (typeof params.content == 'object') {
                        messageVO.content = JSON.stringify(params.content);
                    }
                    else {
                        messageVO.content = params.content;
                    }
                }

                if (params.metadata) {
                    messageVO.metadata = params.metadata;
                }

                if (params.systemMetadata) {
                    messageVO.systemMetadata = params.systemMetadata;
                }

                if (params.repliedTo) {
                    messageVO.repliedTo = params.repliedTo;
                }

                var uniqueId;

                if (typeof params.uniqueId != 'undefined') {
                    uniqueId = params.uniqueId;
                }
                else if (params.chatMessageVOType !== chatMessageVOTypes.PING) {
                    uniqueId = Utility.generateUUID();
                }

                if (Array.isArray(uniqueId)) {
                    messageVO.uniqueId = JSON.stringify(uniqueId);
                }
                else {
                    messageVO.uniqueId = uniqueId;
                }

                if (typeof callbacks == 'object') {
                    if (callbacks.onSeen || callbacks.onDeliver || callbacks.onSent) {
                        if (!threadCallbacks[threadId]) {
                            threadCallbacks[threadId] = {};
                        }

                        threadCallbacks[threadId][uniqueId] = {};

                        sendMessageCallbacks[uniqueId] = {};

                        if (callbacks.onSent) {
                            sendMessageCallbacks[uniqueId].onSent = callbacks.onSent;
                            threadCallbacks[threadId][uniqueId].onSent = false;
                            threadCallbacks[threadId][uniqueId].uniqueId = uniqueId;
                        }

                        if (callbacks.onSeen) {
                            sendMessageCallbacks[uniqueId].onSeen = callbacks.onSeen;
                            threadCallbacks[threadId][uniqueId].onSeen = false;
                        }

                        if (callbacks.onDeliver) {
                            sendMessageCallbacks[uniqueId].onDeliver = callbacks.onDeliver;
                            threadCallbacks[threadId][uniqueId].onDeliver = false;
                        }

                    }
                    else if (callbacks.onResult) {
                        messagesCallbacks[uniqueId] = callbacks.onResult;
                    }
                }
                else if (typeof callbacks == 'function') {
                    messagesCallbacks[uniqueId] = callbacks;
                }

                /**
                 * Message to send through async SDK
                 *
                 * + MessageWrapperVO  {object}
                 *    - type           {int}       Type of ASYNC message based on content
                 *    + content        {string}
                 *       -peerName     {string}    Name of receiver Peer
                 *       -receivers[]  {long}      Array of receiver peer ids (if you use this, peerName will be ignored)
                 *       -priority     {int}       Priority of message 1-10, lower has more priority
                 *       -messageId    {long}      Id of message on your side, not required
                 *       -ttl          {long}      Time to live for message in milliseconds
                 *       -content      {string}    Chat Message goes here after stringifying
                 *    - trackId        {long}      Tracker id of message that you receive from DIRANA previously (if you are replying a sync message)
                 */

                var data = {
                    type: (parseInt(params.pushMsgType) > 0)
                        ? params.pushMsgType
                        : 3,
                    content: {
                        peerName: serverName,
                        priority: asyncPriority,
                        content: JSON.stringify(messageVO),
                        ttl: (params.messageTtl > 0)
                            ? params.messageTtl
                            : messageTtl
                    }
                };

                asyncClient.send(data, function(res) {
                    if (res.hasError && callbacks) {
                        if (typeof callbacks == 'function') {
                            callbacks(res);
                        }
                        else if (typeof callbacks == 'object' && typeof callbacks.onResult == 'function') {
                            callbacks.onResult(res);
                        }

                        if (messagesCallbacks[uniqueId]) {
                            delete messagesCallbacks[uniqueId];
                        }
                    }
                });

                sendPingTimeout && clearTimeout(sendPingTimeout);
                sendPingTimeout = setTimeout(function() {
                    ping();
                }, chatPingMessageInterval);

                recursiveCallback && recursiveCallback();

                return {
                    uniqueId: uniqueId,
                    threadId: threadId,
                    participant: userInfo,
                    content: params.content
                };
            }
620220
public ConversationMessage viewConversationMessage(final String messageId)
            throws NotFoundException, GeneralException, UnauthorizedException {
        String url = CONVERSATIONS_BASE_URL + CONVERSATION_MESSAGE_PATH;
        return messageBirdService.requestByID(url, messageId, ConversationMessage.class);
    }
329100
public void updateDialogMsg(final int resId) {
        if (dialog != null) {
            this.dialog.setMessage(dialog.getContext().getText(resId));
        }
    }
1068145
function(threadId, pushMessageVO) {
                /**
                 * + MessageVO                       {object}
                 *    - id                           {long}
                 *    - threadId                     {long}
                 *    - ownerId                      {long}
                 *    - uniqueId                     {string}
                 *    - previousId                   {long}
                 *    - message                      {string}
                 *    - messageType                  {int}
                 *    - edited                       {boolean}
                 *    - editable                     {boolean}
                 *    - deletable                    {boolean}
                 *    - delivered                    {boolean}
                 *    - seen                         {boolean}
                 *    - participant                  {object : ParticipantVO}
                 *    - conversation                 {object : ConversationVO}
                 *    - replyInfo                    {object : replyInfoVO}
                 *    - forwardInfo                  {object : forwardInfoVO}
                 *    - metadata                     {string}
                 *    - systemMetadata               {string}
                 *    - time                         {long}
                 *    - timeNanos                    {long}
                 */

                var message = {
                    id: pushMessageVO.id,
                    threadId: threadId,
                    ownerId: (pushMessageVO.ownerId)
                        ? pushMessageVO.ownerId
                        : undefined,
                    uniqueId: pushMessageVO.uniqueId,
                    previousId: pushMessageVO.previousId,
                    message: pushMessageVO.message,
                    messageType: pushMessageVO.messageType,
                    edited: pushMessageVO.edited,
                    editable: pushMessageVO.editable,
                    deletable: pushMessageVO.deletable,
                    delivered: pushMessageVO.delivered,
                    seen: pushMessageVO.seen,
                    participant: undefined,
                    conversation: undefined,
                    replyInfo: undefined,
                    forwardInfo: undefined,
                    metadata: pushMessageVO.metadata,
                    systemMetadata: pushMessageVO.systemMetadata,
                    time: (pushMessageVO.timeNanos)
                        ? (parseInt(parseInt(pushMessageVO.time) / 1000) * 1000000000) + parseInt(pushMessageVO.timeNanos)
                        : (parseInt(pushMessageVO.time)),
                    timeMiliSeconds: parseInt(pushMessageVO.time),
                    timeNanos: parseInt(pushMessageVO.timeNanos)
                };

                if (pushMessageVO.participant) {
                    message.ownerId = pushMessageVO.participant.id;
                }

                if (pushMessageVO.conversation) {
                    message.conversation = formatDataToMakeConversation(pushMessageVO.conversation);
                    message.threadId = pushMessageVO.conversation.id;
                }

                if (pushMessageVO.replyInfoVO) {
                    message.replyInfo = formatDataToMakeReplyInfo(pushMessageVO.replyInfoVO, threadId);
                }

                if (pushMessageVO.forwardInfo) {
                    message.forwardInfo = formatDataToMakeForwardInfo(pushMessageVO.forwardInfo, threadId);
                }

                if (pushMessageVO.participant) {
                    message.participant = formatDataToMakeParticipant(pushMessageVO.participant, threadId);
                }

                return message;
            }
699976
public void delete(long msgId) {
		String url = WxEndpoint.get("url.mass.message.delete");
		String request = "{\"msg_id\":%s}";

		logger.debug("delete message: {}", msgId);
		wxClient.post(url, String.format(request, msgId));
	}
745174
def update_time_offset(self, correct_msg_id):
        
        bad = self._get_new_msg_id()
        old = self.time_offset

        now = int(time.time())
        correct = correct_msg_id >> 32
        self.time_offset = correct - now

        if self.time_offset != old:
            self._last_msg_id = 0
            self._log.debug(
                'Updated time offset (old offset %d, bad %d, good %d, new %d)',
                old, bad, correct_msg_id, self.time_offset
            )

        return self.time_offset
511620
public void processMessageTimeout(String strTrxID)
    {
        MessageLog recMessageLog = (MessageLog)this.getMainRecord();
        BaseMessage message = (BaseMessage)recMessageLog.createMessage(strTrxID);
        String strMessageError = "Message timeout";
        BaseMessage messageReply = (BaseMessage)BaseMessageProcessor.processErrorMessage(this, message, strMessageError);
        if (messageReply != null)    // No reply if null.
        {
            BaseMessageTransport transport = this.getMessageTransport(message);
            transport.setupReplyMessage(messageReply, message, MessageInfoType.REPLY, MessageType.MESSAGE_IN);
            transport.processIncomingMessage(messageReply, message);
        }
    }
623183
private void acknowledgeQueueMessage(String messageId) {
        try {
            _queueService.acknowledge(_queueName, ImmutableList.of(messageId));
        } catch (Exception e) {
            _log.error("Failed to acknowledge message: [messageId={}]", messageId, e);
        }
    }
764905
Observable<ChatResult> synchroniseConversation(String conversationId) {

        return checkState().flatMap(client -> client.service().messaging()
                .queryMessages(conversationId, null, 1)
                .map(result -> {
                    if (result.isSuccessful() && result.getResult() != null) {
                        return (long) result.getResult().getLatestEventId();
                    }
                    return -1L;
                })
                .flatMap(result -> persistenceController.getConversation(conversationId).map(loaded -> compare(result, loaded)))
                .flatMap(this::updateLocalConversationList)
                .flatMap(result -> lookForMissingEvents(client, result))
                .map(result -> new ChatResult(result.isSuccessful, null)));
    }