Browse Source

Merge remote-tracking branch 'origin/master' into resync-to-master

Change-Id: Id59697351a2f5d00d8e145e95bfe6e7a919b86f2
tags/5.9.0
Kenneth Giusti 2 years ago
parent
commit
39c3901b8c
94 changed files with 2835 additions and 1681 deletions
  1. 17
    26
      doc/source/zmq_driver.rst
  2. 32
    7
      oslo_messaging/_cmd/zmq_proxy.py
  3. 0
    2
      oslo_messaging/_drivers/amqp1_driver/opts.py
  4. 5
    2
      oslo_messaging/_drivers/base.py
  5. 12
    1
      oslo_messaging/_drivers/impl_kafka.py
  6. 0
    0
      oslo_messaging/_drivers/impl_pika.py
  7. 58
    24
      oslo_messaging/_drivers/impl_rabbit.py
  8. 9
    82
      oslo_messaging/_drivers/impl_zmq.py
  9. 4
    4
      oslo_messaging/_drivers/pika_driver/pika_connection_factory.py
  10. 45
    20
      oslo_messaging/_drivers/pool.py
  11. 0
    108
      oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_call_publisher.py
  12. 0
    91
      oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher.py
  13. 106
    0
      oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_base.py
  14. 58
    0
      oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_direct.py
  15. 52
    147
      oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_proxy.py
  16. 0
    68
      oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_reply_waiter.py
  17. 29
    133
      oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py
  18. 0
    52
      oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_push_publisher.py
  19. 111
    0
      oslo_messaging/_drivers/zmq_driver/client/zmq_ack_manager.py
  20. 32
    43
      oslo_messaging/_drivers/zmq_driver/client/zmq_client.py
  21. 19
    13
      oslo_messaging/_drivers/zmq_driver/client/zmq_client_base.py
  22. 0
    89
      oslo_messaging/_drivers/zmq_driver/client/zmq_envelope.py
  23. 180
    0
      oslo_messaging/_drivers/zmq_driver/client/zmq_receivers.py
  24. 0
    15
      oslo_messaging/_drivers/zmq_driver/client/zmq_request.py
  25. 36
    22
      oslo_messaging/_drivers/zmq_driver/client/zmq_response.py
  26. 66
    0
      oslo_messaging/_drivers/zmq_driver/client/zmq_routing_table.py
  27. 140
    0
      oslo_messaging/_drivers/zmq_driver/client/zmq_senders.py
  28. 97
    0
      oslo_messaging/_drivers/zmq_driver/client/zmq_sockets_manager.py
  29. 19
    13
      oslo_messaging/_drivers/zmq_driver/matchmaker/base.py
  30. 6
    4
      oslo_messaging/_drivers/zmq_driver/matchmaker/matchmaker_redis.py
  31. 5
    0
      oslo_messaging/_drivers/zmq_driver/poller/green_poller.py
  32. 4
    0
      oslo_messaging/_drivers/zmq_driver/poller/threading_poller.py
  33. 0
    0
      oslo_messaging/_drivers/zmq_driver/proxy/__init__.py
  34. 20
    2
      oslo_messaging/_drivers/zmq_driver/proxy/zmq_proxy.py
  35. 11
    8
      oslo_messaging/_drivers/zmq_driver/proxy/zmq_publisher_proxy.py
  36. 40
    28
      oslo_messaging/_drivers/zmq_driver/proxy/zmq_queue_proxy.py
  37. 16
    3
      oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_consumer_base.py
  38. 77
    82
      oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_dealer_consumer.py
  39. 0
    69
      oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_pull_consumer.py
  40. 32
    41
      oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_router_consumer.py
  41. 4
    21
      oslo_messaging/_drivers/zmq_driver/server/consumers/zmq_sub_consumer.py
  42. 30
    28
      oslo_messaging/_drivers/zmq_driver/server/zmq_incoming_message.py
  43. 12
    9
      oslo_messaging/_drivers/zmq_driver/server/zmq_server.py
  44. 79
    0
      oslo_messaging/_drivers/zmq_driver/server/zmq_ttl_cache.py
  45. 2
    2
      oslo_messaging/_drivers/zmq_driver/zmq_address.py
  46. 9
    0
      oslo_messaging/_drivers/zmq_driver/zmq_async.py
  47. 6
    11
      oslo_messaging/_drivers/zmq_driver/zmq_names.py
  48. 153
    0
      oslo_messaging/_drivers/zmq_driver/zmq_options.py
  49. 7
    0
      oslo_messaging/_drivers/zmq_driver/zmq_poller.py
  50. 87
    18
      oslo_messaging/_drivers/zmq_driver/zmq_socket.py
  51. 6
    1
      oslo_messaging/_drivers/zmq_driver/zmq_updater.py
  52. 4
    3
      oslo_messaging/conffixture.py
  53. 61
    0
      oslo_messaging/locale/en_GB/LC_MESSAGES/oslo_messaging-log-error.po
  54. 21
    3
      oslo_messaging/locale/en_GB/LC_MESSAGES/oslo_messaging-log-warning.po
  55. 1
    1
      oslo_messaging/notify/listener.py
  56. 46
    1
      oslo_messaging/notify/notifier.py
  57. 3
    2
      oslo_messaging/opts.py
  58. 68
    68
      oslo_messaging/tests/drivers/pika/test_message.py
  59. 20
    20
      oslo_messaging/tests/drivers/pika/test_poller.py
  60. 59
    47
      oslo_messaging/tests/drivers/test_amqp_driver.py
  61. 5
    5
      oslo_messaging/tests/drivers/test_impl_rabbit.py
  62. 5
    5
      oslo_messaging/tests/drivers/zmq/matchmaker/test_impl_matchmaker.py
  63. 2
    2
      oslo_messaging/tests/drivers/zmq/test_impl_zmq.py
  64. 35
    8
      oslo_messaging/tests/drivers/zmq/test_pub_sub.py
  65. 185
    0
      oslo_messaging/tests/drivers/zmq/test_zmq_ack_manager.py
  66. 4
    25
      oslo_messaging/tests/drivers/zmq/test_zmq_async.py
  67. 116
    0
      oslo_messaging/tests/drivers/zmq/test_zmq_ttl_cache.py
  68. 16
    9
      oslo_messaging/tests/drivers/zmq/zmq_common.py
  69. 3
    3
      oslo_messaging/tests/functional/notify/test_logger.py
  70. 147
    0
      oslo_messaging/tests/functional/test_rabbitmq.py
  71. 26
    4
      oslo_messaging/tests/functional/utils.py
  72. 2
    1
      oslo_messaging/tests/functional/zmq/multiproc_utils.py
  73. 3
    3
      oslo_messaging/tests/functional/zmq/test_startup.py
  74. 1
    1
      oslo_messaging/tests/notify/test_dispatcher.py
  75. 39
    39
      oslo_messaging/tests/notify/test_middleware.py
  76. 13
    13
      oslo_messaging/tests/rpc/test_server.py
  77. 10
    16
      oslo_messaging/tests/test_config_opts_proxy.py
  78. 2
    1
      oslo_messaging/tests/test_opts.py
  79. 5
    5
      oslo_messaging/tests/test_serializer.py
  80. 10
    8
      oslo_messaging/tests/test_transport.py
  81. 3
    0
      oslo_messaging/tests/utils.py
  82. 8
    0
      releasenotes/notes/connection_ttl-2cf0fe6e1ab8c73c.yaml
  83. 5
    0
      releasenotes/notes/option-rabbitmq-max_retries-has-been-deprecated-471f66a9e6d672a2.yaml
  84. 30
    0
      releasenotes/source/locale/en_GB/LC_MESSAGES/releasenotes.po
  85. 6
    6
      requirements.txt
  86. 0
    32
      setup-test-env-pika.sh
  87. 0
    32
      setup-test-env-rabbit.sh
  88. 37
    0
      setup-test-env-zmq-proxy.sh
  89. 32
    0
      setup-test-env-zmq-pub-sub.sh
  90. 6
    3
      setup-test-env-zmq.sh
  91. 2
    0
      setup.cfg
  92. 4
    5
      test-requirements.txt
  93. 20
    14
      tools/simulator.py
  94. 37
    7
      tox.ini

+ 17
- 26
doc/source/zmq_driver.rst View File

@@ -85,12 +85,14 @@ Configuration
85 85
 Enabling (mandatory)
86 86
 --------------------
87 87
 
88
-To enable the driver, in the section [DEFAULT] of the conf file,
89
-the 'rpc_backend' flag must be set to 'zmq' and the 'rpc_zmq_host' flag
88
+To enable the driver the 'transport_url' option must be set to 'zmq://'
89
+in the section [DEFAULT] of the conf file, the 'rpc_zmq_host' flag
90 90
 must be set to the hostname of the current node. ::
91 91
 
92 92
         [DEFAULT]
93
-        rpc_backend = zmq
93
+        transport_url = "zmq://"
94
+
95
+        [oslo_messaging_zmq]
94 96
         rpc_zmq_host = {hostname}
95 97
 
96 98
 
@@ -110,27 +112,17 @@ RedisMatchMaker: loads the hash table from a remote Redis server, supports
110 112
 dynamic host/topic registrations, host expiration, and hooks for consuming
111 113
 applications to acknowledge or neg-acknowledge topic.host service availability.
112 114
 
113
-To set the MatchMaker class, use option 'rpc_zmq_matchmaker' in [DEFAULT]. ::
114
-
115
-        rpc_zmq_matchmaker = dummy
116
-
117
-or::
118
-
119
-        rpc_zmq_matchmaker = redis
115
+For ZeroMQ driver Redis is configured in transport_url also. For using Redis
116
+specify the URL as follows::
120 117
 
121
-To specify the Redis server for RedisMatchMaker, use options in
122
-[matchmaker_redis] of each project. ::
123
-
124
-        [matchmaker_redis]
125
-        host = 127.0.0.1
126
-        port = 6379
118
+        [DEFAULT]
119
+        transport_url = "zmq+redis://127.0.0.1:6379"
127 120
 
128 121
 In order to cleanup redis storage from expired records (e.g. target listener
129 122
 goes down) TTL may be applied for keys. Configure 'zmq_target_expire' option
130 123
 which is 120 (seconds) by default. The option is related not specifically to
131
-redis so it is also defined in [DEFAULT] section. If option value is <= 0
132
-then keys don't expire and live forever in the storage.
133
-
124
+redis so it is also defined in [oslo_messaging_zmq] section. If option value
125
+is <= 0 then keys don't expire and live forever in the storage.
134 126
 
135 127
 MatchMaker Data Source (mandatory)
136 128
 ----------------------------------
@@ -159,11 +151,10 @@ we use Sentinel solution and redis master-slave-slave configuration (if we have
159 151
 3 controllers and run Redis on each of them).
160 152
 
161 153
 To deploy redis with HA follow the `sentinel-install`_ instructions. From the
162
-messaging driver's side you will need to setup following configuration which
163
-is different from a single-node redis deployment ::
154
+messaging driver's side you will need to setup following configuration ::
164 155
 
165
-        [matchmaker_redis]
166
-        sentinel_hosts=host1:26379, host2:26379, host3:26379
156
+        [DEFAULT]
157
+        transport_url = "zmq+redis://host1:26379,host2:26379,host3:26379"
167 158
 
168 159
 
169 160
 Restrict the number of TCP sockets on controller
@@ -174,7 +165,7 @@ controller node in directly connected configuration. To solve the issue
174 165
 ROUTER proxy may be used.
175 166
 
176 167
 In order to configure driver to use ROUTER proxy set up the 'use_router_proxy'
177
-option to true in [DEFAULT] section (false is set by default).
168
+option to true in [oslo_messaging_zmq] section (false is set by default).
178 169
 
179 170
 For example::
180 171
 
@@ -198,7 +189,7 @@ direct DEALER/ROUTER unicast which is possible but less efficient and therefore
198 189
 is not recommended. In a case of direct DEALER/ROUTER unicast proxy is not
199 190
 needed.
200 191
 
201
-This option can be set in [DEFAULT] section.
192
+This option can be set in [oslo_messaging_zmq] section.
202 193
 
203 194
 For example::
204 195
 
@@ -218,7 +209,7 @@ All services bind to an IP address or Ethernet adapter. By default, all services
218 209
 bind to '*', effectively binding to 0.0.0.0. This may be changed with the option
219 210
 'rpc_zmq_bind_address' which accepts a wildcard, IP address, or Ethernet adapter.
220 211
 
221
-This configuration can be set in [DEFAULT] section.
212
+This configuration can be set in [oslo_messaging_zmq] section.
222 213
 
223 214
 For example::
224 215
 

+ 32
- 7
oslo_messaging/_cmd/zmq_proxy.py View File

@@ -17,15 +17,17 @@ import logging
17 17
 
18 18
 from oslo_config import cfg
19 19
 
20
-from oslo_messaging._drivers import impl_zmq
21
-from oslo_messaging._drivers.zmq_driver.broker import zmq_proxy
22
-from oslo_messaging._drivers.zmq_driver.broker import zmq_queue_proxy
23
-from oslo_messaging import server
20
+from oslo_messaging._drivers.zmq_driver.proxy import zmq_proxy
21
+from oslo_messaging._drivers.zmq_driver.proxy import zmq_queue_proxy
22
+from oslo_messaging._drivers.zmq_driver import zmq_options
24 23
 
25 24
 CONF = cfg.CONF
26
-CONF.register_opts(impl_zmq.zmq_opts)
27
-CONF.register_opts(server._pool_opts)
28
-CONF.rpc_zmq_native = True
25
+
26
+zmq_options.register_opts(CONF)
27
+
28
+opt_group = cfg.OptGroup(name='zmq_proxy_opts',
29
+                         title='ZeroMQ proxy options')
30
+CONF.register_opts(zmq_proxy.zmq_proxy_opts, group=opt_group)
29 31
 
