We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
receiver doesn't correctly handle intermediate hops on fts 3.12
Motivation
----------
The last major FTS release started to correctly fill "job_state" for those messages as "ACTIVE". This is a behavioral change which makes rucio skip those messages. 

https://github.com/rucio/rucio/blob/5e911923990cc9c3caa0afbfec919c3ea4121edc/lib/rucio/daemons/conveyor/receiver.py#L74



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lib/rucio/daemons/conveyor/receiver.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # Copyright European Organization for Nuclear Research (CERN) since 2012
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #    http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 """
17 Conveyor is a daemon to manage file transfers.
18 """
19 
20 import datetime
21 import json
22 import logging
23 import socket
24 import threading
25 import time
26 import traceback
27 
28 import stomp
29 
30 import rucio.db.sqla.util
31 from rucio.common import exception
32 from rucio.common.config import config_get, config_get_bool, config_get_int
33 from rucio.common.logging import setup_logging
34 from rucio.common.policy import get_policy
35 from rucio.core import request as request_core
36 from rucio.core.monitor import record_counter
37 from rucio.core.transfer import set_transfer_update_time
38 from rucio.daemons.common import HeartbeatHandler
39 from rucio.db.sqla.session import transactional_session
40 from rucio.transfertool.fts3 import FTS3CompletionMessageTransferStatusReport
41 
42 logging.getLogger("stomp").setLevel(logging.CRITICAL)
43 
44 graceful_stop = threading.Event()
45 
46 
47 class Receiver(object):
48 
49     def __init__(self, broker, id_, total_threads, full_mode=False, all_vos=False):
50         self.__all_vos = all_vos
51         self.__broker = broker
52         self.__id = id_
53         self.__total_threads = total_threads
54         self.__full_mode = full_mode
55 
56     def on_error(self, frame):
57         record_counter('daemons.conveyor.receiver.error')
58         logging.error('[%s] %s' % (self.__broker, frame.body))
59 
60     def on_message(self, frame):
61         record_counter('daemons.conveyor.receiver.message_all')
62 
63         msg = json.loads(frame.body)
64 
65         if not self.__all_vos:
66             if 'vo' not in msg or msg['vo'] != get_policy():
67                 return
68 
69         if 'job_metadata' in msg.keys() \
70            and isinstance(msg['job_metadata'], dict) \
71            and 'issuer' in msg['job_metadata'].keys() \
72            and str(msg['job_metadata']['issuer']) == str('rucio'):
73 
74             if 'job_state' in msg.keys() and str(msg['job_state']) != str('ACTIVE'):
75                 record_counter('daemons.conveyor.receiver.message_rucio')
76 
77                 self._perform_request_update(msg)
78 
79     @transactional_session
80     def _perform_request_update(self, msg, session=None, logger=logging.log):
81         external_host = msg.get('endpnt', None)
82         request_id = msg['file_metadata'].get('request_id', None)
83         try:
84             tt_status_report = FTS3CompletionMessageTransferStatusReport(external_host, request_id=request_id, fts_message=msg)
85             if tt_status_report.get_db_fields_to_update(session=session, logger=logger):
86                 logging.info('RECEIVED %s', tt_status_report)
87 
88                 if self.__full_mode:
89                     ret = request_core.update_request_state(tt_status_report, session=session, logger=logger)
90                     record_counter('daemons.conveyor.receiver.update_request_state.{updated}', labels={'updated': ret})
91                 else:
92                     try:
93                         logging.debug("Update request %s update time" % request_id)
94                         set_transfer_update_time(external_host, tt_status_report.external_id, datetime.datetime.utcnow() - datetime.timedelta(hours=24), session=session)
95                         record_counter('daemons.conveyor.receiver.set_transfer_update_time')
96                     except Exception as error:
97                         logging.debug("Failed to update transfer's update time: %s" % str(error))
98         except Exception:
99             logging.critical(traceback.format_exc())
100 
101 
102 def receiver(id_, total_threads=1, full_mode=False, all_vos=False):
103     """
104     Main loop to consume messages from the FTS3 producer.
105     """
106 
107     logging.info('receiver starting in full mode: %s' % full_mode)
108 
109     logger_prefix = executable = 'conveyor-receiver'
110 
111     brokers_alias = []
112     brokers_resolved = []
113     try:
114         brokers_alias = [b.strip() for b in config_get('messaging-fts3', 'brokers').split(',')]
115     except Exception:
116         raise Exception('Could not load brokers from configuration')
117 
118     logging.info('resolving broker dns alias: %s' % brokers_alias)
119 
120     brokers_resolved = []
121     for broker in brokers_alias:
122         addrinfos = socket.getaddrinfo(broker, 0, socket.AF_INET, 0, socket.IPPROTO_TCP)
123         brokers_resolved.extend(ai[4][0] for ai in addrinfos)
124 
125     logging.info('brokers resolved to %s', brokers_resolved)
126 
127     logging.info('checking authentication method')
128     use_ssl = True
129     try:
130         use_ssl = config_get_bool('messaging-fts3', 'use_ssl')
131     except:
132         logging.info('could not find use_ssl in configuration -- please update your rucio.cfg')
133 
134     port = config_get_int('messaging-fts3', 'port')
135     vhost = config_get('messaging-fts3', 'broker_virtual_host', raise_exception=False)
136     if not use_ssl:
137         username = config_get('messaging-fts3', 'username')
138         password = config_get('messaging-fts3', 'password')
139         port = config_get_int('messaging-fts3', 'nonssl_port')
140 
141     conns = []
142     for broker in brokers_resolved:
143         if not use_ssl:
144             logging.info('setting up username/password authentication: %s' % broker)
145             con = stomp.Connection12(host_and_ports=[(broker, port)],
146                                      use_ssl=False,
147                                      vhost=vhost,
148                                      reconnect_attempts_max=999)
149         else:
150             logging.info('setting up ssl cert/key authentication: %s' % broker)
151             con = stomp.Connection12(host_and_ports=[(broker, port)],
152                                      use_ssl=True,
153                                      ssl_key_file=config_get('messaging-fts3', 'ssl_key_file'),
154                                      ssl_cert_file=config_get('messaging-fts3', 'ssl_cert_file'),
155                                      vhost=vhost,
156                                      reconnect_attempts_max=999)
157         conns.append(con)
158 
159     logging.info('receiver started')
160 
161     with HeartbeatHandler(executable=executable, renewal_interval=30, logger_prefix=logger_prefix) as heartbeat_handler:
162 
163         while not graceful_stop.is_set():
164 
165             _, _, logger = heartbeat_handler.live()
166 
167             for conn in conns:
168 
169                 if not conn.is_connected():
170                     logger(logging.INFO, 'connecting to %s' % conn.transport._Transport__host_and_ports[0][0])
171                     record_counter('daemons.messaging.fts3.reconnect.{host}', labels={'host': conn.transport._Transport__host_and_ports[0][0].split('.')[0]})
172 
173                     conn.set_listener('rucio-messaging-fts3', Receiver(broker=conn.transport._Transport__host_and_ports[0],
174                                                                        id_=id_, total_threads=total_threads,
175                                                                        full_mode=full_mode, all_vos=all_vos))
176                     if not use_ssl:
177                         conn.connect(username, password, wait=True)
178                     else:
179                         conn.connect(wait=True)
180                     conn.subscribe(destination=config_get('messaging-fts3', 'destination'),
181                                    id='rucio-messaging-fts3',
182                                    ack='auto')
183             time.sleep(1)
184 
185         for conn in conns:
186             try:
187                 conn.disconnect()
188             except Exception:
189                 pass
190 
191 
192 def stop(signum=None, frame=None):
193     """
194     Graceful exit.
195     """
196 
197     graceful_stop.set()
198 
199 
200 def run(once=False, total_threads=1, full_mode=False):
201     """
202     Starts up the receiver thread
203     """
204     setup_logging()
205 
206     if rucio.db.sqla.util.is_old_db():
207         raise exception.DatabaseException('Database was not updated, daemon won\'t start')
208 
209     logging.info('starting receiver thread')
210     threads = [threading.Thread(target=receiver, kwargs={'id_': i,
211                                                          'full_mode': full_mode,
212                                                          'total_threads': total_threads}) for i in range(0, total_threads)]
213 
214     [thread.start() for thread in threads]
215 
216     logging.info('waiting for interrupts')
217 
218     # Interruptible joins require a timeout.
219     while threads:
220         threads = [thread.join(timeout=3.14) for thread in threads if thread and thread.is_alive()]
221
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lib/rucio/daemons/conveyor/receiver.py b/lib/rucio/daemons/conveyor/receiver.py
--- a/lib/rucio/daemons/conveyor/receiver.py
+++ b/lib/rucio/daemons/conveyor/receiver.py
@@ -71,7 +71,7 @@
            and 'issuer' in msg['job_metadata'].keys() \
            and str(msg['job_metadata']['issuer']) == str('rucio'):
 
-            if 'job_state' in msg.keys() and str(msg['job_state']) != str('ACTIVE'):
+            if 'job_state' in msg.keys() and (str(msg['job_state']) != str('ACTIVE') or msg.get('job_multihop', False) is True):
                 record_counter('daemons.conveyor.receiver.message_rucio')
 
                 self._perform_request_update(msg)

{"golden_diff": "diff --git a/lib/rucio/daemons/conveyor/receiver.py b/lib/rucio/daemons/conveyor/receiver.py\n--- a/lib/rucio/daemons/conveyor/receiver.py\n+++ b/lib/rucio/daemons/conveyor/receiver.py\n@@ -71,7 +71,7 @@\n            and 'issuer' in msg['job_metadata'].keys() \\\n            and str(msg['job_metadata']['issuer']) == str('rucio'):\n \n-            if 'job_state' in msg.keys() and str(msg['job_state']) != str('ACTIVE'):\n+            if 'job_state' in msg.keys() and (str(msg['job_state']) != str('ACTIVE') or msg.get('job_multihop', False) is True):\n                 record_counter('daemons.conveyor.receiver.message_rucio')\n \n                 self._perform_request_update(msg)\n", "issue": "receiver doesn't correctly handle intermediate hops on fts 3.12\nMotivation\r\n----------\r\nThe last major FTS release started to correctly fill \"job_state\" for those messages as \"ACTIVE\". This is a behavioral change which makes rucio skip those messages. \r\n\r\nhttps://github.com/rucio/rucio/blob/5e911923990cc9c3caa0afbfec919c3ea4121edc/lib/rucio/daemons/conveyor/receiver.py#L74\r\n\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright European Organization for Nuclear Research (CERN) since 2012\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nConveyor is a daemon to manage file transfers.\n\"\"\"\n\nimport datetime\nimport json\nimport logging\nimport socket\nimport threading\nimport time\nimport traceback\n\nimport stomp\n\nimport rucio.db.sqla.util\nfrom rucio.common import exception\nfrom rucio.common.config import config_get, config_get_bool, config_get_int\nfrom rucio.common.logging import setup_logging\nfrom rucio.common.policy import get_policy\nfrom rucio.core import request as request_core\nfrom rucio.core.monitor import record_counter\nfrom rucio.core.transfer import set_transfer_update_time\nfrom rucio.daemons.common import HeartbeatHandler\nfrom rucio.db.sqla.session import transactional_session\nfrom rucio.transfertool.fts3 import FTS3CompletionMessageTransferStatusReport\n\nlogging.getLogger(\"stomp\").setLevel(logging.CRITICAL)\n\ngraceful_stop = threading.Event()\n\n\nclass Receiver(object):\n\n    def __init__(self, broker, id_, total_threads, full_mode=False, all_vos=False):\n        self.__all_vos = all_vos\n        self.__broker = broker\n        self.__id = id_\n        self.__total_threads = total_threads\n        self.__full_mode = full_mode\n\n    def on_error(self, frame):\n        record_counter('daemons.conveyor.receiver.error')\n        logging.error('[%s] %s' % (self.__broker, frame.body))\n\n    def on_message(self, frame):\n        record_counter('daemons.conveyor.receiver.message_all')\n\n        msg = json.loads(frame.body)\n\n        if not self.__all_vos:\n            if 'vo' not in msg or msg['vo'] != get_policy():\n                return\n\n        if 'job_metadata' in msg.keys() \\\n           and isinstance(msg['job_metadata'], dict) \\\n           and 'issuer' in msg['job_metadata'].keys() \\\n           and str(msg['job_metadata']['issuer']) == str('rucio'):\n\n            if 'job_state' in msg.keys() and str(msg['job_state']) != str('ACTIVE'):\n                record_counter('daemons.conveyor.receiver.message_rucio')\n\n                self._perform_request_update(msg)\n\n    @transactional_session\n    def _perform_request_update(self, msg, session=None, logger=logging.log):\n        external_host = msg.get('endpnt', None)\n        request_id = msg['file_metadata'].get('request_id', None)\n        try:\n            tt_status_report = FTS3CompletionMessageTransferStatusReport(external_host, request_id=request_id, fts_message=msg)\n            if tt_status_report.get_db_fields_to_update(session=session, logger=logger):\n                logging.info('RECEIVED %s', tt_status_report)\n\n                if self.__full_mode:\n                    ret = request_core.update_request_state(tt_status_report, session=session, logger=logger)\n                    record_counter('daemons.conveyor.receiver.update_request_state.{updated}', labels={'updated': ret})\n                else:\n                    try:\n                        logging.debug(\"Update request %s update time\" % request_id)\n                        set_transfer_update_time(external_host, tt_status_report.external_id, datetime.datetime.utcnow() - datetime.timedelta(hours=24), session=session)\n                        record_counter('daemons.conveyor.receiver.set_transfer_update_time')\n                    except Exception as error:\n                        logging.debug(\"Failed to update transfer's update time: %s\" % str(error))\n        except Exception:\n            logging.critical(traceback.format_exc())\n\n\ndef receiver(id_, total_threads=1, full_mode=False, all_vos=False):\n    \"\"\"\n    Main loop to consume messages from the FTS3 producer.\n    \"\"\"\n\n    logging.info('receiver starting in full mode: %s' % full_mode)\n\n    logger_prefix = executable = 'conveyor-receiver'\n\n    brokers_alias = []\n    brokers_resolved = []\n    try:\n        brokers_alias = [b.strip() for b in config_get('messaging-fts3', 'brokers').split(',')]\n    except Exception:\n        raise Exception('Could not load brokers from configuration')\n\n    logging.info('resolving broker dns alias: %s' % brokers_alias)\n\n    brokers_resolved = []\n    for broker in brokers_alias:\n        addrinfos = socket.getaddrinfo(broker, 0, socket.AF_INET, 0, socket.IPPROTO_TCP)\n        brokers_resolved.extend(ai[4][0] for ai in addrinfos)\n\n    logging.info('brokers resolved to %s', brokers_resolved)\n\n    logging.info('checking authentication method')\n    use_ssl = True\n    try:\n        use_ssl = config_get_bool('messaging-fts3', 'use_ssl')\n    except:\n        logging.info('could not find use_ssl in configuration -- please update your rucio.cfg')\n\n    port = config_get_int('messaging-fts3', 'port')\n    vhost = config_get('messaging-fts3', 'broker_virtual_host', raise_exception=False)\n    if not use_ssl:\n        username = config_get('messaging-fts3', 'username')\n        password = config_get('messaging-fts3', 'password')\n        port = config_get_int('messaging-fts3', 'nonssl_port')\n\n    conns = []\n    for broker in brokers_resolved:\n        if not use_ssl:\n            logging.info('setting up username/password authentication: %s' % broker)\n            con = stomp.Connection12(host_and_ports=[(broker, port)],\n                                     use_ssl=False,\n                                     vhost=vhost,\n                                     reconnect_attempts_max=999)\n        else:\n            logging.info('setting up ssl cert/key authentication: %s' % broker)\n            con = stomp.Connection12(host_and_ports=[(broker, port)],\n                                     use_ssl=True,\n                                     ssl_key_file=config_get('messaging-fts3', 'ssl_key_file'),\n                                     ssl_cert_file=config_get('messaging-fts3', 'ssl_cert_file'),\n                                     vhost=vhost,\n                                     reconnect_attempts_max=999)\n        conns.append(con)\n\n    logging.info('receiver started')\n\n    with HeartbeatHandler(executable=executable, renewal_interval=30, logger_prefix=logger_prefix) as heartbeat_handler:\n\n        while not graceful_stop.is_set():\n\n            _, _, logger = heartbeat_handler.live()\n\n            for conn in conns:\n\n                if not conn.is_connected():\n                    logger(logging.INFO, 'connecting to %s' % conn.transport._Transport__host_and_ports[0][0])\n                    record_counter('daemons.messaging.fts3.reconnect.{host}', labels={'host': conn.transport._Transport__host_and_ports[0][0].split('.')[0]})\n\n                    conn.set_listener('rucio-messaging-fts3', Receiver(broker=conn.transport._Transport__host_and_ports[0],\n                                                                       id_=id_, total_threads=total_threads,\n                                                                       full_mode=full_mode, all_vos=all_vos))\n                    if not use_ssl:\n                        conn.connect(username, password, wait=True)\n                    else:\n                        conn.connect(wait=True)\n                    conn.subscribe(destination=config_get('messaging-fts3', 'destination'),\n                                   id='rucio-messaging-fts3',\n                                   ack='auto')\n            time.sleep(1)\n\n        for conn in conns:\n            try:\n                conn.disconnect()\n            except Exception:\n                pass\n\n\ndef stop(signum=None, frame=None):\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n\n    graceful_stop.set()\n\n\ndef run(once=False, total_threads=1, full_mode=False):\n    \"\"\"\n    Starts up the receiver thread\n    \"\"\"\n    setup_logging()\n\n    if rucio.db.sqla.util.is_old_db():\n        raise exception.DatabaseException('Database was not updated, daemon won\\'t start')\n\n    logging.info('starting receiver thread')\n    threads = [threading.Thread(target=receiver, kwargs={'id_': i,\n                                                         'full_mode': full_mode,\n                                                         'total_threads': total_threads}) for i in range(0, total_threads)]\n\n    [thread.start() for thread in threads]\n\n    logging.info('waiting for interrupts')\n\n    # Interruptible joins require a timeout.\n    while threads:\n        threads = [thread.join(timeout=3.14) for thread in threads if thread and thread.is_alive()]\n", "path": "lib/rucio/daemons/conveyor/receiver.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright European Organization for Nuclear Research (CERN) since 2012\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nConveyor is a daemon to manage file transfers.\n\"\"\"\n\nimport datetime\nimport json\nimport logging\nimport socket\nimport threading\nimport time\nimport traceback\n\nimport stomp\n\nimport rucio.db.sqla.util\nfrom rucio.common import exception\nfrom rucio.common.config import config_get, config_get_bool, config_get_int\nfrom rucio.common.logging import setup_logging\nfrom rucio.common.policy import get_policy\nfrom rucio.core import request as request_core\nfrom rucio.core.monitor import record_counter\nfrom rucio.core.transfer import set_transfer_update_time\nfrom rucio.daemons.common import HeartbeatHandler\nfrom rucio.db.sqla.session import transactional_session\nfrom rucio.transfertool.fts3 import FTS3CompletionMessageTransferStatusReport\n\nlogging.getLogger(\"stomp\").setLevel(logging.CRITICAL)\n\ngraceful_stop = threading.Event()\n\n\nclass Receiver(object):\n\n    def __init__(self, broker, id_, total_threads, full_mode=False, all_vos=False):\n        self.__all_vos = all_vos\n        self.__broker = broker\n        self.__id = id_\n        self.__total_threads = total_threads\n        self.__full_mode = full_mode\n\n    def on_error(self, frame):\n        record_counter('daemons.conveyor.receiver.error')\n        logging.error('[%s] %s' % (self.__broker, frame.body))\n\n    def on_message(self, frame):\n        record_counter('daemons.conveyor.receiver.message_all')\n\n        msg = json.loads(frame.body)\n\n        if not self.__all_vos:\n            if 'vo' not in msg or msg['vo'] != get_policy():\n                return\n\n        if 'job_metadata' in msg.keys() \\\n           and isinstance(msg['job_metadata'], dict) \\\n           and 'issuer' in msg['job_metadata'].keys() \\\n           and str(msg['job_metadata']['issuer']) == str('rucio'):\n\n            if 'job_state' in msg.keys() and (str(msg['job_state']) != str('ACTIVE') or msg.get('job_multihop', False) is True):\n                record_counter('daemons.conveyor.receiver.message_rucio')\n\n                self._perform_request_update(msg)\n\n    @transactional_session\n    def _perform_request_update(self, msg, session=None, logger=logging.log):\n        external_host = msg.get('endpnt', None)\n        request_id = msg['file_metadata'].get('request_id', None)\n        try:\n            tt_status_report = FTS3CompletionMessageTransferStatusReport(external_host, request_id=request_id, fts_message=msg)\n            if tt_status_report.get_db_fields_to_update(session=session, logger=logger):\n                logging.info('RECEIVED %s', tt_status_report)\n\n                if self.__full_mode:\n                    ret = request_core.update_request_state(tt_status_report, session=session, logger=logger)\n                    record_counter('daemons.conveyor.receiver.update_request_state.{updated}', labels={'updated': ret})\n                else:\n                    try:\n                        logging.debug(\"Update request %s update time\" % request_id)\n                        set_transfer_update_time(external_host, tt_status_report.external_id, datetime.datetime.utcnow() - datetime.timedelta(hours=24), session=session)\n                        record_counter('daemons.conveyor.receiver.set_transfer_update_time')\n                    except Exception as error:\n                        logging.debug(\"Failed to update transfer's update time: %s\" % str(error))\n        except Exception:\n            logging.critical(traceback.format_exc())\n\n\ndef receiver(id_, total_threads=1, full_mode=False, all_vos=False):\n    \"\"\"\n    Main loop to consume messages from the FTS3 producer.\n    \"\"\"\n\n    logging.info('receiver starting in full mode: %s' % full_mode)\n\n    logger_prefix = executable = 'conveyor-receiver'\n\n    brokers_alias = []\n    brokers_resolved = []\n    try:\n        brokers_alias = [b.strip() for b in config_get('messaging-fts3', 'brokers').split(',')]\n    except Exception:\n        raise Exception('Could not load brokers from configuration')\n\n    logging.info('resolving broker dns alias: %s' % brokers_alias)\n\n    brokers_resolved = []\n    for broker in brokers_alias:\n        addrinfos = socket.getaddrinfo(broker, 0, socket.AF_INET, 0, socket.IPPROTO_TCP)\n        brokers_resolved.extend(ai[4][0] for ai in addrinfos)\n\n    logging.info('brokers resolved to %s', brokers_resolved)\n\n    logging.info('checking authentication method')\n    use_ssl = True\n    try:\n        use_ssl = config_get_bool('messaging-fts3', 'use_ssl')\n    except:\n        logging.info('could not find use_ssl in configuration -- please update your rucio.cfg')\n\n    port = config_get_int('messaging-fts3', 'port')\n    vhost = config_get('messaging-fts3', 'broker_virtual_host', raise_exception=False)\n    if not use_ssl:\n        username = config_get('messaging-fts3', 'username')\n        password = config_get('messaging-fts3', 'password')\n        port = config_get_int('messaging-fts3', 'nonssl_port')\n\n    conns = []\n    for broker in brokers_resolved:\n        if not use_ssl:\n            logging.info('setting up username/password authentication: %s' % broker)\n            con = stomp.Connection12(host_and_ports=[(broker, port)],\n                                     use_ssl=False,\n                                     vhost=vhost,\n                                     reconnect_attempts_max=999)\n        else:\n            logging.info('setting up ssl cert/key authentication: %s' % broker)\n            con = stomp.Connection12(host_and_ports=[(broker, port)],\n                                     use_ssl=True,\n                                     ssl_key_file=config_get('messaging-fts3', 'ssl_key_file'),\n                                     ssl_cert_file=config_get('messaging-fts3', 'ssl_cert_file'),\n                                     vhost=vhost,\n                                     reconnect_attempts_max=999)\n        conns.append(con)\n\n    logging.info('receiver started')\n\n    with HeartbeatHandler(executable=executable, renewal_interval=30, logger_prefix=logger_prefix) as heartbeat_handler:\n\n        while not graceful_stop.is_set():\n\n            _, _, logger = heartbeat_handler.live()\n\n            for conn in conns:\n\n                if not conn.is_connected():\n                    logger(logging.INFO, 'connecting to %s' % conn.transport._Transport__host_and_ports[0][0])\n                    record_counter('daemons.messaging.fts3.reconnect.{host}', labels={'host': conn.transport._Transport__host_and_ports[0][0].split('.')[0]})\n\n                    conn.set_listener('rucio-messaging-fts3', Receiver(broker=conn.transport._Transport__host_and_ports[0],\n                                                                       id_=id_, total_threads=total_threads,\n                                                                       full_mode=full_mode, all_vos=all_vos))\n                    if not use_ssl:\n                        conn.connect(username, password, wait=True)\n                    else:\n                        conn.connect(wait=True)\n                    conn.subscribe(destination=config_get('messaging-fts3', 'destination'),\n                                   id='rucio-messaging-fts3',\n                                   ack='auto')\n            time.sleep(1)\n\n        for conn in conns:\n            try:\n                conn.disconnect()\n            except Exception:\n                pass\n\n\ndef stop(signum=None, frame=None):\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n\n    graceful_stop.set()\n\n\ndef run(once=False, total_threads=1, full_mode=False):\n    \"\"\"\n    Starts up the receiver thread\n    \"\"\"\n    setup_logging()\n\n    if rucio.db.sqla.util.is_old_db():\n        raise exception.DatabaseException('Database was not updated, daemon won\\'t start')\n\n    logging.info('starting receiver thread')\n    threads = [threading.Thread(target=receiver, kwargs={'id_': i,\n                                                         'full_mode': full_mode,\n                                                         'total_threads': total_threads}) for i in range(0, total_threads)]\n\n    [thread.start() for thread in threads]\n\n    logging.info('waiting for interrupts')\n\n    # Interruptible joins require a timeout.\n    while threads:\n        threads = [thread.join(timeout=3.14) for thread in threads if thread and thread.is_alive()]\n", "path": "lib/rucio/daemons/conveyor/receiver.py"}]}