Unable to make reconnecting work about stomp.py

[jyzhou17]

Hi，I'm making a durable subscription and want to reconnect when receive a heartbeat_timeout. In my reconnect function, I first call conn.disconnect(), then conn.set_listener(), conn.start(), conn.connect(), but unlike your code, in connect() I assign headers with my 'client-id', but got a 'already connected' error. It seems that when the internet is suddenly broken and quickly recovered, the disconnect() doesn't work well, and the activemq server still think the client-id is already connected.

[jasonrbriggs]

Part of the problem probably stems from the fact that you never actually break out of your loop if the reconnect is successful. I would've thought that after...

connected = True

...you should have:

connected = True
break

I'm not sure why the receiver loop is ending straight away -- but you should probably eliminate the 30 reconnects first.

[cjsoftuk]

Thanks Jason (and well spotted).

When I forced a failure, I did so by unplugging the ethernet cable from the computer (RabbitMQ server is down a VPN tunnel). This causes the host to become unreachable. When those 30 reconnect messages appeared, the ethernet cable was still unplugged.

I will add the break and then see what happens, however, I am not expecting this to be the solution as (and I apologise for not saying this initially) the time.sleep(2) never gets called. Those 30 messages print out one after the other in quick succession (as though the connect is succeeding......).

[jasonrbriggs]

Which version of the protocol are you using (1.0, 1.1 or 1.2)? If 1.2, then the explanation is a bug in the code. The connect method doesn't throw a ConnectFailedException properly if the connection fails...

[cjsoftuk]

Hi Jason,
Sorry for the delay in replying.

I am using 1.1 I believe (as I'm instantiating a Connection object, as opposed to a specific version of it).

Chris

[jasonrbriggs]

Looks like a bug. I'm working on a fix.

[cjsoftuk]

Thanks Jason - sorry I've not been terribly helpful at communicating on
this one.

Chris

On 07/01/14 00:16, jasonrbriggs wrote:

Looks like a bug. I'm working on a fix.

—
Reply to this email directly or view it on GitHub
#4 (comment).

[jasonrbriggs]

No problem.

Issue is hopefully fixed in release 4.0.7. I'm using the following code to test and it seems to work correctly:

import os
import signal
import time
import unittest

import logging
import logging.config
try:
    logging.config.fileConfig('stomp.log.conf')
except:
    pass
log = logging.getLogger('stomp.py')

import stomp
from stomp import exception

from stomp.test.testutils import *

class MyListener(ConnectionListener):
    def __init__(self, conn):
        self.conn = conn

    def on_message(self, headers, msg):
        print(headers, msg)

    def on_heartbeat_timeout(self):
        log.debug("Heartbeat timed out - attempting a reconnect")
        for n in range(1, 31):
            try:
                log.debug("Reconnecting: Attempt %d" % n)
                conn.set_listener('mylistener', self)
                self.conn.start()
                self.conn.connect('admin', 'password', wait=True)
                conn.subscribe('/queue/test', id=1)
                log.debug("Connected")
                break
            except exception.ConnectFailedException:
                _, e, _ = sys.exc_info()
                # Oh, still can't reconnect
                log.error("Reconnect attempt failed: %s" % e)
                time.sleep(2)

conn = stomp.StompConnection12([('192.168.1.92', 62613)], heartbeats=(1000,1000), reconnect_attempts_max=1)
listener = MyListener(conn)
conn.set_listener('mylistener', listener)
conn.start()
conn.connect('admin', 'password', wait=True)
conn.subscribe('/queue/test', id=1)

for x in range(1, 10000):
    if conn.is_connected():
        conn.send('/queue/test', 'hello there %s' % x)
    else:
        log.info('conn isn''t connected')
    time.sleep(2)

[bkjones]

I don't know if this test code ever worked, but it doesn't anymore, and in spite of sifting the library code and mapping out the base objects, inheritance, and condition wait/notify semantics, it's still pretty unclear to me how this is expected to actually work. Do you have an example of this code that would work in the latest stomp.py version on PyPI? Reconnection is a critical feature for anyone in a production environment and, imo, should be included in the documentation in place of the "Handling Disconnects" example, which is overly simplistic (again, imo). If I could understand how this was supposed to work, I'd probably look to issue a PR that makes reconnection "just work" if an init arg is set to True or something, at least in one of the included ConnectionListener subclasses.

The documentation should also be updated to give a full explanation/illustration of how threads are used in this library. It seems it is not possible to write a production-quality consumer without understanding it, and there is almost no mention of threading in the docs.

[bkjones]

Sorry - forgot to report my results running the code.

I can see messages passing, and heartbeats timing out, and even occasionally I can see that a reconnection attempt will fail and then I start seeing messages again, so that appears to operate as expected. However, if I stop activemq, the process immediately appears to hang. There are no stomp messages, no log messages, no exceptions - there is literally nothing. Restarting activemq does not change anything from here - still hangs. Looking on the activemq console, I see no evidence of any messages coming or going. If I kill the process, there is no output other than a KeyboardInterrupt.