30 32
 
31 33
 USAGE = """ Usage: ./zmq-proxy.py [-h] [] ...
@@ -42,9 +44,20 @@ def main():
42 44
 
43 45
     parser.add_argument('--config-file', dest='config_file', type=str,
44 46
                         help='Path to configuration file')
47
+
48
+    parser.add_argument('--host', dest='host', type=str,
49
+                        help='Host FQDN for current proxy')
50
+    parser.add_argument('--frontend-port', dest='frontend_port', type=int,
51
+                        help='Front-end ROUTER port number')
52
+    parser.add_argument('--backend-port', dest='backend_port', type=int,
53
+                        help='Back-end ROUTER port number')
54
+    parser.add_argument('--publisher-port', dest='publisher_port', type=int,
55
+                        help='Back-end PUBLISHER port number')
56
+
45 57
     parser.add_argument('-d', '--debug', dest='debug', type=bool,
46 58
                         default=False,
47 59
                         help="Turn on DEBUG logging level instead of INFO")
60
+
48 61
     args = parser.parse_args()
49 62
 
50 63
     if args.config_file:
@@ -57,6 +70,18 @@ def main():
57 70
                         format='%(asctime)s %(name)s '
58 71
                                '%(levelname)-8s %(message)s')
59 72
 
73
+    if args.host:
74
+        CONF.zmq_proxy_opts.host = args.host
75
+    if args.frontend_port:
76
+        CONF.set_override('frontend_port', args.frontend_port,
77
+                          group='zmq_proxy_opts')
78
+    if args.backend_port:
79
+        CONF.set_override('backend_port', args.backend_port,
80
+                          group='zmq_proxy_opts')
81
+    if args.publisher_port:
82
+        CONF.set_override('publisher_port', args.publisher_port,
83
+                          group='zmq_proxy_opts')
84
+
60 85
     reactor = zmq_proxy.ZmqProxy(CONF, zmq_queue_proxy.UniversalQueueProxy)
61 86
 
62 87
     try:

+ 0
- 2
oslo_messaging/_drivers/amqp1_driver/opts.py View File

@@ -17,7 +17,6 @@ from oslo_config import cfg
17 17
 
18 18
 amqp1_opts = [
19 19
     cfg.StrOpt('container_name',
20
-               default=None,
21 20
                deprecated_group='amqp1',
22 21
                help='Name for the AMQP container. must be globally unique.'
23 22
                     ' Defaults to a generated UUID'),
@@ -48,7 +47,6 @@ amqp1_opts = [
48 47
                help='Private key PEM file used to sign cert_file certificate'),
49 48
 
50 49
     cfg.StrOpt('ssl_key_password',
51
-               default=None,
52 50
                deprecated_group='amqp1',
53 51
                secret=True,
54 52
                help='Password for decrypting ssl_key_file (if encrypted)'),

+ 5
- 2
oslo_messaging/_drivers/base.py View File

@@ -25,10 +25,13 @@ import six
25 25
 from oslo_messaging import exceptions
26 26
 
27 27
 base_opts = [
28
-    cfg.IntOpt('rpc_conn_pool_size',
29
-               default=30,
28
+    cfg.IntOpt('rpc_conn_pool_size', default=30,
30 29
                deprecated_group='DEFAULT',
31 30
                help='Size of RPC connection pool.'),
31
+    cfg.IntOpt('conn_pool_min_size', default=2,
32
+               help='The pool size limit for connections expiration policy'),
33
+    cfg.IntOpt('conn_pool_ttl', default=1200,
34
+               help='The time-to-live in sec of idle connections in the pool')
32 35
 ]
33 36
 
34 37
 

+ 12
- 1
oslo_messaging/_drivers/impl_kafka.py View File

@@ -50,6 +50,12 @@ kafka_opts = [
50 50
 
51 51
     cfg.IntOpt('pool_size', default=10,
52 52
                help='Pool Size for Kafka Consumers'),
53
+
54
+    cfg.IntOpt('conn_pool_min_size', default=2,
55
+               help='The pool size limit for connections expiration policy'),
56
+
57
+    cfg.IntOpt('conn_pool_ttl', default=1200,
58
+               help='The time-to-live in sec of idle connections in the pool')
53 59
 ]
54 60
 
55 61
 CONF = cfg.CONF
@@ -301,8 +307,13 @@ class KafkaDriver(base.BaseDriver):
301 307
         super(KafkaDriver, self).__init__(
302 308
             conf, url, default_exchange, allowed_remote_exmods)
303 309
 
310
+        # the pool configuration properties
311
+        max_size = self.conf.oslo_messaging_kafka.pool_size
312
+        min_size = self.conf.oslo_messaging_kafka.conn_pool_min_size
313
+        ttl = self.conf.oslo_messaging_kafka.conn_pool_ttl
314
+
304 315
         self.connection_pool = driver_pool.ConnectionPool(
305
-            self.conf, self.conf.oslo_messaging_kafka.pool_size,
316
+            self.conf, max_size, min_size, ttl,
306 317
             self._url, Connection)
307 318
         self.listeners = []
308 319
 

+ 0
- 0
oslo_messaging/_drivers/impl_pika.py View File


+ 58
- 24
oslo_messaging/_drivers/impl_rabbit.py View File

@@ -86,7 +86,7 @@ rabbit_opts = [
86 86
     cfg.IntOpt('kombu_missing_consumer_retry_timeout',
87 87
                deprecated_name="kombu_reconnect_timeout",
88 88
                default=60,
89
-               help='How long to wait a missing client beforce abandoning to '
89
+               help='How long to wait a missing client before abandoning to '
90 90
                     'send it its replies. This value should not be longer '
91 91
                     'than rpc_response_timeout.'),
92 92
     cfg.StrOpt('kombu_failover_strategy',
@@ -156,6 +156,7 @@ rabbit_opts = [
156 156
                     'Default is 30 seconds.'),
157 157
     cfg.IntOpt('rabbit_max_retries',
158 158
                default=0,
159
+               deprecated_for_removal=True,
159 160
                deprecated_group='DEFAULT',
160 161
                help='Maximum number of RabbitMQ connection retries. '
161 162
                     'Default is 0 (infinite retry count).'),
@@ -294,8 +295,8 @@ class Consumer(object):
294 295
             queue_arguments=self.queue_arguments)
295 296
 
296 297
         try:
297
-            LOG.trace('ConsumerBase.declare: '
298
-                      'queue %s', self.queue_name)
298
+            LOG.debug('[%s] Queue.declare: %s',
299
+                      conn.connection_id, self.queue_name)
299 300
             self.queue.declare()
300 301
         except conn.connection.channel_errors as exc:
301 302
             # NOTE(jrosenboom): This exception may be triggered by a race
@@ -537,6 +538,10 @@ class Connection(object):
537 538
         else:
538 539
             self._connection_lock = DummyConnectionLock()
539 540
 
541
+        self.connection_id = str(uuid.uuid4())
542
+        self.name = "%s:%d:%s" % (os.path.basename(sys.argv[0]),
543
+                                  os.getpid(),
544
+                                  self.connection_id)
540 545
         self.connection = kombu.connection.Connection(
541 546
             self._url, ssl=self._fetch_ssl_params(),
542 547
             login_method=self.login_method,
@@ -544,17 +549,21 @@ class Connection(object):
544 549
             failover_strategy=self.kombu_failover_strategy,
545 550
             transport_options={
546 551
                 'confirm_publish': True,
547
-                'client_properties': {'capabilities': {
548
-                    'authentication_failure_close': True,
549
-                    'connection.blocked': True,
550
-                    'consumer_cancel_notify': True}},
552
+                'client_properties': {
553
+                    'capabilities': {
554
+                        'authentication_failure_close': True,
555
+                        'connection.blocked': True,
556
+                        'consumer_cancel_notify': True
557
+                    },
558
+                    'connection_name': self.name},
551 559
                 'on_blocked': self._on_connection_blocked,
552 560
                 'on_unblocked': self._on_connection_unblocked,
553 561
             },
554 562
         )
555 563
 
556
-        LOG.debug('Connecting to AMQP server on %(hostname)s:%(port)s',
557
-                  self.connection.info())
564
+        LOG.debug('[%(connection_id)s] Connecting to AMQP server on'
565
+                  ' %(hostname)s:%(port)s',
566
+                  self._get_connection_info())
558 567
 
559 568
         # NOTE(sileht): kombu recommend to run heartbeat_check every
560 569
         # seconds, but we use a lock around the kombu connection
@@ -579,9 +588,10 @@ class Connection(object):
579 588
         if purpose == rpc_common.PURPOSE_SEND:
580 589
             self._heartbeat_start()
581 590
 
582
-        LOG.debug('Connected to AMQP server on %(hostname)s:%(port)s '
583
-                  'via [%(transport)s] client',
584
-                  self.connection.info())
591
+        LOG.debug('[%(connection_id)s] Connected to AMQP server on '
592
+                  '%(hostname)s:%(port)s via [%(transport)s] client with'
593
+                  ' port %(client_port)s.',
594
+                  self._get_connection_info())
585 595
 
586 596
         # NOTE(sileht): value chosen according the best practice from kombu
587 597
         # http://kombu.readthedocs.org/en/latest/reference/kombu.common.html#kombu.common.eventloop
@@ -697,7 +707,8 @@ class Connection(object):
697 707
             retry = None
698 708
 
699 709
         def on_error(exc, interval):
700
-            LOG.debug("Received recoverable error from kombu:",
710
+            LOG.debug("[%s] Received recoverable error from kombu:"
711
+                      % self.connection_id,
701 712
                       exc_info=True)
702 713
 
703 714
             recoverable_error_callback and recoverable_error_callback(exc)
@@ -707,16 +718,19 @@ class Connection(object):
707 718
                         else interval)
708 719
 
709 720
             info = {'err_str': exc, 'sleep_time': interval}
710
-            info.update(self.connection.info())
721
+            info.update(self._get_connection_info())
711 722
 
712 723
             if 'Socket closed' in six.text_type(exc):
713
-                LOG.error(_LE('AMQP server %(hostname)s:%(port)s closed'
724
+                LOG.error(_LE('[%(connection_id)s] AMQP server'
725
+                              ' %(hostname)s:%(port)s closed'
714 726
                               ' the connection. Check login credentials:'
715 727
                               ' %(err_str)s'), info)
716 728
             else:
717
-                LOG.error(_LE('AMQP server on %(hostname)s:%(port)s is '
718
-                              'unreachable: %(err_str)s. Trying again in '
719
-                              '%(sleep_time)d seconds.'), info)
729
+                LOG.error(_LE('[%(connection_id)s] AMQP server on '
730
+                              '%(hostname)s:%(port)s is unreachable: '
731
+                              '%(err_str)s. Trying again in '
732
+                              '%(sleep_time)d seconds. Client port: '
733
+                              '%(client_port)s'), info)
720 734
 
721 735
             # XXX(nic): when reconnecting to a RabbitMQ cluster
722 736
             # with mirrored queues in use, the attempt to release the
@@ -743,9 +757,10 @@ class Connection(object):
743 757
             for consumer in self._consumers:
744 758
                 consumer.declare(self)
745 759
 
746
-            LOG.info(_LI('Reconnected to AMQP server on '
747
-                         '%(hostname)s:%(port)s via [%(transport)s] client'),
748
-                     self.connection.info())
760
+            LOG.info(_LI('[%(connection_id)s] Reconnected to AMQP server on '
761
+                         '%(hostname)s:%(port)s via [%(transport)s] client'
762
+                         'with port %(client_port)s.'),
763
+                     self._get_connection_info())
749 764
 
750 765
         def execute_method(channel):
751 766
             self._set_current_channel(channel)
@@ -829,6 +844,11 @@ class Connection(object):
829 844
         """Close/release this connection."""
830 845
         self._heartbeat_stop()
831 846
         if self.connection:
847
+            for consumer, tag in self._consumers.items():
848
+                if consumer.type == 'fanout':
849
+                    LOG.debug('[connection close] Deleting fanout '
850
+                              'queue: %s ' % consumer.queue.name)
851
+                    consumer.queue.delete()
832 852
             self._set_current_channel(None)
833 853
             self.connection.release()
834 854
             self.connection = None
@@ -837,7 +857,6 @@ class Connection(object):
837 857
         """Reset a connection so it can be used again."""
838 858
         recoverable_errors = (self.connection.recoverable_channel_errors +
839 859
                               self.connection.recoverable_connection_errors)
840
-
841 860
         with self._connection_lock:
842 861
             try:
843 862
                 for consumer, tag in self._consumers.items():
@@ -885,7 +904,8 @@ class Connection(object):
885 904
             sock = self.channel.connection.sock
886 905
         except AttributeError as e:
887 906
             # Level is set to debug because otherwise we would spam the logs
888
-            LOG.debug('Failed to get socket attribute: %s' % str(e))
907
+            LOG.debug('[%s] Failed to get socket attribute: %s'
908
+                      % (self.connection_id, str(e)))
889 909
         else:
890 910
             sock.settimeout(timeout)
891 911
             # TCP_USER_TIMEOUT is not defined on Windows and Mac OS X
@@ -1141,6 +1161,15 @@ class Connection(object):
1141 1161
         with self._connection_lock:
1142 1162
             self.ensure(method, retry=retry, error_callback=_error_callback)
1143 1163
 
1164
+    def _get_connection_info(self):
1165
+        info = self.connection.info()
1166
+        client_port = None
1167
+        if self.channel and hasattr(self.channel.connection, 'sock'):
1168
+            client_port = self.channel.connection.sock.getsockname()[1]
1169
+        info.update({'client_port': client_port,
1170
+                     'connection_id': self.connection_id})
1171
+        return info
1172
+
1144 1173
     def _publish(self, exchange, msg, routing_key=None, timeout=None):
1145 1174
         """Publish a message."""
1146 1175
 
@@ -1296,8 +1325,13 @@ class RabbitDriver(amqpdriver.AMQPDriverBase):
1296 1325
         self.prefetch_size = (
1297 1326
             conf.oslo_messaging_rabbit.rabbit_qos_prefetch_count)
1298 1327
 
1328
+        # the pool configuration properties
1329
+        max_size = conf.oslo_messaging_rabbit.rpc_conn_pool_size
1330
+        min_size = conf.oslo_messaging_rabbit.conn_pool_min_size
1331
+        ttl = conf.oslo_messaging_rabbit.conn_pool_ttl
1332
+
1299 1333
         connection_pool = pool.ConnectionPool(
1300
-            conf, conf.oslo_messaging_rabbit.rpc_conn_pool_size,
1334
+            conf, max_size, min_size, ttl,
1301 1335
             url, Connection)
1302 1336
 
1303 1337
         super(RabbitDriver, self).__init__(

+ 9
- 82
oslo_messaging/_drivers/impl_zmq.py View File

@@ -14,10 +14,8 @@
14 14
 
15 15
 import logging
16 16
 import os
17
-import socket
18 17
 import threading
19 18
 
20
-from oslo_config import cfg
21 19
 from stevedore import driver
22 20
 
23 21
 from oslo_messaging._drivers import base
@@ -25,85 +23,14 @@ from oslo_messaging._drivers import common as rpc_common
25 23
 from oslo_messaging._drivers.zmq_driver.client import zmq_client
26 24
 from oslo_messaging._drivers.zmq_driver.server import zmq_server
27 25
 from oslo_messaging._drivers.zmq_driver import zmq_async
26
+from oslo_messaging._drivers.zmq_driver import zmq_options
28 27
 from oslo_messaging._i18n import _LE
29
-from oslo_messaging import server
30 28
 
31 29
 
32 30
 RPCException = rpc_common.RPCException
33
-_MATCHMAKER_BACKENDS = ('redis', 'dummy')
34
-_MATCHMAKER_DEFAULT = 'redis'
35 31
 LOG = logging.getLogger(__name__)
36 32
 
37 33
 
38
-zmq_opts = [
39
-    cfg.StrOpt('rpc_zmq_bind_address', default='*',
40
-               help='ZeroMQ bind address. Should be a wildcard (*), '
41
-                    'an ethernet interface, or IP. '
42
-                    'The "host" option should point or resolve to this '
43
-                    'address.'),
44
-
45
-    cfg.StrOpt('rpc_zmq_matchmaker', default=_MATCHMAKER_DEFAULT,
46
-               choices=_MATCHMAKER_BACKENDS,
47
-               help='MatchMaker driver.'),
48
-
49
-    cfg.IntOpt('rpc_zmq_contexts', default=1,
50
-               help='Number of ZeroMQ contexts, defaults to 1.'),
51
-
52
-    cfg.IntOpt('rpc_zmq_topic_backlog',
53
-               help='Maximum number of ingress messages to locally buffer '
54
-                    'per topic. Default is unlimited.'),
55
-
56
-    cfg.StrOpt('rpc_zmq_ipc_dir', default='/var/run/openstack',
57
-               help='Directory for holding IPC sockets.'),
58
-
59
-    cfg.StrOpt('rpc_zmq_host', default=socket.gethostname(),
60
-               sample_default='localhost',
61
-               help='Name of this node. Must be a valid hostname, FQDN, or '
62
-                    'IP address. Must match "host" option, if running Nova.'),
63
-
64
-    cfg.IntOpt('rpc_cast_timeout', default=-1,
65
-               help='Seconds to wait before a cast expires (TTL). '
66
-                    'The default value of -1 specifies an infinite linger '
67
-                    'period. The value of 0 specifies no linger period. '
68
-                    'Pending messages shall be discarded immediately '
69
-                    'when the socket is closed. Only supported by impl_zmq.'),
70
-
71
-    cfg.IntOpt('rpc_poll_timeout', default=1,
72
-               help='The default number of seconds that poll should wait. '
73
-                    'Poll raises timeout exception when timeout expired.'),
74
-
75
-    cfg.IntOpt('zmq_target_expire', default=300,
76
-               help='Expiration timeout in seconds of a name service record '
77
-                    'about existing target ( < 0 means no timeout).'),
78
-
79
-    cfg.IntOpt('zmq_target_update', default=180,
80
-               help='Update period in seconds of a name service record '
81
-                    'about existing target.'),
82
-
83
-    cfg.BoolOpt('use_pub_sub', default=True,
84
-                help='Use PUB/SUB pattern for fanout methods. '
85
-                     'PUB/SUB always uses proxy.'),
86
-
87
-    cfg.BoolOpt('use_router_proxy', default=True,
88
-                help='Use ROUTER remote proxy.'),
89
-
90
-    cfg.PortOpt('rpc_zmq_min_port',
91
-                default=49153,
92
-                help='Minimal port number for random ports range.'),
93
-
94
-    cfg.IntOpt('rpc_zmq_max_port',
95
-               min=1,
96
-               max=65536,
97
-               default=65536,
98
-               help='Maximal port number for random ports range.'),
99
-
100
-    cfg.IntOpt('rpc_zmq_bind_port_retries',
101
-               default=100,
102
-               help='Number of retries to find free port number before '
103
-                    'fail with ZMQBindError.')
104
-]
105
-
106
-
107 34
 class LazyDriverItem(object):
108 35
 
109 36
     def __init__(self, item_cls, *args, **kwargs):
@@ -169,9 +96,7 @@ class ZmqDriver(base.BaseDriver):
169 96
         if zmq is None:
170 97
             raise ImportError(_LE("ZeroMQ is not available!"))
171 98
 
172
-        conf.register_opts(zmq_opts)
173
-        conf.register_opts(server._pool_opts)
174
-        conf.register_opts(base.base_opts)
99
+        zmq_options.register_opts(conf)
175 100
         self.conf = conf
176 101
         self.allowed_remote_exmods = allowed_remote_exmods
177 102
 
@@ -181,9 +106,11 @@ class ZmqDriver(base.BaseDriver):
181 106
         ).driver(self.conf, url=url)
182 107
 
183 108
         client_cls = zmq_client.ZmqClientProxy
184
-        if conf.use_pub_sub and not conf.use_router_proxy:
109
+        if conf.oslo_messaging_zmq.use_pub_sub and not \
110
+                conf.oslo_messaging_zmq.use_router_proxy:
185 111
             client_cls = zmq_client.ZmqClientMixDirectPubSub
186
-        elif not conf.use_pub_sub and not conf.use_router_proxy:
112
+        elif not conf.oslo_messaging_zmq.use_pub_sub and not \
113
+                conf.oslo_messaging_zmq.use_router_proxy:
187 114
             client_cls = zmq_client.ZmqClientDirect
188 115
 
189 116
         self.client = LazyDriverItem(
@@ -201,13 +128,13 @@ class ZmqDriver(base.BaseDriver):
201 128
         zmq_transport, p, matchmaker_backend = url.transport.partition('+')
202 129
         assert zmq_transport == 'zmq', "Needs to be zmq for this transport!"
203 130
         if not matchmaker_backend:
204
-            return self.conf.rpc_zmq_matchmaker
205
-        elif matchmaker_backend not in _MATCHMAKER_BACKENDS:
131
+            return self.conf.oslo_messaging_zmq.rpc_zmq_matchmaker
132
+        elif matchmaker_backend not in zmq_options.MATCHMAKER_BACKENDS:
206 133
             raise rpc_common.RPCException(
207 134
                 _LE("Incorrect matchmaker backend name %(backend_name)s!"
208 135
                     "Available names are: %(available_names)s") %
209 136
                 {"backend_name": matchmaker_backend,
210
-                 "available_names": _MATCHMAKER_BACKENDS})
137
+                 "available_names": zmq_options.MATCHMAKER_BACKENDS})
211 138
         return matchmaker_backend
212 139
 
213 140
     def send(self, target, ctxt, message, wait_for_reply=None, timeout=None,

+ 4
- 4
oslo_messaging/_drivers/pika_driver/pika_connection_factory.py View File

@@ -36,15 +36,15 @@ HOST_CONNECTION_LAST_TRY_TIME = "last_try_time"
36 36
 HOST_CONNECTION_LAST_SUCCESS_TRY_TIME = "last_success_try_time"
37 37
 
38 38
 pika_opts = [
39
-    cfg.IntOpt('channel_max', default=None,
39
+    cfg.IntOpt('channel_max',
40 40
                help='Maximum number of channels to allow'),
41
-    cfg.IntOpt('frame_max', default=None,
41
+    cfg.IntOpt('frame_max',
42 42
                help='The maximum byte size for an AMQP frame'),
43 43
     cfg.IntOpt('heartbeat_interval', default=3,
44 44
                help="How often to send heartbeats for consumer's connections"),
45
-    cfg.BoolOpt('ssl', default=None,
45
+    cfg.BoolOpt('ssl',
46 46
                 help='Enable SSL'),
47
-    cfg.DictOpt('ssl_options', default=None,
47
+    cfg.DictOpt('ssl_options',
48 48
                 help='Arguments passed to ssl.wrap_socket'),
49 49
     cfg.FloatOpt('socket_timeout', default=0.25,
50 50
                  help="Set socket timeout in seconds for connection's socket"),

+ 45
- 20
oslo_messaging/_drivers/pool.py View File

@@ -1,4 +1,3 @@
1
-
2 1
 # Copyright 2013 Red Hat, Inc.
3 2
 #
4 3
 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
@@ -19,6 +18,7 @@ import sys
19 18
 import threading
20 19
 
21 20
 from oslo_log import log as logging
21
+from oslo_utils import timeutils
22 22
 import six
23 23
 
24 24
 from oslo_messaging._drivers import common
@@ -38,28 +38,48 @@ else:
38 38
 
39 39
 @six.add_metaclass(abc.ABCMeta)
40 40
 class Pool(object):
41
-
42 41
     """A thread-safe object pool.
