为啥我的 XMPP 客户端重新发送消息？

Posted 2023-02-24

【中文标题】为啥我的 XMPP 客户端重新发送消息？

【英文标题】：why my XMPP client resend messages?为什么我的 XMPP 客户端重新发送消息？

【发布时间】：2016-06-13 19:23:18

【问题描述】：

我在 heroku 上有一个 XMPP 客户端，它可以与 Google Cloud Messaging 一起使用，但我的应用程序有不良行为 我检查了我的代码很多次，但我没有发现任何错误，但是有些消息被重新发送，问题不在于我的消息的确认，因为我正在确认每条消息并且我没有收到来自 GCM 服务器的任何 nack 消息，所以我不知道是什么问题

如果有任何帮助，我将不胜感激

这是我的代码

SERVER = 'gcm.googleapis.com'
PORT = 5235
USERNAME = "secret"
PASSWORD = "secret"
N_TIMER=40 
EXP_TIMER=1 
unacked_messages_quota = 100
send_queue = [] 
error_send_queue = []   
lock=threading.Lock()

def unique_id():    
    return  str(uuid.uuid4().hex)

@synchronized
def message_callback(session, message):  
    global unacked_messages_quota   
    gcm = message.getTags('gcm')
    if gcm:
        gcm_json = gcm[0].getData()
        msg = json.loads(gcm_json)
    if not msg.has_key('message_type'):     
      # Acknowledge the incoming message immediately.           
      send('to': msg['from'],
            'message_type': 'ack',
            'message_id': msg['message_id'])
      # Queue a response back to the server.
      if msg.has_key('from'):
        # Send a response back to the app that sent the upstream message.
        try:            
            msg['data']['idCel'] = msg['from']
            payloadObj= payload("command", msg['data']['command'] , msg['data'])
            rpc = RpcClient()       
            response = rpc.call(payloadObj)     
            if  'response' in response and response['response'] == 'ok':                    
                pass
            elif  response['type'] == 'response'  :             
                send_queue.append('to': msg['from'],
                                    'priority':'high',
                                    'delay_while_idle':True,                                                        
                                   'message_id': unique_id(),
                                   'data': 'response': response['response'],'type': 'response'
                                   )  
            else:
                send_queue.append('to': msg['from'],                                                          
                                   'message_id': unique_id(),
                                   'data': 'error': response['error'],'type': 'error')                   
        except Exception as e:              
             traceback.print_exc()
             print str(e)                   

    elif msg['message_type'] == 'ack' or msg['message_type'] == 'nack':     
        if  msg['message_type'] == 'nack':
            error_send_queue.append(
                'to': msg['from'],
                'message_type': 'ack',
                'message_id': msg['message_id'])           
        unacked_messages_quota += 1     

def send(json_dict):
    template = ("<message><gcm xmlns='google:mobile:data'>1</gcm></message>") 
    try:
        client.send(xmpp.protocol.Message(
        node=template.format(client.Bind.bound[0], json.dumps(json_dict))))
    except Exception as e:      
        traceback.print_exc()
        print str(e)        

def flush_queued_messages():    
    global unacked_messages_quota       
    while len(send_queue) and unacked_messages_quota > 0:       
        send(send_queue.pop(0))     
        unacked_messages_quota -= 1         

def flush_queued_errors_messages():
    lock.acquire()  
    global unacked_messages_quota   
    global error_send_queue
    global EXP_TIMER    
    while len(error_send_queue) and unacked_messages_quota > 0:         
        send(error_send_queue.pop(0))
        unacked_messages_quota -= 1 
        time.sleep( (2**EXP_TIMER) )        
        EXP_TIMER += 1
    EXP_TIMER=1 
    lock.release()

client = xmpp.Client('gcm.googleapis.com',debug=['always', 'roster'],
            port=int(os.environ.get("PORT")))                             
client.connect(server=(SERVER,PORT), secure=1, use_srv=False)           
auth = client.auth(USERNAME, PASSWORD)
if not auth:
  print 'Authentication failed!'
  sys.exit(1)   
client.RegisterHandler('message', message_callback)
t1 = threading.Thread(target=flush_queued_errors_messages)      
while True:     
    client.Process(1)
    flush_queued_messages()     
    if N_TIMER == 0:
        client.send(" ")    
        N_TIMER = 40        
        if not t1.isAlive():            
            t1 = threading.Thread(target=flush_queued_errors_messages)  
            t1.start()          
    N_TIMER -= 1    

【参考方案1】：

我可以知道我真正的问题，问题不是来自我的代码，而是来自 GCM 服务器。

Here 是解释。