java - 连接排空后打开新连接。谷歌云消息

Question

我对 Google 云消息传递有些陌生。我们已经使用它几个月了，但就在最近我们收到了“Connection Draining”消息。发生这种情况时，所有通信都会停止。

谷歌说:https://developer.android.com/google/gcm/ccs.html#response

When you receive a CONNECTION_DRAINING message, you should immediately begin sending messages to another CCS connection, opening a new connection if necessary. You should, however, keep the original connection open and continue receiving messages that may come over the connection (and ACKing them)—CCS will handle initiating a connection close when it is ready.

我的问题是

1. 如果我手动打开一个新连接，如果我不关闭现有连接，它如何知道要使用哪个连接？
2. 如果同时发送 6 条消息，我该如何阻止该方法打开 6 个连接？还是我对此感到困惑？
3. 为什么会发生连接耗尽？

我很惊讶这还没有在他们的示例代码中发挥作用。似乎它几乎是您需要的一切。它是否已在代码中为我完成，但我遗漏了它？

我的代码中没有 main 方法，而是使用 servlet 作为触发器。我的连接是这样初始化的

@PostConstruct
    public void init() throws Exception{
        try {
            smackCcsClient.connect(Long.parseLong(env.getProperty("gcm.api")), env.getProperty("gcm.key"));
        }catch (IOException e ){
            e.printStackTrace();
        }catch(SmackException e){
            e.printStackTrace();
        }catch(XMPPException e){
            e.printStackTrace();
        }
    }

但是在这之后我再也没有接触过连接。我处理这个问题了吗？连接是我应该更频繁地接触的东西还是我需要跟踪的东西？

_________________________________ 问题后添加的________________________

我在他们的示例代码中添加了一个连接以尝试重新初始化连接。它看起来像这样:

if ("CONNECTION_DRAINING".equals(controlType)) {
            connectionDraining = true;
            //Open new connection because old connection will be closing or is already closed.
            try {
                connect(Long.parseLong(env.getProperty("gcm.api")), env.getProperty("gcm.key"));
            } catch (XMPPException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (SmackException e) {
                e.printStackTrace();
            }

        } else {
            logger.log(Level.INFO, "Unrecognized control type: %s. This could happen if new features are " + "added to the CCS protocol.",
                    controlType);
        }

Answer 1

我已经编写了处理此类情况的代码(基本上将新的下游消息转移到新连接)... 未经过彻底测试...

import java.util.Deque;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedDeque;

import javax.net.ssl.SSLSocketFactory;

import org.jivesoftware.smack.ConnectionConfiguration;
import org.jivesoftware.smack.ConnectionConfiguration.SecurityMode;
import org.jivesoftware.smack.ConnectionListener;
import org.jivesoftware.smack.PacketInterceptor;
import org.jivesoftware.smack.PacketListener;
import org.jivesoftware.smack.SmackException.NotConnectedException;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.filter.PacketTypeFilter;
import org.jivesoftware.smack.packet.DefaultPacketExtension;
import org.jivesoftware.smack.packet.Message;
import org.jivesoftware.smack.packet.Packet;
import org.jivesoftware.smack.packet.PacketExtension;
import org.jivesoftware.smack.provider.PacketExtensionProvider;
import org.jivesoftware.smack.provider.ProviderManager;
import org.jivesoftware.smack.tcp.XMPPTCPConnection;
import org.jivesoftware.smack.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xmlpull.v1.XmlPullParser;

import com.fasterxml.jackson.core.type.TypeReference;


/**
 * Based on https://developer.android.com/google/gcm/ccs.html#smack
 * 
 * @author Abhinav.Dwivedi
 *
 */
public class SmackCcsClient implements CcsClient {
    private static final Logger logger = LoggerFactory.getLogger(SmackCcsClient.class);
    private static final String GCM_SERVER = "gcm.googleapis.com";
    private static final int GCM_PORT = 5235;
    private static final String GCM_ELEMENT_NAME = "gcm";
    private static final String GCM_NAMESPACE = "google:mobile:data";
    private static volatile SmackCcsClient instance;
    static {
        ProviderManager.addExtensionProvider(GCM_ELEMENT_NAME, GCM_NAMESPACE, new PacketExtensionProvider() {
            @Override
            public PacketExtension parseExtension(XmlPullParser parser) throws Exception {
                String json = parser.nextText();
                return new GcmPacketExtension(json);
            }
        });
    }
    private final Deque<Channel> channels;

    public static SmackCcsClient instance() {
        if (instance == null) {
            synchronized (SmackCcsClient.class) {
                if (instance == null) {
                    instance = new SmackCcsClient();
                }
            }
        }
        return instance;
    }

    private SmackCcsClient() {
        channels = new ConcurrentLinkedDeque<Channel>();
        channels.addFirst(connect());
    }

    private class Channel {
        private XMPPConnection connection;
        /**
         * Indicates whether the connection is in draining state, which means that it will not accept any new downstream
         * messages.
         */
        private volatile boolean connectionDraining = false;

        /**
         * Sends a packet with contents provided.
         */
        private void send(String jsonRequest) throws NotConnectedException {
            Packet request = new GcmPacketExtension(jsonRequest).toPacket();
            connection.sendPacket(request);
        }

        private void handleControlMessage(Map<String, Object> jsonObject) {
            logger.debug("handleControlMessage(): {}", jsonObject);
            String controlType = (String) jsonObject.get("control_type");
            if ("CONNECTION_DRAINING".equals(controlType)) {
                connectionDraining = true;
            } else {
                logger.info("Unrecognized control type: {}. This could happen if new features are "
                        + "added to the CCS protocol.", controlType);
            }
        }
    }

    /**
     * Sends a downstream message to GCM.
     *
     */
    @Override
    public void sendDownstreamMessage(String message) throws Exception {
        Channel channel = channels.peekFirst();
        if (channel.connectionDraining) {
            synchronized (channels) {
                channel = channels.peekFirst();
                if (channel.connectionDraining) {
                    channels.addFirst(connect());
                    channel = channels.peekFirst();
                }
            }
        }
        channel.send(message);
        logger.debug("Message Sent via CSS: ({})", message);
    }

    /**
     * Handles an upstream data message from a device application.
     *
     */
    protected void handleUpstreamMessage(Map<String, Object> jsonObject) {
        // PackageName of the application that sent this message.
        String category = (String) jsonObject.get("category");
        String from = (String) jsonObject.get("from");
        @SuppressWarnings("unchecked")
        Map<String, String> payload = (Map<String, String>) jsonObject.get("data");
        logger.info("Message received from device: category ({}), from ({}), payload: ({})", category, from,
                JsonUtil.toJson(payload));
    }

    /**
     * Handles an ACK.
     *
     * <p>
     * Logs a INFO message, but subclasses could override it to properly handle ACKs.
     */
    public void handleAckReceipt(Map<String, Object> jsonObject) {
        String messageId = (String) jsonObject.get("message_id");
        String from = (String) jsonObject.get("from");
        logger.debug("handleAckReceipt() from: {}, messageId: {}", from, messageId);
    }

    /**
     * Handles a NACK.
     *
     * <p>
     * Logs a INFO message, but subclasses could override it to properly handle NACKs.
     */
    protected void handleNackReceipt(Map<String, Object> jsonObject) {
        String messageId = (String) jsonObject.get("message_id");
        String from = (String) jsonObject.get("from");
        logger.debug("handleNackReceipt() from: {}, messageId: ", from, messageId);
    }

    /**
     * Creates a JSON encoded ACK message for an upstream message received from an application.
     *
     * @param to
     *            RegistrationId of the device who sent the upstream message.
     * @param messageId
     *            messageId of the upstream message to be acknowledged to CCS.
     * @return JSON encoded ack.
     */
    protected static String createJsonAck(String to, String messageId) {
        Map<String, Object> message = new HashMap<String, Object>();
        message.put("message_type", "ack");
        message.put("to", to);
        message.put("message_id", messageId);
        return JsonUtil.toJson(message);
    }

    /**
     * Connects to GCM Cloud Connection Server using the supplied credentials.
     * 
     * @return
     */
    @Override
    public Channel connect() {
        try {
            Channel channel = new Channel();
            ConnectionConfiguration config = new ConnectionConfiguration(GCM_SERVER, GCM_PORT);
            config.setSecurityMode(SecurityMode.enabled);
            config.setReconnectionAllowed(true);
            config.setRosterLoadedAtLogin(false);
            config.setSendPresence(false);
            config.setSocketFactory(SSLSocketFactory.getDefault());

            channel.connection = new XMPPTCPConnection(config);
            channel.connection.connect();

            channel.connection.addConnectionListener(new LoggingConnectionListener());

            // Handle incoming packets
            channel.connection.addPacketListener(new PacketListener() {
                @Override
                public void processPacket(Packet packet) {
                    logger.debug("Received: ({})", packet.toXML());
                    Message incomingMessage = (Message) packet;
                    GcmPacketExtension gcmPacket = (GcmPacketExtension) incomingMessage.getExtension(GCM_NAMESPACE);
                    String json = gcmPacket.getJson();
                    try {
                        Map<String, Object> jsonObject = JacksonUtil.DEFAULT.mapper().readValue(json,
                                new TypeReference<Map<String, Object>>() {});
                        // present for ack, nack and control, null otherwise
                        Object messageType = jsonObject.get("message_type");
                        if (messageType == null) {
                            // Normal upstream data message
                            handleUpstreamMessage(jsonObject);
                            // Send ACK to CCS
                            String messageId = (String) jsonObject.get("message_id");
                            String from = (String) jsonObject.get("from");
                            String ack = createJsonAck(from, messageId);
                            channel.send(ack);
                        } else if ("ack".equals(messageType.toString())) {
                            // Process Ack
                            handleAckReceipt(jsonObject);
                        } else if ("nack".equals(messageType.toString())) {
                            // Process Nack
                            handleNackReceipt(jsonObject);
                        } else if ("control".equals(messageType.toString())) {
                            // Process control message
                            channel.handleControlMessage(jsonObject);
                        } else {
                            logger.error("Unrecognized message type ({})", messageType.toString());
                        }
                    } catch (Exception e) {
                        logger.error("Failed to process packet ({})", packet.toXML(), e);
                    }
                }
            }, new PacketTypeFilter(Message.class));

            // Log all outgoing packets
            channel.connection.addPacketInterceptor(new PacketInterceptor() {
                @Override
                public void interceptPacket(Packet packet) {
                    logger.debug("Sent: {}", packet.toXML());
                }
            }, new PacketTypeFilter(Message.class));

            channel.connection.login(ExternalConfig.gcmSenderId() + "@gcm.googleapis.com", ExternalConfig.gcmApiKey());
            return channel;
        } catch (Exception e) {
            logger.error(Logging.FATAL, "Error in creating channel for GCM communication", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * XMPP Packet Extension for GCM Cloud Connection Server.
     */
    private static final class GcmPacketExtension extends DefaultPacketExtension {

        private final String json;

        public GcmPacketExtension(String json) {
            super(GCM_ELEMENT_NAME, GCM_NAMESPACE);
            this.json = json;
        }

        public String getJson() {
            return json;
        }

        @Override
        public String toXML() {
            return String.format("<%s xmlns=\"%s\">%s</%s>", GCM_ELEMENT_NAME, GCM_NAMESPACE,
                    StringUtils.escapeForXML(json), GCM_ELEMENT_NAME);
        }

        public Packet toPacket() {
            Message message = new Message();
            message.addExtension(this);
            return message;
        }
    }

    private static final class LoggingConnectionListener implements ConnectionListener {

        @Override
        public void connected(XMPPConnection xmppConnection) {
            logger.info("Connected.");
        }

        @Override
        public void authenticated(XMPPConnection xmppConnection) {
            logger.info("Authenticated.");
        }

        @Override
        public void reconnectionSuccessful() {
            logger.info("Reconnecting..");
        }

        @Override
        public void reconnectionFailed(Exception e) {
            logger.error("Reconnection failed.. ", e);
        }

        @Override
        public void reconnectingIn(int seconds) {
            logger.info("Reconnecting in {} secs", seconds);
        }

        @Override
        public void connectionClosedOnError(Exception e) {
            logger.info("Connection closed on error.");
        }

        @Override
        public void connectionClosed() {
            logger.info("Connection closed.");
        }
    }
}