43 42
 
44 43
     Modelled after the eventlet.pools.Pool interface, but designed to be safe
45 44
     when using native threads without the GIL.
46 45
 
47 46
     Resizing is not supported.
47
+
48 48
     """
49 49
 
50
-    def __init__(self, max_size=4):
50
+    def __init__(self, max_size=4, min_size=2, ttl=1200, on_expire=None):
51 51
         super(Pool, self).__init__()
52
-
52
+        self._min_size = min_size
53 53
         self._max_size = max_size
54
+        self._item_ttl = ttl
54 55
         self._current_size = 0
55 56
         self._cond = threading.Condition()
56
-
57 57
         self._items = collections.deque()
58
+        self._on_expire = on_expire
59
+
60
+    def expire(self):
61
+        """Remove expired items from left (the oldest item) to
62
+        right (the newest item).
63
+        """
64
+        with self._cond:
65
+            while len(self._items) > self._min_size:
66
+                try:
67
+                    ttl_watch, item = self._items.popleft()
68
+                    if ttl_watch.expired():
69
+                        self._on_expire and self._on_expire(item)
70
+                        self._current_size -= 1
71
+                    else:
72
+                        self._items.appendleft((ttl_watch, item))
73
+                        return
74
+                except IndexError:
75
+                    break
58 76
 
59 77
     def put(self, item):
60 78
         """Return an item to the pool."""
61 79
         with self._cond:
62
-            self._items.appendleft(item)
80
+            ttl_watch = timeutils.StopWatch(duration=self._item_ttl)
81
+            ttl_watch.start()
82
+            self._items.append((ttl_watch, item))
63 83
             self._cond.notify()
64 84
 
65 85
     def get(self):
@@ -70,7 +90,9 @@ class Pool(object):
70 90
         with self._cond:
71 91
             while True:
72 92
                 try:
73
-                    return self._items.popleft()
93
+                    ttl_watch, item = self._items.pop()
94
+                    self.expire()
95
+                    return item
74 96
                 except IndexError:
75 97
                     pass
76 98
 
@@ -90,12 +112,12 @@ class Pool(object):
90 112
 
91 113
     def iter_free(self):
92 114
         """Iterate over free items."""
93
-        with self._cond:
94
-            while True:
95
-                try:
96
-                    yield self._items.popleft()
97
-                except IndexError:
98
-                    break
115
+        while True:
116
+            try:
117
+                _, item = self._items.pop()
118
+                yield item
119
+            except IndexError:
120
+                raise StopIteration
99 121
 
100 122
     @abc.abstractmethod
101 123
     def create(self):
@@ -104,17 +126,20 @@ class Pool(object):
104 126
 
105 127
 class ConnectionPool(Pool):
106 128
     """Class that implements a Pool of Connections."""
107
-    def __init__(self, conf, rpc_conn_pool_size, url, connection_cls):
129
+
130
+    def __init__(self, conf, max_size, min_size, ttl, url, connection_cls):
108 131
         self.connection_cls = connection_cls
109 132
         self.conf = conf
110 133
         self.url = url
111
-        super(ConnectionPool, self).__init__(rpc_conn_pool_size)
112
-        self.reply_proxy = None
134
+        super(ConnectionPool, self).__init__(max_size, min_size, ttl,
135
+                                             self._on_expire)
136
+
137
+    def _on_expire(self, connection):
138
+        connection.close()
139
+        LOG.debug("Idle connection has expired and been closed."
140
+                  " Pool size: %d" % len(self._items))
113 141
 
114
-    # TODO(comstud): Timeout connections not used in a while
115
-    def create(self, purpose=None):
116
-        if purpose is None:
117
-            purpose = common.PURPOSE_SEND
142
+    def create(self, purpose=common.PURPOSE_SEND):
118 143
         LOG.debug('Pool creating new connection')
119 144
         return self.connection_cls(self.conf, self.url, purpose)
120 145
 

+ 0
- 108
oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_call_publisher.py View File

@@ -1,108 +0,0 @@
1
-#    Copyright 2015 Mirantis, Inc.
2
-#
3
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
-#    not use this file except in compliance with the License. You may obtain
5
-#    a copy of the License at
6
-#
7
-#         http://www.apache.org/licenses/LICENSE-2.0
8
-#
9
-#    Unless required by applicable law or agreed to in writing, software
10
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
-#    License for the specific language governing permissions and limitations
13
-#    under the License.
14
-
15
-import logging
16
-
17
-from concurrent import futures
18
-import futurist
19
-
20
-import oslo_messaging
21
-from oslo_messaging._drivers import common as rpc_common
22
-from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
23
-    import zmq_reply_waiter
24
-from oslo_messaging._drivers.zmq_driver.client.publishers \
25
-    import zmq_publisher_base
26
-from oslo_messaging._drivers.zmq_driver import zmq_async
27
-from oslo_messaging._i18n import _LE
28
-
29
-LOG = logging.getLogger(__name__)
30
-
31
-zmq = zmq_async.import_zmq()
32
-
33
-
34
-class DealerCallPublisher(object):
35
-    """Thread-safe CALL publisher
36
-
37
-        Used as faster and thread-safe publisher for CALL
38
-        instead of ReqPublisher.
39
-    """
40
-
41
-    def __init__(self, conf, matchmaker, sockets_manager, sender=None,
42
-                 reply_waiter=None):
43
-        super(DealerCallPublisher, self).__init__()
44
-        self.conf = conf
45
-        self.matchmaker = matchmaker
46
-        self.reply_waiter = reply_waiter or zmq_reply_waiter.ReplyWaiter(conf)
47
-        self.sockets_manager = sockets_manager
48
-        self.sender = sender or CallSender(self.sockets_manager,
49
-                                           self.reply_waiter)
50
-
51
-    def send_request(self, request):
52
-        reply_future = self.sender.send_request(request)
53
-        try:
54
-            reply = reply_future.result(timeout=request.timeout)
55
-            LOG.debug("Received reply %s", request.message_id)
56
-        except AssertionError:
57
-            LOG.error(_LE("Message format error in reply %s"),
58
-                      request.message_id)
59
-            return None
60
-        except futures.TimeoutError:
61
-            raise oslo_messaging.MessagingTimeout(
62
-                "Timeout %(tout)s seconds was reached for message %(id)s" %
63
-                {"tout": request.timeout,
64
-                 "id": request.message_id})
65
-        finally:
66
-            self.reply_waiter.untrack_id(request.message_id)
67
-
68
-        if reply.failure:
69
-            raise rpc_common.deserialize_remote_exception(
70
-                reply.failure,
71
-                request.allowed_remote_exmods)
72
-        else:
73
-            return reply.reply_body
74
-
75
-    def cleanup(self):
76
-        self.reply_waiter.cleanup()
77
-        self.sender.cleanup()
78
-
79
-
80
-class CallSender(zmq_publisher_base.QueuedSender):
81
-
82
-    def __init__(self, sockets_manager, reply_waiter):
83
-        super(CallSender, self).__init__(sockets_manager,
84
-                                         self._do_send_request)
85
-        assert reply_waiter, "Valid ReplyWaiter expected!"
86
-        self.reply_waiter = reply_waiter
87
-
88
-    def _do_send_request(self, socket, request):
89
-        envelope = request.create_envelope()
90
-        # DEALER socket specific envelope empty delimiter
91
-        socket.send(b'', zmq.SNDMORE)
92
-        socket.send_pyobj(envelope, zmq.SNDMORE)
93
-        socket.send_pyobj(request)
94
-
95
-        LOG.debug("Sent message_id %(message)s to a target %(target)s",
96
-                  {"message": request.message_id,
97
-                   "target": request.target})
98
-
99
-    def send_request(self, request):
100
-        reply_future = futurist.Future()
101
-        self.reply_waiter.track_reply(reply_future, request.message_id)
102
-        self.queue.put(request)
103
-        return reply_future
104
-
105
-    def _connect_socket(self, target):
106
-        socket = self.outbound_sockets.get_socket(target)
107
-        self.reply_waiter.poll_socket(socket)
108
-        return socket

+ 0
- 91
oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher.py View File

@@ -1,91 +0,0 @@
1
-#    Copyright 2015 Mirantis, Inc.
2
-#
3
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
-#    not use this file except in compliance with the License. You may obtain
5
-#    a copy of the License at
6
-#
7
-#         http://www.apache.org/licenses/LICENSE-2.0
8
-#
9
-#    Unless required by applicable law or agreed to in writing, software
10
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
-#    License for the specific language governing permissions and limitations
13
-#    under the License.
14
-
15
-import logging
16
-
17
-from oslo_messaging._drivers.zmq_driver.client.publishers\
18
-    import zmq_publisher_base
19
-from oslo_messaging._drivers.zmq_driver import zmq_async
20
-from oslo_messaging._drivers.zmq_driver import zmq_names
21
-
22
-LOG = logging.getLogger(__name__)
23
-
24
-zmq = zmq_async.import_zmq()
25
-
26
-
27
-class DealerPublisher(zmq_publisher_base.QueuedSender):
28
-
29
-    def __init__(self, conf, matchmaker):
30
-
31
-        def _send_message_data(socket, request):
32
-            socket.send(b'', zmq.SNDMORE)
33
-            socket.send_pyobj(request.create_envelope(), zmq.SNDMORE)
34
-            socket.send_pyobj(request)
35
-
36
-            LOG.debug("Sent message_id %(message)s to a target %(target)s",
37
-                      {"message": request.message_id,
38
-                       "target": request.target})
39
-
40
-        def _do_send_request(socket, request):
41
-            if request.msg_type in zmq_names.MULTISEND_TYPES:
42
-                for _ in range(socket.connections_count()):
43
-                    _send_message_data(socket, request)
44
-            else:
45
-                _send_message_data(socket, request)
46
-
47
-        sockets_manager = zmq_publisher_base.SocketsManager(
48
-            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
49
-        super(DealerPublisher, self).__init__(sockets_manager,
50
-                                              _do_send_request)
51
-
52
-    def send_request(self, request):
53
-        if request.msg_type == zmq_names.CALL_TYPE:
54
-            raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
55
-        super(DealerPublisher, self).send_request(request)
56
-
57
-
58
-class DealerPublisherAsync(object):
59
-    """This simplified publisher is to be used with eventlet only.
60
-        Eventlet takes care about zmq sockets sharing between green threads
61
-        using queued lock.
62
-        Use DealerPublisher for other concurrency models.
63
-    """
64
-
65
-    def __init__(self, conf, matchmaker):
66
-        self.sockets_manager = zmq_publisher_base.SocketsManager(
67
-            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
68
-
69
-    @staticmethod
70
-    def _send_message_data(socket, request):
71
-        socket.send(b'', zmq.SNDMORE)
72
-        socket.send_pyobj(request.create_envelope(), zmq.SNDMORE)
73
-        socket.send_pyobj(request)
74
-
75
-        LOG.debug("Sent message_id %(message)s to a target %(target)s",
76
-                  {"message": request.message_id,
77
-                   "target": request.target})
78
-
79
-    def send_request(self, request):
80
-        if request.msg_type == zmq_names.CALL_TYPE:
81
-            raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
82
-        socket = self.sockets_manager.get_socket(request.target)
83
-
84
-        if request.msg_type in zmq_names.MULTISEND_TYPES:
85
-            for _ in range(socket.connections_count()):
86
-                self._send_message_data(socket, request)
87
-        else:
88
-            self._send_message_data(socket, request)
89
-
90
-    def cleanup(self):
91
-        self.sockets_manager.cleanup()

+ 106
- 0
oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_base.py View File

@@ -0,0 +1,106 @@
1
+#    Copyright 2016 Mirantis, Inc.
2
+#
3
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
+#    not use this file except in compliance with the License. You may obtain
5
+#    a copy of the License at
6
+#
7
+#         http://www.apache.org/licenses/LICENSE-2.0
8
+#
9
+#    Unless required by applicable law or agreed to in writing, software
10
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
+#    License for the specific language governing permissions and limitations
13
+#    under the License.
14
+
15
+import abc
16
+from concurrent import futures
17
+import logging
18
+
19
+import oslo_messaging
20
+from oslo_messaging._drivers import common as rpc_common
21
+from oslo_messaging._drivers.zmq_driver.client.publishers \
22
+    import zmq_publisher_base
23
+from oslo_messaging._drivers.zmq_driver.client import zmq_response
24
+from oslo_messaging._drivers.zmq_driver.client import zmq_sockets_manager
25
+from oslo_messaging._drivers.zmq_driver import zmq_async
26
+from oslo_messaging._drivers.zmq_driver import zmq_names
27
+from oslo_messaging._i18n import _LE
28
+
29
+LOG = logging.getLogger(__name__)
30
+
31
+zmq = zmq_async.import_zmq()
32
+
33
+
34
+class DealerPublisherBase(zmq_publisher_base.PublisherBase):
35
+    """Abstract DEALER-publisher."""
36
+
37
+    def __init__(self, conf, matchmaker, sender, receiver):
38
+        sockets_manager = zmq_sockets_manager.SocketsManager(
39
+            conf, matchmaker, zmq.ROUTER, zmq.DEALER
40
+        )
41
+        super(DealerPublisherBase, self).__init__(sockets_manager, sender,
42
+                                                  receiver)
43
+
44
+    @staticmethod
45
+    def _check_pattern(request, supported_pattern):
46
+        if request.msg_type != supported_pattern:
47
+            raise zmq_publisher_base.UnsupportedSendPattern(
48
+                zmq_names.message_type_str(request.msg_type)
49
+            )
50
+
51
+    @staticmethod
52
+    def _raise_timeout(request):
53
+        raise oslo_messaging.MessagingTimeout(
54
+            "Timeout %(tout)s seconds was reached for message %(msg_id)s" %
55
+            {"tout": request.timeout, "msg_id": request.message_id}
56
+        )
57
+
58
+    def _recv_reply(self, request):
59
+        reply_future = \
60
+            self.receiver.track_request(request)[zmq_names.REPLY_TYPE]
61
+
62
+        try:
63
+            _, reply = reply_future.result(timeout=request.timeout)
64
+            assert isinstance(reply, zmq_response.Reply), "Reply expected!"
65
+        except AssertionError:
66
+            LOG.error(_LE("Message format error in reply for %s"),
67
+                      request.message_id)
68
+            return None
69
+        except futures.TimeoutError:
70
+            self._raise_timeout(request)
71
+        finally:
72
+            self.receiver.untrack_request(request)
73
+
74
+        if reply.failure:
75
+            raise rpc_common.deserialize_remote_exception(
76
+                reply.failure, request.allowed_remote_exmods
77
+            )
78
+        else:
79
+            return reply.reply_body
80
+
81
+    def send_call(self, request):
82
+        self._check_pattern(request, zmq_names.CALL_TYPE)
83
+
84
+        socket = self.connect_socket(request)
85
+        if not socket:
86
+            self._raise_timeout(request)
87
+
88
+        self.sender.send(socket, request)
89
+        self.receiver.register_socket(socket)
90
+        return self._recv_reply(request)
91
+
92
+    @abc.abstractmethod
93
+    def _send_non_blocking(self, request):
94
+        pass
95
+
96
+    def send_cast(self, request):
97
+        self._check_pattern(request, zmq_names.CAST_TYPE)
98
+        self._send_non_blocking(request)
99
+
100
+    def send_fanout(self, request):
101
+        self._check_pattern(request, zmq_names.CAST_FANOUT_TYPE)
102
+        self._send_non_blocking(request)
103
+
104
+    def send_notify(self, request):
105
+        self._check_pattern(request, zmq_names.NOTIFY_TYPE)
106
+        self._send_non_blocking(request)

+ 58
- 0
oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_direct.py View File

@@ -0,0 +1,58 @@
1
+#    Copyright 2015 Mirantis, Inc.
2
+#
3
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
+#    not use this file except in compliance with the License. You may obtain
5
+#    a copy of the License at
6
+#
7
+#         http://www.apache.org/licenses/LICENSE-2.0
8
+#
9
+#    Unless required by applicable law or agreed to in writing, software
10
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
+#    License for the specific language governing permissions and limitations
13
+#    under the License.
14
+
15
+import logging
16
+
17
+import retrying
18
+
19
+from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
20
+    import zmq_dealer_publisher_base
21
+from oslo_messaging._drivers.zmq_driver.client import zmq_receivers
22
+from oslo_messaging._drivers.zmq_driver.client import zmq_senders
23
+from oslo_messaging._drivers.zmq_driver import zmq_async
24
+from oslo_messaging._drivers.zmq_driver import zmq_names
25
+
26
+LOG = logging.getLogger(__name__)
27
+
28
+zmq = zmq_async.import_zmq()
29
+
30
+
31
+class DealerPublisherDirect(zmq_dealer_publisher_base.DealerPublisherBase):
32
+    """DEALER-publisher using direct connections."""
33
+
34
+    def __init__(self, conf, matchmaker):
35
+        sender = zmq_senders.RequestSenderDirect(conf)
36
+        if conf.oslo_messaging_zmq.rpc_use_acks:
37
+            receiver = zmq_receivers.AckAndReplyReceiverDirect(conf)
38
+        else:
39
+            receiver = zmq_receivers.ReplyReceiverDirect(conf)
40
+        super(DealerPublisherDirect, self).__init__(conf, matchmaker, sender,
41
+                                                    receiver)
42
+
43
+    def connect_socket(self, request):
44
+        try:
45
+            return self.sockets_manager.get_socket(request.target)
46
+        except retrying.RetryError:
47
+            return None
48
+
49
+    def _send_non_blocking(self, request):
50
+        socket = self.connect_socket(request)
51
+        if not socket:
52
+            return
53
+
54
+        if request.msg_type in zmq_names.MULTISEND_TYPES:
55
+            for _ in range(socket.connections_count()):
56
+                self.sender.send(socket, request)
57
+        else:
58
+            self.sender.send(socket, request)

+ 52
- 147
oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_dealer_publisher_proxy.py View File

@@ -13,168 +13,73 @@
13 13
 #    under the License.
14 14
 
15 15
 import logging
16
-import six
17
-import time
16
+
17
+import retrying
18 18
 
19 19
 from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
20
-    import zmq_dealer_call_publisher
21
-from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
22
-    import zmq_reply_waiter
23
-from oslo_messaging._drivers.zmq_driver.client.publishers \
24
-    import zmq_publisher_base
20
+    import zmq_dealer_publisher_base
21
+from oslo_messaging._drivers.zmq_driver.client import zmq_receivers
22
+from oslo_messaging._drivers.zmq_driver.client import zmq_routing_table
23
+from oslo_messaging._drivers.zmq_driver.client import zmq_senders
25 24
 from oslo_messaging._drivers.zmq_driver import zmq_address
26 25
 from oslo_messaging._drivers.zmq_driver import zmq_async
27 26
 from oslo_messaging._drivers.zmq_driver import zmq_names
28 27
 from oslo_messaging._drivers.zmq_driver import zmq_updater
29 28
 
30
-zmq = zmq_async.import_zmq()
31
-
32 29
 LOG = logging.getLogger(__name__)
33 30
 
31
+zmq = zmq_async.import_zmq()
34 32
 
35
-class DealerPublisherProxy(object):
36
-    """Used when publishing to a proxy. """
37
-
38
-    def __init__(self, conf, matchmaker, socket_to_proxy):
39
-        self.conf = conf
40
-        self.sockets_manager = zmq_publisher_base.SocketsManager(
41
-            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
42
-        self.socket = socket_to_proxy
43
-        self.routing_table = RoutingTable(conf, matchmaker)
44
-        self.connection_updater = PublisherConnectionUpdater(
45
-            conf, matchmaker, self.socket)
46
-
47
-    def send_request(self, request):
48
-        if request.msg_type == zmq_names.CALL_TYPE:
49
-            raise zmq_publisher_base.UnsupportedSendPattern(
50
-                request.msg_type)
51
-
52
-        if self.conf.use_pub_sub:
53
-            routing_key = self.routing_table.get_routable_host(request.target) \
54
-                if request.msg_type in zmq_names.DIRECT_TYPES else \
55
-                zmq_address.target_to_subscribe_filter(request.target)
56
-            self._do_send_request(request, routing_key)
57
-        else:
58
-            routing_keys = self.routing_table.get_all_hosts(request.target)
59
-            for routing_key in routing_keys:
60
-                self._do_send_request(request, routing_key)
61
-
62
-    def _do_send_request(self, request, routing_key):
63
-        self.socket.send(b'', zmq.SNDMORE)
64
-        self.socket.send(six.b(str(request.msg_type)), zmq.SNDMORE)
65
-        self.socket.send(six.b(routing_key), zmq.SNDMORE)
66
-        self.socket.send(six.b(request.message_id), zmq.SNDMORE)
67
-        self.socket.send_pyobj(request.context, zmq.SNDMORE)
68
-        self.socket.send_pyobj(request.message)
69
-
70
-        LOG.debug("->[proxy:%(addr)s] Sending message_id %(message)s to "
71
-                  "a target %(target)s",
72
-                  {"message": request.message_id,
73
-                   "target": request.target,
74
-                   "addr": list(self.socket.connections)})
75
-
76
-    def cleanup(self):
77
-        self.socket.close()
78
-
79
-
80
-class DealerCallPublisherProxy(zmq_dealer_call_publisher.DealerCallPublisher):
81
-
82
-    def __init__(self, conf, matchmaker, sockets_manager):
83
-        reply_waiter = ReplyWaiterProxy(conf)
84
-        sender = CallSenderProxy(conf, matchmaker, sockets_manager,
85
-                                 reply_waiter)
86
-        super(DealerCallPublisherProxy, self).__init__(
87
-            conf, matchmaker, sockets_manager, sender, reply_waiter)
88
-
89
-
90
-class CallSenderProxy(zmq_dealer_call_publisher.CallSender):
91 33
 
92
-    def __init__(self, conf, matchmaker, sockets_manager, reply_waiter):
93
-        super(CallSenderProxy, self).__init__(
94
-            sockets_manager, reply_waiter)
95
-        self.socket = self.outbound_sockets.get_socket_to_publishers()
96
-        self.reply_waiter.poll_socket(self.socket)
97
-        self.routing_table = RoutingTable(conf, matchmaker)
98
-        self.connection_updater = PublisherConnectionUpdater(
99
-            conf, matchmaker, self.socket)
34
+class DealerPublisherProxy(zmq_dealer_publisher_base.DealerPublisherBase):
35
+    """DEALER-publisher via proxy."""
100 36
 
101
-    def _connect_socket(self, target):
37
+    def __init__(self, conf, matchmaker):
38
+        sender = zmq_senders.RequestSenderProxy(conf)
39
+        if conf.oslo_messaging_zmq.rpc_use_acks:
40
+            receiver = zmq_receivers.AckAndReplyReceiverProxy(conf)
41
+        else:
42
+            receiver = zmq_receivers.ReplyReceiverProxy(conf)
43
+        super(DealerPublisherProxy, self).__init__(conf, matchmaker, sender,
44
+                                                   receiver)
45
+        self.socket = self.sockets_manager.get_socket_to_publishers()
46
+        self.routing_table = zmq_routing_table.RoutingTable(self.conf,
47
+                                                            self.matchmaker)
48
+        self.connection_updater = \
49
+            PublisherConnectionUpdater(self.conf, self.matchmaker, self.socket)
50
+
51
+    def connect_socket(self, request):
102 52
         return self.socket
103 53
 
104
-    def _do_send_request(self, socket, request):
105
-        routing_key = self.routing_table.get_routable_host(request.target)
106
-
107
-        # DEALER socket specific envelope empty delimiter
108
-        socket.send(b'', zmq.SNDMORE)
109
-        socket.send(six.b(str(request.msg_type)), zmq.SNDMORE)
110
-        socket.send(six.b(routing_key), zmq.SNDMORE)
111
-        socket.send(six.b(request.message_id), zmq.SNDMORE)
112
-        socket.send_pyobj(request.context, zmq.SNDMORE)
113
-        socket.send_pyobj(request.message)
114
-
115
-        LOG.debug("Sent message_id %(message)s to a target %(target)s",
116
-                  {"message": request.message_id,
117
-                   "target": request.target})
118
-
119
-
120
-class ReplyWaiterProxy(zmq_reply_waiter.ReplyWaiter):
54
+    def send_call(self, request):
55
+        try:
56
+            request.routing_key = \
57
+                self.routing_table.get_routable_host(request.target)
58
+        except retrying.RetryError:
59
+            self._raise_timeout(request)
60
+        return super(DealerPublisherProxy, self).send_call(request)
61
+
62
+    def _get_routing_keys(self, request):
63
+        try:
64
+            if request.msg_type in zmq_names.DIRECT_TYPES:
65
+                return [self.routing_table.get_routable_host(request.target)]
66
+            else:
67
+                return \
68
+                    [zmq_address.target_to_subscribe_filter(request.target)] \
69
+                    if self.conf.oslo_messaging_zmq.use_pub_sub else \
70
+                    self.routing_table.get_all_hosts(request.target)
71
+        except retrying.RetryError:
72
+            return []
73
+
74
+    def _send_non_blocking(self, request):
75
+        for routing_key in self._get_routing_keys(request):
76
+            request.routing_key = routing_key
77
+            self.sender.send(self.socket, request)
121 78
 
122
-    def receive_method(self, socket):
123
-        empty = socket.recv()
124
-        assert empty == b'', "Empty expected!"
125
-        reply_id = socket.recv()
126
-        assert reply_id is not None, "Reply ID expected!"
127
-        message_type = int(socket.recv())
128
-        assert message_type == zmq_names.REPLY_TYPE, "Reply is expected!"
129
-        message_id = socket.recv()
130
-        reply = socket.recv_pyobj()
131
-        LOG.debug("Received reply %s", message_id)
132
-        return reply
133
-
134
-
135
-class RoutingTable(object):
136
-    """This class implements local routing-table cache
137
-        taken from matchmaker. Its purpose is to give the next routable
138
-        host id (remote DEALER's id) by request for specific target in
139
-        round-robin fashion.
140
-    """
141
-
142
-    def __init__(self, conf, matchmaker):
143
-        self.conf = conf
144
-        self.matchmaker = matchmaker
145
-        self.routing_table = {}
146
-        self.routable_hosts = {}
147
-
148
-    def get_all_hosts(self, target):
149
-        self._update_routing_table(target)
150
-        return list(self.routable_hosts.get(str(target)) or [])
151
-
152
-    def get_routable_host(self, target):
153
-        self._update_routing_table(target)
154
-        hosts_for_target = self.routable_hosts[str(target)]
155
-        host = hosts_for_target.pop(0)
156
-        if not hosts_for_target:
157
-            self._renew_routable_hosts(target)
158
-        return host
159
-
160
-    def _is_tm_expired(self, tm):
161
-        return 0 <= self.conf.zmq_target_expire <= time.time() - tm
162
-
163
-    def _update_routing_table(self, target):
164
-        routing_record = self.routing_table.get(str(target))
165
-        if routing_record is None:
166
-            self._fetch_hosts(target)
167
-            self._renew_routable_hosts(target)
168
-        elif self._is_tm_expired(routing_record[1]):
169
-            self._fetch_hosts(target)
170
-
171
-    def _fetch_hosts(self, target):
172
-        self.routing_table[str(target)] = (self.matchmaker.get_hosts(
173
-            target, zmq_names.socket_type_str(zmq.DEALER)), time.time())
174
-
175
-    def _renew_routable_hosts(self, target):
176
-        hosts, _ = self.routing_table[str(target)]
177
-        self.routable_hosts[str(target)] = list(hosts)
79
+    def cleanup(self):
80
+        super(DealerPublisherProxy, self).cleanup()
81
+        self.connection_updater.stop()
82
+        self.socket.close()
178 83
 
179 84
 
180 85
 class PublisherConnectionUpdater(zmq_updater.ConnectionUpdater):

+ 0
- 68
oslo_messaging/_drivers/zmq_driver/client/publishers/dealer/zmq_reply_waiter.py View File

@@ -1,68 +0,0 @@
1
-#    Copyright 2016 Mirantis, Inc.
2
-#
3
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
-#    not use this file except in compliance with the License. You may obtain
5
-#    a copy of the License at
6
-#
7
-#         http://www.apache.org/licenses/LICENSE-2.0
8
-#
9
-#    Unless required by applicable law or agreed to in writing, software
10
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
-#    License for the specific language governing permissions and limitations
13
-#    under the License.
14
-
15
-import logging
16
-import threading
17
-
18
-from oslo_messaging._drivers.zmq_driver import zmq_async
19
-from oslo_messaging._i18n import _LW
20
-
21
-LOG = logging.getLogger(__name__)
22
-
23
-zmq = zmq_async.import_zmq()
24
-
25
-
26
-class ReplyWaiter(object):
27
-
28
-    def __init__(self, conf):
29
-        self.conf = conf
30
-        self.replies = {}
31
-        self.poller = zmq_async.get_poller()
32
-        self.executor = zmq_async.get_executor(self.run_loop)
33
-        self.executor.execute()
34
-        self._lock = threading.Lock()
35
-
36
-    def track_reply(self, reply_future, message_id):
37
-        with self._lock:
38
-            self.replies[message_id] = reply_future
39
-
40
-    def untrack_id(self, message_id):
41
-        with self._lock:
42
-            self.replies.pop(message_id)
43
-
44
-    def poll_socket(self, socket):
45
-        self.poller.register(socket, recv_method=self.receive_method)
46
-
47
-    def receive_method(self, socket):
48
-        empty = socket.recv()
49
-        assert empty == b'', "Empty expected!"
50
-        envelope = socket.recv_pyobj()
51
-        assert envelope is not None, "Invalid envelope!"
52
-        reply = socket.recv_pyobj()
53
-        LOG.debug("Received reply %s", envelope)
54
-        return reply
55
-
56
-    def run_loop(self):
57
-        reply, socket = self.poller.poll(
58
-            timeout=self.conf.rpc_poll_timeout)
59
-        if reply is not None:
60
-            call_future = self.replies.get(reply.message_id)
61
-            if call_future:
62
-                call_future.set_result(reply)
63
-            else:
64
-                LOG.warning(_LW("Received timed out reply: %s"),
65
-                            reply.message_id)
66
-
67
-    def cleanup(self):
68
-        self.poller.close()

+ 29
- 133
oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_publisher_base.py View File

@@ -14,14 +14,11 @@
14 14
 
15 15
 import abc
16 16
 import logging
17
-import time
18 17
 
19 18
 import six
20 19
 
21 20
 from oslo_messaging._drivers import common as rpc_common
22 21
 from oslo_messaging._drivers.zmq_driver import zmq_async
23
-from oslo_messaging._drivers.zmq_driver import zmq_names
24
-from oslo_messaging._drivers.zmq_driver import zmq_socket
25 22
 from oslo_messaging._i18n import _LE
26 23
 
27 24
 LOG = logging.getLogger(__name__)
@@ -56,149 +53,48 @@ class PublisherBase(object):
56 53
     Publisher can send request objects from zmq_request.
57 54
     """
58 55
 
59
-    def __init__(self, sockets_manager):
56
+    def __init__(self, sockets_manager, sender, receiver):
60 57
 
61 58
         """Construct publisher
62 59
 
63
-        Accept configuration object and Name Service interface object.
64
-        Create zmq.Context and connected sockets dictionary.
60
+        Accept sockets manager, sender and receiver objects.
65 61
 
66
-        :param conf: configuration object
67
-        :type conf: oslo_config.CONF
62
+        :param sockets_manager: sockets manager object
63
+        :type sockets_manager: zmq_sockets_manager.SocketsManager
64
+        :param senders: request sender object
65
+        :type senders: zmq_senders.RequestSender
66
+        :param receiver: reply receiver object
67
+        :type receiver: zmq_receivers.ReplyReceiver
68 68
         """
69
-        self.outbound_sockets = sockets_manager
69
+        self.sockets_manager = sockets_manager
70 70
         self.conf = sockets_manager.conf
71 71
         self.matchmaker = sockets_manager.matchmaker
72
-        super(PublisherBase, self).__init__()
72
+        self.sender = sender
73
+        self.receiver = receiver
73 74
 
74 75
     @abc.abstractmethod
75
-    def send_request(self, request):
76
-        """Send request to consumer
77
-
78
-        :param request: Message data and destination container object
79
-        :type request: zmq_request.Request
76
+    def connect_socket(self, request):
77
+        """Get connected socket ready for sending given request
78
+        or None otherwise (i.e. if connection can't be established).
80 79
         """
81 80
 
82
-    def _send_request(self, socket, request):
83
-        """Send request to consumer.
84
-        Helper private method which defines basic sending behavior.
85
-
86
-        :param socket: Socket to publish message on
87
-        :type socket: zmq.Socket
88
-        :param request: Message data and destination container object
89
-        :type request: zmq_request.Request
90
-        """
91
-        LOG.debug("Sending %(type)s message_id %(message)s to a target "
92
-                  "%(target)s",
93
-                  {"type": request.msg_type,
94
-                   "message": request.message_id,
95
-                   "target": request.target})
96
-        socket.send_pyobj(request)
97
-
98
-    def cleanup(self):
99
-        """Cleanup publisher. Close allocated connections."""
100
-        self.outbound_sockets.cleanup()
101
-
102
-
103
-class SocketsManager(object):
104
-
105
-    def __init__(self, conf, matchmaker, listener_type, socket_type):
106
-        self.conf = conf
107
-        self.matchmaker = matchmaker
108
-        self.listener_type = listener_type
109
-        self.socket_type = socket_type
110
-        self.zmq_context = zmq.Context()
111
-        self.outbound_sockets = {}
112
-        self.socket_to_publishers = None
113
-        self.socket_to_routers = None
114
-
115
-    def get_hosts(self, target):
116
-        return self.matchmaker.get_hosts(
117
-            target, zmq_names.socket_type_str(self.listener_type))
118
-
119
-    @staticmethod
120
-    def _key_from_target(target):
121
-        return target.topic if target.fanout else str(target)
122
-
123
-    def _get_hosts_and_connect(self, socket, target):
124
-        hosts = self.get_hosts(target)
125
-        self._connect_to_hosts(socket, target, hosts)
126
-
127
-    def _track_socket(self, socket, target):
128
-        key = self._key_from_target(target)
129
-        self.outbound_sockets[key] = (socket, time.time())
130
-
131
-    def _connect_to_hosts(self, socket, target, hosts):
132
-        for host in hosts:
133
-            socket.connect_to_host(host)
134
-        self._track_socket(socket, target)
135
-
136
-    def _check_for_new_hosts(self, target):
137
-        key = self._key_from_target(target)
138
-        socket, tm = self.outbound_sockets[key]
139
-        if 0 <= self.conf.zmq_target_expire <= time.time() - tm:
140
-            self._get_hosts_and_connect(socket, target)
141
-        return socket
142
-
143
-    def get_socket(self, target):
144
-        key = self._key_from_target(target)
145
-        if key in self.outbound_sockets:
146
-            socket = self._check_for_new_hosts(target)
147
-        else:
148
-            socket = zmq_socket.ZmqSocket(self.conf, self.zmq_context,
149
-                                          self.socket_type)
150
-            self._get_hosts_and_connect(socket, target)
151
-        return socket
152
-
153
-    def get_socket_to_publishers(self):
154
-        if self.socket_to_publishers is not None:
155
-            return self.socket_to_publishers
156
-        self.socket_to_publishers = zmq_socket.ZmqSocket(
157
-            self.conf, self.zmq_context, self.socket_type)
158
-        publishers = self.matchmaker.get_publishers()
159
-        for pub_address, router_address in publishers:
160
-            self.socket_to_publishers.connect_to_host(router_address)
161
-        return self.socket_to_publishers
162
-
163
-    def get_socket_to_routers(self):
164
-        if self.socket_to_routers is not None:
165
-            return self.socket_to_routers
166
-        self.socket_to_routers = zmq_socket.ZmqSocket(
167
-            self.conf, self.zmq_context, self.socket_type)
168
-        routers = self.matchmaker.get_routers()
169
-        for router_address in routers:
170
-            self.socket_to_routers.connect_to_host(router_address)
171
-        return self.socket_to_routers
172
-
173
-    def cleanup(self):
174
-        for socket, tm in self.outbound_sockets.values():
175
-            socket.close()
176
-
177
-
178
-class QueuedSender(PublisherBase):
179
-
180
-    def __init__(self, sockets_manager, _do_send_request):
181
-        super(QueuedSender, self).__init__(sockets_manager)
182
-        self._do_send_request = _do_send_request
183
-        self.queue, self.empty_except = zmq_async.get_queue()
184
-        self.executor = zmq_async.get_executor(self.run_loop)
185
-        self.executor.execute()
186
-
187
-    def send_request(self, request):
188
-        self.queue.put(request)
81
+    @abc.abstractmethod
82
+    def send_call(self, request):
83
+        pass
189 84
 
190
-    def _connect_socket(self, target):
191
-        return self.outbound_sockets.get_socket(target)
85
+    @abc.abstractmethod
86
+    def send_cast(self, request):
87
+        pass
192 88
 
193
-    def run_loop(self):
194
-        try:
195
-            request = self.queue.get(timeout=self.conf.rpc_poll_timeout)
196
-        except self.empty_except:
197
-            return
89
+    @abc.abstractmethod
90
+    def send_fanout(self, request):
91
+        pass
198 92
 
199
-        socket = self._connect_socket(request.target)
200
-        self._do_send_request(socket, request)
93
+    @abc.abstractmethod
94
+    def send_notify(self, request):
95
+        pass
201 96
 
202 97
     def cleanup(self):
203
-        self.executor.stop()
204
-        super(QueuedSender, self).cleanup()
98
+        """Cleanup publisher. Close allocated connections."""
99
+        self.receiver.stop()
100
+        self.sockets_manager.cleanup()

+ 0
- 52
oslo_messaging/_drivers/zmq_driver/client/publishers/zmq_push_publisher.py View File

@@ -1,52 +0,0 @@
1
-#    Copyright 2015 Mirantis, Inc.
2
-#
3
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
-#    not use this file except in compliance with the License. You may obtain
5
-#    a copy of the License at
6
-#
7
-#         http://www.apache.org/licenses/LICENSE-2.0
8
-#
9
-#    Unless required by applicable law or agreed to in writing, software
10
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
-#    License for the specific language governing permissions and limitations
13
-#    under the License.
14
-
15
-import logging
16
-
17
-from oslo_messaging._drivers.zmq_driver.client.publishers\
18
-    import zmq_publisher_base
19
-from oslo_messaging._drivers.zmq_driver import zmq_async
20
-from oslo_messaging._drivers.zmq_driver import zmq_names
21
-
22
-LOG = logging.getLogger(__name__)
23
-
24
-zmq = zmq_async.import_zmq()
25
-
26
-
27
-class PushPublisher(object):
28
-
29
-    def __init__(self, conf, matchmaker):
30
-        super(PushPublisher, self).__init__()
31
-        sockets_manager = zmq_publisher_base.SocketsManager(
32
-            conf, matchmaker, zmq.PULL, zmq.PUSH)
33
-
34
-        def _do_send_request(push_socket, request):
35
-            push_socket.send_pyobj(request)
36
-
37
-            LOG.debug("Sending message_id %(message)s to a target %(target)s",
38
-                      {"message": request.message_id,
39
-                       "target": request.target})
40
-
41
-        self.sender = zmq_publisher_base.QueuedSender(
42
-            sockets_manager, _do_send_request)
43
-
44
-    def send_request(self, request):
45
-
46
-        if request.msg_type != zmq_names.CAST_TYPE:
47
-            raise zmq_publisher_base.UnsupportedSendPattern(request.msg_type)
48
-
49
-        self.sender.send_request(request)
50
-
51
-    def cleanup(self):
52
-        self.sender.cleanup()

+ 111
- 0
oslo_messaging/_drivers/zmq_driver/client/zmq_ack_manager.py View File

@@ -0,0 +1,111 @@
1
+#    Copyright 2016 Mirantis, Inc.
2
+#
3
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
+#    not use this file except in compliance with the License. You may obtain
5
+#    a copy of the License at
6
+#
7
+#         http://www.apache.org/licenses/LICENSE-2.0
8
+#
9
+#    Unless required by applicable law or agreed to in writing, software
10
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
+#    License for the specific language governing permissions and limitations
13
+#    under the License.
14
+
15
+from concurrent import futures
16
+import logging
17
+
18
+from oslo_messaging._drivers.zmq_driver import zmq_async
19
+from oslo_messaging._drivers.zmq_driver import zmq_names
20
+from oslo_messaging._i18n import _LE, _LW
21
+
22
+LOG = logging.getLogger(__name__)
23
+
24
+zmq = zmq_async.import_zmq()
25
+
26
+
27
+class AckManagerBase(object):
28
+
29
+    def __init__(self, publisher):
30
+        self.publisher = publisher
31
+        self.conf = publisher.conf
32
+        self.sender = publisher.sender
33
+        self.receiver = publisher.receiver
34
+
35
+    def send_call(self, request):
36
+        return self.publisher.send_call(request)
37
+
38
+    def send_cast(self, request):
39
+        self.publisher.send_cast(request)
40
+
41
+    def send_fanout(self, request):
42
+        self.publisher.send_fanout(request)
43
+
44
+    def send_notify(self, request):
45
+        self.publisher.send_notify(request)
46
+
47
+    def cleanup(self):
48
+        self.publisher.cleanup()
49
+
50
+
51
+class AckManagerDirect(AckManagerBase):
52
+    pass
53
+
54
+
55
+class AckManagerProxy(AckManagerBase):
56
+
57
+    def __init__(self, publisher):
58
+        super(AckManagerProxy, self).__init__(publisher)
59
+        self._pool = zmq_async.get_pool(
60
+            size=self.conf.oslo_messaging_zmq.rpc_thread_pool_size
61
+        )
62
+
63
+    def _wait_for_ack(self, ack_future):
64
+        request, socket = ack_future.args
65
+        retries = \
66
+            request.retry or self.conf.oslo_messaging_zmq.rpc_retry_attempts
67
+        timeout = self.conf.oslo_messaging_zmq.rpc_ack_timeout_base
68
+
69
+        done = False
70
+        while not done:
71
+            try:
72
+                reply_id, response = ack_future.result(timeout=timeout)
73
+                done = True
74
+                assert response is None, "Ack expected!"
75
+                assert reply_id == request.routing_key, \
76
+                    "Ack from recipient expected!"
77
+            except AssertionError:
78
+                LOG.error(_LE("Message format error in ack for %s"),
79
+                          request.message_id)
80
+            except futures.TimeoutError:
81
+                LOG.warning(_LW("No ack received within %(tout)s seconds "
82
+                                "for %(msg_id)s"),
83
+                            {"tout": timeout,
84
+                             "msg_id": request.message_id})
85
+                if retries is None or retries != 0:
86
+                    if retries is not None and retries > 0:
87
+                        retries -= 1
88
+                    self.sender.send(socket, request)
89
+                    timeout *= \
90
+                        self.conf.oslo_messaging_zmq.rpc_ack_timeout_multiplier
91
+                else:
92
+                    LOG.warning(_LW("Exhausted number of retries for %s"),
93
+                                request.message_id)
94
+                    done = True
95
+
96
+        self.receiver.untrack_request(request)
97
+
98
+    def _get_ack_future(self, request):
99
+        socket = self.publisher.connect_socket(request)
100
+        self.receiver.register_socket(socket)
101
+        ack_future = self.receiver.track_request(request)[zmq_names.ACK_TYPE]
102
+        ack_future.args = request, socket
103
+        return ack_future
104
+
105
+    def send_cast(self, request):
106
+        self.publisher.send_cast(request)
107
+        self._pool.submit(self._wait_for_ack, self._get_ack_future(request))
108
+
109
+    def cleanup(self):
110
+        self._pool.shutdown(wait=True)
111
+        super(AckManagerProxy, self).cleanup()

+ 32
- 43
oslo_messaging/_drivers/zmq_driver/client/zmq_client.py View File

@@ -15,13 +15,10 @@
15 15
 
16 16
 from oslo_messaging._drivers import common
17 17
 from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
18
-    import zmq_dealer_call_publisher
19
-from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
20
-    import zmq_dealer_publisher
18
+    import zmq_dealer_publisher_direct
21 19
 from oslo_messaging._drivers.zmq_driver.client.publishers.dealer \
22 20
     import zmq_dealer_publisher_proxy
23
-from oslo_messaging._drivers.zmq_driver.client.publishers \
24
-    import zmq_publisher_base
21
+from oslo_messaging._drivers.zmq_driver.client import zmq_ack_manager
25 22
 from oslo_messaging._drivers.zmq_driver.client import zmq_client_base
26 23
 from oslo_messaging._drivers.zmq_driver import zmq_async
27 24
 from oslo_messaging._drivers.zmq_driver import zmq_names
@@ -43,28 +40,28 @@ class ZmqClientMixDirectPubSub(zmq_client_base.ZmqClientBase):
43 40
 
44 41
     def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None):
45 42
 
46
-        if conf.use_router_proxy or not conf.use_pub_sub:
43
+        if conf.oslo_messaging_zmq.use_router_proxy or not \
44
+                conf.oslo_messaging_zmq.use_pub_sub:
47 45
             raise WrongClientException()
48 46
 
49
-        self.sockets_manager = zmq_publisher_base.SocketsManager(
50
-            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
47
+        publisher_direct = self.create_publisher(
48
+            conf, matchmaker,
49
+            zmq_dealer_publisher_direct.DealerPublisherDirect,
50
+            zmq_ack_manager.AckManagerDirect
51
+        )
51 52
 
52
-        fanout_publisher = zmq_dealer_publisher_proxy.DealerPublisherProxy(
53
-            conf, matchmaker, self.sockets_manager.get_socket_to_publishers())
53
+        publisher_proxy = self.create_publisher(
54
+            conf, matchmaker,
55
+            zmq_dealer_publisher_proxy.DealerPublisherProxy,
56
+            zmq_ack_manager.AckManagerProxy
57
+        )
54 58
 
55 59
         super(ZmqClientMixDirectPubSub, self).__init__(
56 60
             conf, matchmaker, allowed_remote_exmods,
57 61
             publishers={
58
-                zmq_names.CALL_TYPE:
59
-                    zmq_dealer_call_publisher.DealerCallPublisher(
60
-                        conf, matchmaker, self.sockets_manager),
61
-
62
-                zmq_names.CAST_FANOUT_TYPE: fanout_publisher,
63
-
64
-                zmq_names.NOTIFY_TYPE: fanout_publisher,
65
-
66
-                "default": zmq_dealer_publisher.DealerPublisherAsync(
67
-                    conf, matchmaker)
62
+                zmq_names.CAST_FANOUT_TYPE: publisher_proxy,
63
+                zmq_names.NOTIFY_TYPE: publisher_proxy,
64
+                "default": publisher_direct
68 65
             }
69 66
         )
70 67
 
@@ -79,22 +76,19 @@ class ZmqClientDirect(zmq_client_base.ZmqClientBase):
79 76
 
80 77
     def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None):
81 78
 
82
-        if conf.use_pub_sub or conf.use_router_proxy:
79
+        if conf.oslo_messaging_zmq.use_pub_sub or \
80
+                conf.oslo_messaging_zmq.use_router_proxy:
83 81
             raise WrongClientException()
84 82
 
85
-        self.sockets_manager = zmq_publisher_base.SocketsManager(
86
-            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
83
+        publisher = self.create_publisher(
84
+            conf, matchmaker,
85
+            zmq_dealer_publisher_direct.DealerPublisherDirect,
86
+            zmq_ack_manager.AckManagerDirect
87
+        )
87 88
 
88 89
         super(ZmqClientDirect, self).__init__(
89 90
             conf, matchmaker, allowed_remote_exmods,
90
-            publishers={
91
-                zmq_names.CALL_TYPE:
92
-                    zmq_dealer_call_publisher.DealerCallPublisher(
93
-                        conf, matchmaker, self.sockets_manager),
94
-
95
-                "default": zmq_dealer_publisher.DealerPublisher(
96
-                    conf, matchmaker)
97
-            }
91
+            publishers={"default": publisher}
98 92
         )
99 93
 
100 94
 
@@ -110,21 +104,16 @@ class ZmqClientProxy(zmq_client_base.ZmqClientBase):
110 104
 
111 105
     def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None):
112 106
 
113
-        if not conf.use_router_proxy:
107
+        if not conf.oslo_messaging_zmq.use_router_proxy:
114 108
             raise WrongClientException()
115 109
 
116
-        self.sockets_manager = zmq_publisher_base.SocketsManager(
117
-            conf, matchmaker, zmq.ROUTER, zmq.DEALER)
110
+        publisher = self.create_publisher(
111
+            conf, matchmaker,
112
+            zmq_dealer_publisher_proxy.DealerPublisherProxy,
113
+            zmq_ack_manager.AckManagerProxy
114
+        )
118 115
 
119 116
         super(ZmqClientProxy, self).__init__(
120 117
             conf, matchmaker, allowed_remote_exmods,
121
-            publishers={
122
-                zmq_names.CALL_TYPE:
123
-                    zmq_dealer_publisher_proxy.DealerCallPublisherProxy(
124
-                        conf, matchmaker, self.sockets_manager),
125
-
126
-                "default": zmq_dealer_publisher_proxy.DealerPublisherProxy(
127
-                        conf, matchmaker,
128
-                        self.sockets_manager.get_socket_to_publishers())
129
-            }
118
+            publishers={"default": publisher}
130 119
         )

+ 19
- 13
oslo_messaging/_drivers/zmq_driver/client/zmq_client_base.py View File

@@ -24,45 +24,51 @@ class ZmqClientBase(object):
24 24
     def __init__(self, conf, matchmaker=None, allowed_remote_exmods=None,
25 25
                  publishers=None):
26 26
         self.conf = conf
27
-        self.context = zmq.Context()
28 27
         self.matchmaker = matchmaker
29 28
         self.allowed_remote_exmods = allowed_remote_exmods or []
30 29
 
31 30
         self.publishers = publishers
32
-        self.call_publisher = publishers.get(zmq_names.CALL_TYPE) \
33
-            or publishers["default"]
34
-        self.cast_publisher = publishers.get(zmq_names.CAST_TYPE) \
35
-            or publishers["default"]
36
-        self.fanout_publisher = publishers.get(zmq_names.CAST_FANOUT_TYPE) \
37
-            or publishers["default"]
38
-        self.notify_publisher = publishers.get(zmq_names.NOTIFY_TYPE) \
39
-            or publishers["default"]
31
+        self.call_publisher = publishers.get(zmq_names.CALL_TYPE,
32
+                                             publishers["default"])
33
+        self.cast_publisher = publishers.get(zmq_names.CAST_TYPE,
34
+                                             publishers["default"])
35
+        self.fanout_publisher = publishers.get(zmq_names.CAST_FANOUT_TYPE,
36
+                                               publishers["default"])
37
+        self.notify_publisher = publishers.get(zmq_names.NOTIFY_TYPE,
38
+                                               publishers["default"])
39
+
40
+    @staticmethod
41
+    def create_publisher(conf, matchmaker, publisher_cls, ack_manager_cls):
42
+        publisher = publisher_cls(conf, matchmaker)
43
+        if conf.oslo_messaging_zmq.rpc_use_acks:
44
+            publisher = ack_manager_cls(publisher)
45
+        return publisher
40 46
 
41 47
     def send_call(self, target, context, message, timeout=None, retry=None):
42 48
         request = zmq_request.CallRequest(
43 49
             target, context=context, message=message, retry=retry,
44 50
             timeout=timeout, allowed_remote_exmods=self.allowed_remote_exmods
45 51
         )
46
-        return self.call_publisher.send_request(request)
52
+        return self.call_publisher.send_call(request)
47 53
 
48 54
     def send_cast(self, target, context, message, retry=None):
49 55
         request = zmq_request.CastRequest(
50 56
             target, context=context, message=message, retry=retry
51 57
         )
52
-        self.cast_publisher.send_request(request)
58
+        self.cast_publisher.send_cast(request)
53 59
 
54 60
     def send_fanout(self, target, context, message, retry=None):
55 61
         request = zmq_request.FanoutRequest(
56 62
             target, context=context, message=message, retry=retry
57 63
         )
58
-        self.fanout_publisher.send_request(request)
64
+        self.fanout_publisher.send_fanout(request)
59 65
 
60 66
     def send_notify(self, target, context, message, version, retry=None):
61 67
         request = zmq_request.NotificationRequest(
62 68
             target, context=context, message=message, retry=retry,
63 69
             version=version
64 70
         )
65
-        self.notify_publisher.send_request(request)
71
+        self.notify_publisher.send_notify(request)
66 72
 
67 73
     def cleanup(self):
68 74
         cleaned = set()

+ 0
- 89
oslo_messaging/_drivers/zmq_driver/client/zmq_envelope.py View File

@@ -1,89 +0,0 @@
1
-#    Copyright 2015 Mirantis, Inc.
2
-#
3
-#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
-#    not use this file except in compliance with the License. You may obtain
5
-#    a copy of the License at
6
-#
7
-#         http://www.apache.org/licenses/LICENSE-2.0
8
-#
9
-#    Unless required by applicable law or agreed to in writing, software
10
-#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
-#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
-#    License for the specific language governing permissions and limitations
13
-#    under the License.
14
-
15
-from oslo_messaging._drivers.zmq_driver import zmq_address
16
-from oslo_messaging._drivers.zmq_driver import zmq_names
17
-
18
-
19
-class Envelope(object):
20
-
21
-    def __init__(self, msg_type=None, message_id=None, target=None,
22
-                 routing_key=None, **kwargs):
23
-        self._msg_type = msg_type
24
-        self._message_id = message_id
25
-        self._target = target
26
-        self._reply_id = None
27
-        self._routing_key = routing_key
28
-        self._kwargs = kwargs
29
-
30
-    @property
31
-    def reply_id(self):
32
-        return self._reply_id
33
-
34
-    @reply_id.setter
35
-    def reply_id(self, value):
36
-        self._reply_id = value
37
-
38
-    @property
39
-    def routing_key(self):
40
-        return self._routing_key
41
-
42
-    @routing_key.setter
43
-    def routing_key(self, value):
44
-        self._routing_key = value
45
-
46
-    @property
47
-    def msg_type(self):
48
-        return self._msg_type
49
-
50
-    @msg_type.setter
51
-    def msg_type(self, value):
52
-        self._msg_type = value
53
-
54
-    @property
55
-    def message_id(self):
56
-        return self._message_id
57
-
58
-    @property
59
-    def target(self):
60
-        return self._target
61
-
62
-    @property
63
-    def is_mult_send(self):
64
-        return self._msg_type in zmq_names.MULTISEND_TYPES
65
-
66
-    @property
67
-    def topic_filter(self):
68
-        return zmq_address.target_to_subscribe_filter(self._target)
69
-
70
-    def has(self, key):
71
-        return key in self._kwargs
72
-
73
-    def set(self, key, value):
74
-        self._kwargs[key] = value
75
-
76
-    def get(self, key):
77
-        self._kwargs.get(key)
78
-
79
-    def to_dict(self):
80
-        envelope = {zmq_names.FIELD_MSG_TYPE: self._msg_type,
81
-                    zmq_names.FIELD_MSG_ID: self._message_id,
82
-                    zmq_names.FIELD_TARGET: self._target,
83
-                    zmq_names.FIELD_ROUTING_KEY: self._routing_key}
84
-        envelope.update({k: v for k, v in self._kwargs.items()
85
-                         if v is not None})
86
-        return envelope
87
-
88
-    def __str__(self):
89
-        return str(self.to_dict())

+ 180
- 0
oslo_messaging/_drivers/zmq_driver/client/zmq_receivers.py View File

@@ -0,0 +1,180 @@
1
+#    Copyright 2016 Mirantis, Inc.
2
+#
3
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
+#    not use this file except in compliance with the License. You may obtain
5
+#    a copy of the License at
6
+#
7
+#         http://www.apache.org/licenses/LICENSE-2.0
8
+#
9
+#    Unless required by applicable law or agreed to in writing, software
10
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
+#    License for the specific language governing permissions and limitations
13
+#    under the License.
14
+
15
+import abc
16
+import logging
17
+import threading
18
+
19
+import futurist
20
+import six
21
+
22
+from oslo_messaging._drivers.zmq_driver.client import zmq_response
23
+from oslo_messaging._drivers.zmq_driver import zmq_async
24
+from oslo_messaging._drivers.zmq_driver import zmq_names
25
+
26
+LOG = logging.getLogger(__name__)
27
+
28
+zmq = zmq_async.import_zmq()
29
+
30
+
31
+@six.add_metaclass(abc.ABCMeta)
32
+class ReceiverBase(object):
33
+    """Base response receiving interface."""
34
+
35
+    def __init__(self, conf):
36
+        self.conf = conf
37
+        self._lock = threading.Lock()
38
+        self._requests = {}
39
+        self._poller = zmq_async.get_poller()
40
+        self._executor = zmq_async.get_executor(method=self._run_loop)
41
+        self._executor.execute()
42
+
43
+    @abc.abstractproperty
44
+    def message_types(self):
45
+        """A set of supported incoming response types."""
46
+
47
+    def register_socket(self, socket):
48
+        """Register a socket for receiving data."""
49
+        self._poller.register(socket, recv_method=self.recv_response)
50
+
51
+    def unregister_socket(self, socket):
52
+        """Unregister a socket from receiving data."""
53
+        self._poller.unregister(socket)
54
+
55
+    @abc.abstractmethod
56
+    def recv_response(self, socket):
57
+        """Receive a response and return a tuple of the form
58
+        (reply_id, message_type, message_id, response).
59
+        """
60
+
61
+    def track_request(self, request):
62
+        """Track a request via already registered sockets and return
63
+        a dict of futures for monitoring all types of responses.
64
+        """
65
+        futures = {}
66
+        for message_type in self.message_types:
67
+            future = futurist.Future()
68
+            self._set_future(request.message_id, message_type, future)
69
+            futures[message_type] = future
70
+        return futures
71
+
72
+    def untrack_request(self, request):
73
+        """Untrack a request and stop monitoring any responses."""
74
+        for message_type in self.message_types:
75
+            self._pop_future(request.message_id, message_type)
76
+
77
+    def stop(self):
78
+        self._poller.close()
79
+        self._executor.stop()
80
+
81
+    def _get_future(self, message_id, message_type):
82
+        with self._lock:
83
+            return self._requests.get((message_id, message_type))
84
+
85
+    def _set_future(self, message_id, message_type, future):
86
+        with self._lock:
87
+            self._requests[(message_id, message_type)] = future
88
+
89
+    def _pop_future(self, message_id, message_type):
90
+        with self._lock:
91
+            return self._requests.pop((message_id, message_type), None)
92
+
93
+    def _run_loop(self):
94
+        data, socket = self._poller.poll(
95
+            timeout=self.conf.oslo_messaging_zmq.rpc_poll_timeout)
96
+        if data is None:
97
+            return
98
+        reply_id, message_type, message_id, response = data
99
+        assert message_type in self.message_types, \
100
+            "%s is not supported!" % zmq_names.message_type_str(message_type)
101
+        future = self._get_future(message_id, message_type)
102
+        if future is not None:
103
+            LOG.debug("Received %(msg_type)s for %(msg_id)s",
104
+                      {"msg_type": zmq_names.message_type_str(message_type),
105
+                       "msg_id": message_id})
106
+            future.set_result((reply_id, response))
107
+
108
+
109
+class ReplyReceiver(ReceiverBase):
110
+
111
+    message_types = {zmq_names.REPLY_TYPE}
112
+
113
+
114
+class ReplyReceiverProxy(ReplyReceiver):
115
+
116
+    def recv_response(self, socket):
117
+        empty = socket.recv()
118
+        assert empty == b'', "Empty expected!"
119
+        reply_id = socket.recv()
120
+        assert reply_id is not None, "Reply ID expected!"
121
+        message_type = int(socket.recv())
122
+        assert message_type == zmq_names.REPLY_TYPE, "Reply expected!"
123
+        message_id = socket.recv_string()
124
+        reply_body, failure = socket.recv_loaded()
125
+        reply = zmq_response.Reply(
126
+            message_id=message_id, reply_id=reply_id,
127
+            reply_body=reply_body, failure=failure
128
+        )
129
+        return reply_id, message_type, message_id, reply
130
+
131
+
132
+class ReplyReceiverDirect(ReplyReceiver):
133
+
134
+    def recv_response(self, socket):
135
+        empty = socket.recv()
136
+        assert empty == b'', "Empty expected!"
137
+        raw_reply = socket.recv_loaded()
138
+        assert isinstance(raw_reply, dict), "Dict expected!"
139
+        reply = zmq_response.Reply(**raw_reply)
140
+        return reply.reply_id, reply.msg_type, reply.message_id, reply
141
+
142
+
143
+class AckAndReplyReceiver(ReceiverBase):
144
+
145
+    message_types = {zmq_names.ACK_TYPE, zmq_names.REPLY_TYPE}
146
+
147
+
148
+class AckAndReplyReceiverProxy(AckAndReplyReceiver):
149
+
150
+    def recv_response(self, socket):
151
+        empty = socket.recv()
152
+        assert empty == b'', "Empty expected!"
153
+        reply_id = socket.recv()
154
+        assert reply_id is not None, "Reply ID expected!"
155
+        message_type = int(socket.recv())
156
+        assert message_type in (zmq_names.ACK_TYPE, zmq_names.REPLY_TYPE), \
157
+            "Ack or reply expected!"
158
+        message_id = socket.recv_string()
159
+        if message_type == zmq_names.REPLY_TYPE:
160
+            reply_body, failure = socket.recv_loaded()
161
+            reply = zmq_response.Reply(
162
+                message_id=message_id, reply_id=reply_id,
163
+                reply_body=reply_body, failure=failure
164
+            )
165
+            response = reply
166
+        else:
167
+            response = None
168
+        return reply_id, message_type, message_id, response
169
+
170
+
171
+class AckAndReplyReceiverDirect(AckAndReplyReceiver):
172
+
173
+    def recv_response(self, socket):
174
+        # acks are not supported yet
175
+        empty = socket.recv()
176
+        assert empty == b'', "Empty expected!"
177
+        raw_reply = socket.recv_loaded()
178
+        assert isinstance(raw_reply, dict), "Dict expected!"
179
+        reply = zmq_response.Reply(**raw_reply)
180
+        return reply.reply_id, reply.msg_type, reply.message_id, reply

+ 0
- 15
oslo_messaging/_drivers/zmq_driver/client/zmq_request.py View File

@@ -18,7 +18,6 @@ import uuid
18 18
 
19 19
 import six
20 20
 
21
-from oslo_messaging._drivers.zmq_driver.client import zmq_envelope
22 21
 from oslo_messaging._drivers.zmq_driver import zmq_async
23 22
 from oslo_messaging._drivers.zmq_driver import zmq_names
24 23
 from oslo_messaging._i18n import _LE
@@ -70,14 +69,6 @@ class Request(object):
70 69
 
71 70
         self.message_id = str(uuid.uuid1())
72 71
 
73
-    def create_envelope(self, routing_key=None, reply_id=None):
74
-        envelope = zmq_envelope.Envelope(msg_type=self.msg_type,
75
-                                         message_id=self.message_id,
76
-                                         target=self.target,
77
-                                         routing_key=routing_key)
78
-        envelope.reply_id = reply_id
79
-        return envelope
80
-
81 72
     @abc.abstractproperty
82 73
     def msg_type(self):
83 74
         """ZMQ message type"""
@@ -112,12 +103,6 @@ class CallRequest(RpcRequest):
112 103
 
113 104
         super(CallRequest, self).__init__(*args, **kwargs)
114 105
 
115
-    def create_envelope(self, routing_key=None, reply_id=None):
116
-        envelope = super(CallRequest, self).create_envelope(
117
-            routing_key, reply_id)
118
-        envelope.set('timeout', self.timeout)
119
-        return envelope
120
-
121 106
 
122 107
 class CastRequest(RpcRequest):
123 108
 

+ 36
- 22
oslo_messaging/_drivers/zmq_driver/client/zmq_response.py View File

@@ -12,28 +12,24 @@
12 12
 #    License for the specific language governing permissions and limitations
13 13
 #    under the License.
14 14
 
15
+import abc
16
+
17
+import six
18
+
15 19
 from oslo_messaging._drivers.zmq_driver import zmq_names
16 20
 
17 21
 
22
+@six.add_metaclass(abc.ABCMeta)
18 23
 class Response(object):
19 24
 
20
-    def __init__(self, id=None, type=None, message_id=None,
21
-                 reply_id=None, reply_body=None, failure=None):
25
+    def __init__(self, message_id=None, reply_id=None):
22 26
 
23
-        self._id = id
24
-        self._type = type
25 27
         self._message_id = message_id
26 28
         self._reply_id = reply_id
27
-        self._reply_body = reply_body
28
-        self._failure = failure
29
-
30
-    @property
31
-    def id_(self):
32
-        return self._id
33 29
 
34
-    @property
35
-    def type_(self):
36
-        return self._type
30
+    @abc.abstractproperty
31
+    def msg_type(self):
32
+        pass
37 33
 
38 34
     @property
39 35
     def message_id(self):
@@ -43,6 +39,29 @@ class Response(object):
43 39
     def reply_id(self):
44 40
         return self._reply_id
45 41
 
42
+    def to_dict(self):
43
+        return {zmq_names.FIELD_MSG_ID: self._message_id,
44
+                zmq_names.FIELD_REPLY_ID: self._reply_id}
45
+
46
+    def __str__(self):
47
+        return str(self.to_dict())
48
+
49
+
50
+class Ack(Response):
51
+
52
+    msg_type = zmq_names.ACK_TYPE
53
+
54
+
55
+class Reply(Response):
56
+
57
+    msg_type = zmq_names.REPLY_TYPE
58
+
59
+    def __init__(self, message_id=None, reply_id=None, reply_body=None,
60
+                 failure=None):
61
+        super(Reply, self).__init__(message_id, reply_id)
62
+        self._reply_body = reply_body
63
+        self._failure = failure
64
+
46 65
     @property
47 66
     def reply_body(self):
48 67
         return self._reply_body
@@ -52,12 +71,7 @@ class Response(object):
52 71
         return self._failure
53 72
 
54 73
     def to_dict(self):
55
-        return {zmq_names.FIELD_ID: self._id,
56
-                zmq_names.FIELD_TYPE: self._type,
57
-                zmq_names.FIELD_MSG_ID: self._message_id,
58
-                zmq_names.FIELD_REPLY_ID: self._reply_id,
59
-                zmq_names.FIELD_REPLY: self._reply_body,
60
-                zmq_names.FIELD_FAILURE: self._failure}
61
-
62
-    def __str__(self):
63
-        return str(self.to_dict())
74
+        dict_ = super(Reply, self).to_dict()
75
+        dict_.update({zmq_names.FIELD_REPLY_BODY: self._reply_body,
76
+                      zmq_names.FIELD_FAILURE: self._failure})
77
+        return dict_

+ 66
- 0
oslo_messaging/_drivers/zmq_driver/client/zmq_routing_table.py View File

@@ -0,0 +1,66 @@
1
+#    Copyright 2016 Mirantis, Inc.
2
+#
3
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
+#    not use this file except in compliance with the License. You may obtain
5
+#    a copy of the License at
6
+#
7
+#         http://www.apache.org/licenses/LICENSE-2.0
8
+#
9
+#    Unless required by applicable law or agreed to in writing, software
10
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
+#    License for the specific language governing permissions and limitations
13
+#    under the License.
14
+
15
+import time
16
+
17
+from oslo_messaging._drivers.zmq_driver import zmq_async
18
+from oslo_messaging._drivers.zmq_driver import zmq_names
19
+
20
+zmq = zmq_async.import_zmq()
21
+
22
+
23
+class RoutingTable(object):
24
+    """This class implements local routing-table cache
25
+        taken from matchmaker. Its purpose is to give the next routable
26
+        host id (remote DEALER's id) by request for specific target in
27
+        round-robin fashion.
28
+    """
29
+
30
+    def __init__(self, conf, matchmaker):
31
+        self.conf = conf
32
+        self.matchmaker = matchmaker
33
+        self.routing_table = {}
34
+        self.routable_hosts = {}
35
+
36
+    def get_all_hosts(self, target):
37
+        self._update_routing_table(target)
38
+        return list(self.routable_hosts.get(str(target), []))
39
+
40
+    def get_routable_host(self, target):
41
+        self._update_routing_table(target)
42
+        hosts_for_target = self.routable_hosts[str(target)]
43
+        host = hosts_for_target.pop()
44
+        if not hosts_for_target:
45
+            self._renew_routable_hosts(target)
46
+        return host
47
+
48
+    def _is_tm_expired(self, tm):
49
+        return 0 <= self.conf.oslo_messaging_zmq.zmq_target_expire \
50
+            <= time.time() - tm
51
+
52
+    def _update_routing_table(self, target):
53
+        routing_record = self.routing_table.get(str(target))
54
+        if routing_record is None:
55
+            self._fetch_hosts(target)
56
+            self._renew_routable_hosts(target)
57
+        elif self._is_tm_expired(routing_record[1]):
58
+            self._fetch_hosts(target)
59
+
60
+    def _fetch_hosts(self, target):
61
+        self.routing_table[str(target)] = (self.matchmaker.get_hosts(
62
+            target, zmq_names.socket_type_str(zmq.DEALER)), time.time())
63
+
64
+    def _renew_routable_hosts(self, target):
65
+        hosts, _ = self.routing_table[str(target)]
66
+        self.routable_hosts[str(target)] = list(hosts)

+ 140
- 0
oslo_messaging/_drivers/zmq_driver/client/zmq_senders.py View File

@@ -0,0 +1,140 @@
1
+#    Copyright 2016 Mirantis, Inc.
2
+#
3
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
+#    not use this file except in compliance with the License. You may obtain
5
+#    a copy of the License at
6
+#
7
+#         http://www.apache.org/licenses/LICENSE-2.0
8
+#
9
+#    Unless required by applicable law or agreed to in writing, software
10
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
+#    License for the specific language governing permissions and limitations
13
+#    under the License.
14
+
15
+import abc
16
+import logging
17
+
18
+import six
19
+
20
+from oslo_messaging._drivers.zmq_driver import zmq_async
21
+from oslo_messaging._drivers.zmq_driver import zmq_names
22
+
23
+LOG = logging.getLogger(__name__)
24
+
25
+zmq = zmq_async.import_zmq()