连接排空后打开新的连接。 Google云消息传递
我对Google Cloud Messaging有点新鲜。 我们一直在使用它几个月，但最近我们得到了“连接排水”信息。 发生这种情况时，所有通信都会停止
Google说：https：//developer.android.com/google/gcm/ccs.html#response
当您收到CONNECTION_DRAINING消息时，应立即开始将消息发送到另一个CCS连接，并根据需要打开一个新连接。 但是，您应该保持原始连接处于打开状态，并继续接收可能通过连接发送的消息（并确认它们）--CSCS将在准备就绪后处理关闭连接。
我的问题是
我很惊讶这个例子中的代码还没有发挥作用。 它看起来像它几乎所有你需要的东西。 它是否已经在代码中完成了，我错过了它？
我没有在我的代码中的主要方法，我用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);
}
我已经编写了一个处理这种情况的代码（基本上将新的下游消息转移到一个新的连接）......未经彻底测试......
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.");
}
}
}
我也是GCM的新手，面临同样的问题...我通过在CONNECTION_DRAINING消息上创建新的SmackCcsClient（）来解决它。 较旧的连接应该仍然存在并接收消息，但不会发送，因为：
protected volatile boolean connectionDraining = true;
Google表示CCS将关闭连接：
CCS将在准备就绪时处理启动连接。
在CCS关闭连接之前，您将能够接收来自两个连接的消息，但只能以新消息发送消息。 当旧连接关闭时，它应该被销毁，我不知道是否垃圾收集器被调用或不...试图解决这个问题
PS：我不能100％确定这个答案，但也许会为讨论开辟更多的空间。
我只是将FCM Connection Draining的代码推送到我的FCM XMPP Server示例中。
项目：用于FCM的XMPP连接服务器，使用最新版本的Smack库（4.2.2）+连接排水实施。
GitHub链接：https：//github.com/carlosCharz/fcmxmppserverv2
YouTube链接：https：//youtu.be/KVKEj6PeLTc
如果您遇到问题，请查看我的疑难解答部分。 希望你能找到它有用。 问候！
链接地址: http://www.djcxy.com/p/23487.html上一篇: Opening new connection after Connection Draining. Google Cloud Messaging