import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
/**
 * Build a DeliveryConfirmationMessage using the source Message
 * @param source
 * @return DeliveryConfirmationMessage if the source message represents a DeliveryConfirmation
 * null other wise.
 */
public static DeliveryConfirmationMessage build (Message source) {
  DeliveryConfirmationMessage dc = null;
  if (source != null) {
    PacketExtension confirmation = source.getExtension(Constants.XMPP_RECEIVED, Constants.XMPP_NS_RECEIPTS);
    if (confirmation != null) {
      //make sure we have the id attribute which identifies the id of the message
      //for which this message represents a delivery confirmation
      Element extensionElement = confirmation.getElement();
      String idVal = extensionElement.attributeValue(Constants.XMPP_ATTR_ID);
      if (idVal != null || !idVal.isEmpty()) {
        dc = new DeliveryConfirmationMessage();
        dc.messageId = idVal;
        dc.wrapped = source;

        JID from = source.getFrom();
        if (from != null && from.getResource() != null) {
          dc.confirmingDeviceId = from.getResource();
        }
      }
    }
  }
  return dc;
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
private MMXid[] getRecipients(PacketExtension payload) {
  Element mmxElement = payload.getElement();
  if (mmxElement == null) {
    return null;
  }
  Element element = mmxElement.element(Constants.MMX_MMXMETA);
  if (element == null) {
    return null;
  }
  MmxHeaders mmxMeta = GsonData.getGson().fromJson(element.getText(),
      MmxHeaders.class);
  List<Map<String, String>> list = (List<Map<String, String>>) mmxMeta
      .getHeader(MmxHeaders.TO, null);
  if (list == null) {
    return null;
  }
  int i = 0;
  MMXid[] recipients = new MMXid[list.size()];
  for (Map<String, String> map : list) {
    recipients[i++] = MMXid.fromMap(map);
  }
  return recipients;
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
public static boolean isPubSubMessage(Message message) {
  if (message == null) {
    return false;
  }
  final String namespace = "http://jabber.org/protocol/pubsub#event";
  final String event = "event";
  final String items = "items";

  PacketExtension extension = message.getExtension(event, namespace);
  if (extension != null) {
    Element eventElement = extension.getElement();
    if (eventElement != null) {
      Element itemElement = eventElement.element(items);
      if (itemElement != null) {
        return true;
      }
    }
  } else {
    return false;
  }
  return false;
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
private Object getMmxMeta(PacketExtension payload, String key, Object defVal) {
  Element mmxElement = payload.getElement();
  if (mmxElement == null) {
    return defVal;
  }
  Element element = mmxElement.element(Constants.MMX_MMXMETA);
  if (element == null) {
    return defVal;
  }
  MmxHeaders mmxMeta = GsonData.getGson().fromJson(element.getText(),
      MmxHeaders.class);
  return mmxMeta.getHeader(key, defVal);
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
@Test
public void shouldNotStoreEmptyChatMessagesWithOnlyChatStates() {
    Message message = new Message();
    message.setType(Message.Type.chat);
    PacketExtension chatState = new PacketExtension("composing", "http://jabber.org/protocol/chatstates");
    message.addExtension(chatState);
    assertFalse(OfflineMessageStore.shouldStoreMessage(message));
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
@Test
public void shouldNotStoreEmptyChatMessagesWithOnlyChatStatesAndThread() {
    Message message = new Message();
    message.setType(Message.Type.chat);
    message.setThread("1234");
    PacketExtension chatState = new PacketExtension("composing", "http://jabber.org/protocol/chatstates");
    message.addExtension(chatState);
    assertFalse(OfflineMessageStore.shouldStoreMessage(message));
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
@Test
public void shouldStoreEmptyChatMessagesWithOtherExtensions() {
    Message message = new Message();
    message.setType(Message.Type.chat);
    PacketExtension chatState = new PacketExtension("composing", "http://jabber.org/protocol/chatstates");
    message.addExtension(chatState);
    PacketExtension packetExtension2 = new PacketExtension("received", "urn:xmpp:receipts");
    message.addExtension(packetExtension2);
    assertTrue(OfflineMessageStore.shouldStoreMessage(message));
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
public void handleMMXMulticast(MMXMsgRuleInput input)
    throws PacketRejectedException {
  Message message = input.getMessage();
  JID mcastJid = message.getTo();
  String appId = JIDUtil.getAppId(mcastJid);
  PacketExtension payload = message.getExtension(Constants.MMX,
      Constants.MMX_NS_MSG_PAYLOAD);
  if (payload == null) {
    LOGGER.warn("Dropping a malformed MMX multicast message.");
    if (!(Boolean) getMmxMeta(payload, MmxHeaders.NO_ACK, Boolean.FALSE)) {
      sendBeginAckMessageOnce(message, appId, 0, ErrorCode.SEND_MESSAGE_MALFORMED);
    }
    throw new PacketRejectedException(
        "Stop processing for malformed multicast message");
  }

  MMXid[] recipients = getRecipients(payload);
  if (recipients == null || recipients.length == 0) {
    LOGGER.warn("No recipients found in MMX multicast message");
    if (!(Boolean) getMmxMeta(payload, MmxHeaders.NO_ACK, Boolean.FALSE)) {
      sendBeginAckMessageOnce(message, appId, 0, ErrorCode.SEND_MESSAGE_NO_TARGET);
    }
  } else {
    // Save a recipient counter for the multicast message and send a begin
    // ack. The count will be decremented when each routed message is handled
    // later.  When the count reaches zero, the end ack will be sent.  Note,
    // the packet routing in the for-loop is done asynchronously.
    if (!(Boolean) getMmxMeta(payload, MmxHeaders.NO_ACK, Boolean.FALSE)) {
      sendBeginAckMessageOnce(message, appId, recipients.length, ErrorCode.NO_ERROR);
    }

    PacketRouter pktRouter = XMPPServer.getInstance().getPacketRouter();
    for (MMXid recipient : recipients) {
      JID jid = new JID(JIDUtil.makeNode(recipient.getUserId(), appId),
          mcastJid.getDomain(), recipient.getDeviceId(), true);
      // TODO: need a deep copy because payload cannot be shared with
      // multiple messages; it has a DOM parent.
      Message unicastMsg = message.createCopy();
      unicastMsg.setTo(jid);
      pktRouter.route(unicastMsg);
    }
  }
  throw new PacketRejectedException("MMX multicast message is processed");
}
 

import org.xmpp.packet.PacketExtension; //导入依赖的package包/类
private void handleBareJID(Message message) {
  if (message.getTo().getNode() == null) {
    LOGGER.trace("handleBareJID: ignoring a multicast message={}", message);
    // It is a multicast message (XEP-0033); let MulticastRouter handle it.
    return;
  }
  // annotate the message to indicate that we have distributed it.
  MessageAnnotator annotator = new MessageDistributedAnnotator();
  annotator.annotate(message);
  LOGGER.trace("handleBareJID : message={}", message);
  MessageEntity messageEntity = MMXMessageHandlingRule
      .getMessageEntity(message);
  String domain = message.getTo().getDomain();
  String userId = JIDUtil.getUserId(message.getTo());
  MessageDistributor distributor = new MessageDistributorImpl();
  MessageDistributor.DistributionContext context = new DistributionContextImpl(
      userId, messageEntity.getAppId(), domain, messageEntity.getMessageId());
  MessageDistributor.DistributionResult result = distributor.distribute(
      message, context);
  AppDAO appDAO = DBUtil.getAppDAO();
  AppEntity appEntity = appDAO.getAppForAppKey(messageEntity.getAppId());
  List<MessageDistributor.JIDDevicePair> undistributed = result
      .getNotDistributed();
  for (MessageDistributor.JIDDevicePair pair : undistributed) {
    message.setTo(pair.getJID());
    MMXOfflineStorageUtil.storeMessage(message);
    messageEntity.setTo(message.getTo().toString());
    messageEntity.setDeviceId(pair.getJID().getResource());
    boolean wokenUpPossible = canBeWokenUp(pair.getDevice());
    if (wokenUpPossible) {
      messageEntity.setState(MessageEntity.MessageState.WAKEUP_REQUIRED);
      WakeupUtil.queueWakeup(appEntity, pair.getDevice(),
          messageEntity.getMessageId());
    } else {
      messageEntity.setState(MessageEntity.MessageState.PENDING);
    }
    DBUtil.getMessageDAO().persist(messageEntity);
  }

  if (result.noDevices()) {
    LOGGER.warn(
            "message={} addressed to user={} is dropped because the user has no active devices. Sending an error message back to originator.",
            message, userId);
  }

  PacketExtension payload = message.getExtension(Constants.MMX,
      Constants.MMX_NS_MSG_PAYLOAD);
  if (payload == null || !(Boolean) getMmxMeta(payload, MmxHeaders.NO_ACK,
      Boolean.FALSE)) {
    if (isMMXMulticastMessage(message)) {
      // Send end-ack for the last recipient in the multicast message.
      sendEndAckMessageOnce(message, appEntity.getAppId(), result.noDevices());
    } else {
      // Send ack or nack for the unicast message.
      sendServerAckMessage(message, appEntity.getAppId(), result.noDevices());
    }
  }
}