/
connection.py
1547 lines (1208 loc) · 56.2 KB
/
connection.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Core connection objects"""
import ast
import collections
import logging
import math
import platform
import urllib
import urlparse
import warnings
from pika import __version__
from pika import callback
from pika import channel
from pika import credentials as pika_credentials
from pika import exceptions
from pika import frame
from pika import heartbeat
from pika import utils
from pika import spec
BACKPRESSURE_WARNING = ("Pika: Write buffer exceeded warning threshold at "
"%i bytes and an estimated %i frames behind")
PRODUCT = "Pika Python Client Library"
LOGGER = logging.getLogger(__name__)
class Parameters(object):
"""Base connection parameters class definition
:param str DEFAULT_HOST: 'localhost'
:param int DEFAULT_PORT: 5672
:param str DEFAULT_VIRTUAL_HOST: '/'
:param str DEFAULT_USERNAME: 'guest'
:param str DEFAULT_PASSWORD: 'guest'
:param int DEFAULT_HEARTBEAT_INTERVAL: 0
:param int DEFAULT_CHANNEL_MAX: 0
:param int DEFAULT_FRAME_MAX: pika.spec.FRAME_MAX_SIZE
:param str DEFAULT_LOCALE: 'en_US'
:param int DEFAULT_CONNECTION_ATTEMPTS: 1
:param int|float DEFAULT_RETRY_DELAY: 2.0
:param int|float DEFAULT_SOCKET_TIMEOUT: 0.25
:param bool DEFAULT_SSL: False
:param dict DEFAULT_SSL_OPTIONS: {}
:param int DEFAULT_SSL_PORT: 5671
:param bool DEFAULT_BACKPRESSURE_DETECTION: False
"""
DEFAULT_BACKPRESSURE_DETECTION = False
DEFAULT_CONNECTION_ATTEMPTS = 1
DEFAULT_CHANNEL_MAX = 0
DEFAULT_FRAME_MAX = spec.FRAME_MAX_SIZE
DEFAULT_HEARTBEAT_INTERVAL = 0
DEFAULT_HOST = 'localhost'
DEFAULT_LOCALE = 'en_US'
DEFAULT_PASSWORD = 'guest'
DEFAULT_PORT = 5672
DEFAULT_RETRY_DELAY = 2.0
DEFAULT_SOCKET_TIMEOUT = 0.25
DEFAULT_SSL = False
DEFAULT_SSL_OPTIONS = {}
DEFAULT_SSL_PORT = 5671
DEFAULT_USERNAME = 'guest'
DEFAULT_VIRTUAL_HOST = '/'
def __init__(self):
self.virtual_host = self.DEFAULT_VIRTUAL_HOST
self.backpressure_detection = self.DEFAULT_BACKPRESSURE_DETECTION
self.channel_max = self.DEFAULT_CHANNEL_MAX
self.connection_attempts = self.DEFAULT_CONNECTION_ATTEMPTS
self.credentials = self._credentials(self.DEFAULT_USERNAME,
self.DEFAULT_PASSWORD)
self.frame_max = self.DEFAULT_FRAME_MAX
self.heartbeat = self.DEFAULT_HEARTBEAT_INTERVAL
self.host = self.DEFAULT_HOST
self.locale = self.DEFAULT_LOCALE
self.port = self.DEFAULT_PORT
self.retry_delay = self.DEFAULT_RETRY_DELAY
self.ssl = self.DEFAULT_SSL
self.ssl_options = self.DEFAULT_SSL_OPTIONS
self.socket_timeout = self.DEFAULT_SOCKET_TIMEOUT
def __repr__(self):
"""Represent the info about the instance.
:rtype: str
"""
return ('<%s host=%s port=%s virtual_host=%s ssl=%s>' %
(self.__class__.__name__, self.host, self.port,
self.virtual_host, self.ssl))
def _credentials(self, username, password):
"""Return a plain credentials object for the specified username and
password.
:param str username: The username to use
:param str password: The password to use
:rtype: pika_credentials.PlainCredentials
"""
return pika_credentials.PlainCredentials(username, password)
def _validate_backpressure(self, backpressure_detection):
"""Validate that the backpressure detection option is a bool.
:param bool backpressure_detection: The backpressure detection value
:rtype: bool
:raises: TypeError
"""
if not isinstance(backpressure_detection, bool):
raise TypeError('backpressure detection must be a bool')
return True
def _validate_channel_max(self, channel_max):
"""Validate that the channel_max value is an int
:param int channel_max: The value to validate
:rtype: bool
:raises: TypeError
:raises: ValueError
"""
if not isinstance(channel_max, int):
raise TypeError('channel_max must be an int')
if channel_max < 1 or channel_max > 65535:
raise ValueError('channel_max must be <= 65535 and > 0')
return True
def _validate_connection_attempts(self, connection_attempts):
"""Validate that the channel_max value is an int
:param int connection_attempts: The value to validate
:rtype: bool
:raises: TypeError
:raises: ValueError
"""
if not isinstance(connection_attempts, int):
raise TypeError('connection_attempts must be an int')
if connection_attempts < 1:
raise ValueError('connection_attempts must be None or > 0')
return True
def _validate_credentials(self, credentials):
"""Validate the credentials passed in are using a valid object type.
:param pika.credentials.Credentials credentials: Credentials to validate
:rtype: bool
:raises: TypeError
"""
for credential_type in pika_credentials.VALID_TYPES:
if isinstance(credentials, credential_type):
return True
raise TypeError('Credentials must be an object of type: %r' %
pika_credentials.VALID_TYPES)
def _validate_frame_max(self, frame_max):
"""Validate that the frame_max value is an int and does not exceed
the maximum frame size and is not less than the frame min size.
:param int frame_max: The value to validate
:rtype: bool
:raises: TypeError
:raises: InvalidMinimumFrameSize
"""
if not isinstance(frame_max, int):
raise TypeError('frame_max must be an int')
if frame_max < spec.FRAME_MIN_SIZE:
raise exceptions.InvalidMinimumFrameSize
elif frame_max > spec.FRAME_MAX_SIZE:
raise exceptions.InvalidMaximumFrameSize
return True
def _validate_heartbeat_interval(self, heartbeat_interval):
"""Validate that the heartbeat_interval value is an int
:param int heartbeat_interval: The value to validate
:rtype: bool
:raises: TypeError
:raises: ValueError
"""
if not isinstance(heartbeat_interval, int):
raise TypeError('heartbeat must be an int')
if heartbeat_interval < 0:
raise ValueError('heartbeat_interval must >= 0')
return True
def _validate_host(self, host):
"""Validate that the host value is an str
:param str|unicode host: The value to validate
:rtype: bool
:raises: TypeError
"""
if not isinstance(host, basestring):
raise TypeError('host must be a str or unicode str')
return True
def _validate_locale(self, locale):
"""Validate that the locale value is an str
:param str locale: The value to validate
:rtype: bool
:raises: TypeError
"""
if not isinstance(locale, str):
raise TypeError('locale must be a str')
return True
def _validate_port(self, port):
"""Validate that the port value is an int
:param int port: The value to validate
:rtype: bool
:raises: TypeError
"""
if not isinstance(port, int):
raise TypeError('port must be an int')
return True
def _validate_retry_delay(self, retry_delay):
"""Validate that the retry_delay value is an int or float
:param int|float retry_delay: The value to validate
:rtype: bool
:raises: TypeError
"""
if not any([isinstance(retry_delay, int),
isinstance(retry_delay, float)]):
raise TypeError('retry_delay must be a float or int')
return True
def _validate_socket_timeout(self, socket_timeout):
"""Validate that the socket_timeout value is an int or float
:param int|float socket_timeout: The value to validate
:rtype: bool
:raises: TypeError
"""
if not any([isinstance(socket_timeout, int),
isinstance(socket_timeout, float)]):
raise TypeError('socket_timeout must be a float or int')
if not socket_timeout > 0:
raise ValueError('socket_timeout must be > 0')
return True
def _validate_ssl(self, ssl):
"""Validate the SSL toggle is a bool
:param bool ssl: The SSL enabled/disabled value
:rtype: bool
:raises: TypeError
"""
if not isinstance(ssl, bool):
raise TypeError('ssl must be a bool')
return True
def _validate_ssl_options(self, ssl_options):
"""Validate the SSL options value is a dictionary.
:param dict|None ssl_options: SSL Options to validate
:rtype: bool
:raises: TypeError
"""
if not isinstance(ssl_options, dict) and ssl_options is not None:
raise TypeError('ssl_options must be either None or dict')
return True
def _validate_virtual_host(self, virtual_host):
"""Validate that the virtual_host value is an str
:param str virtual_host: The value to validate
:rtype: bool
:raises: TypeError
"""
if not isinstance(virtual_host, str):
raise TypeError('virtual_host must be a str')
return True
class ConnectionParameters(Parameters):
"""Connection parameters object that is passed into the connection adapter
upon construction.
:param str host: Hostname or IP Address to connect to
:param int port: TCP port to connect to
:param str virtual_host: RabbitMQ virtual host to use
:param pika.credentials.Credentials credentials: auth credentials
:param int channel_max: Maximum number of channels to allow
:param int frame_max: The maximum byte size for an AMQP frame
:param int heartbeat_interval: How often to send heartbeats
:param bool ssl: Enable SSL
:param dict ssl_options: Arguments passed to ssl.wrap_socket as
:param int connection_attempts: Maximum number of retry attempts
:param int|float retry_delay: Time to wait in seconds, before the next
:param int|float socket_timeout: Use for high latency networks
:param str locale: Set the locale value
:param bool backpressure_detection: Toggle backpressure detection
"""
def __init__(self,
host=None,
port=None,
virtual_host=None,
credentials=None,
channel_max=None,
frame_max=None,
heartbeat_interval=None,
ssl=None,
ssl_options=None,
connection_attempts=None,
retry_delay=None,
socket_timeout=None,
locale=None,
backpressure_detection=None):
"""Create a new ConnectionParameters instance.
:param str host: Hostname or IP Address to connect to
:param int port: TCP port to connect to
:param str virtual_host: RabbitMQ virtual host to use
:param pika.credentials.Credentials credentials: auth credentials
:param int channel_max: Maximum number of channels to allow
:param int frame_max: The maximum byte size for an AMQP frame
:param int heartbeat_interval: How often to send heartbeats
:param bool ssl: Enable SSL
:param dict ssl_options: Arguments passed to ssl.wrap_socket
:param int connection_attempts: Maximum number of retry attempts
:param int|float retry_delay: Time to wait in seconds, before the next
:param int|float socket_timeout: Use for high latency networks
:param str locale: Set the locale value
:param bool backpressure_detection: Toggle backpressure detection
"""
super(ConnectionParameters, self).__init__()
# Create the default credentials object
if not credentials:
credentials = self._credentials(self.DEFAULT_USERNAME,
self.DEFAULT_PASSWORD)
# Assign the values
if host and self._validate_host(host):
self.host = host
if port is not None and self._validate_port(port):
self.port = port
if virtual_host and self._validate_host(virtual_host):
self.virtual_host = virtual_host
if credentials and self._validate_credentials(credentials):
self.credentials = credentials
if channel_max is not None and self._validate_channel_max(channel_max):
self.channel_max = channel_max
if frame_max is not None and self._validate_frame_max(frame_max):
self.frame_max = frame_max
if locale and self._validate_locale(locale):
self.locale = locale
if (heartbeat_interval is not None and
self._validate_heartbeat_interval(heartbeat_interval)):
self.heartbeat = heartbeat_interval
if ssl is not None and self._validate_ssl(ssl):
self.ssl = ssl
if ssl_options and self._validate_ssl_options(ssl_options):
self.ssl_options = ssl_options or dict()
if (connection_attempts is not None and
self._validate_connection_attempts(connection_attempts)):
self.connection_attempts = connection_attempts
if retry_delay is not None and self._validate_retry_delay(retry_delay):
self.retry_delay = retry_delay
if (socket_timeout is not None and
self._validate_socket_timeout(socket_timeout)):
self.socket_timeout = socket_timeout
if (backpressure_detection is not None and
self._validate_backpressure(backpressure_detection)):
self.backpressure_detection = backpressure_detection
class URLParameters(Parameters):
"""Connect to RabbitMQ via an AMQP URL in the format::
amqp://username:password@host:port/<virtual_host>[?query-string]
Ensure that the virtual host is URI encoded when specified. For example if
you are using the default "/" virtual host, the value should be `%2f`.
Valid query string values are:
- backpressure_detection:
Toggle backpressure detection, possible values are `t` or `f`
- channel_max:
Override the default maximum channel count value
- connection_attempts:
Specify how many times pika should try and reconnect before it gives up
- frame_max:
Override the default maximum frame size for communication
- heartbeat_interval:
Specify the number of seconds between heartbeat frames to ensure that
the link between RabbitMQ and your application is up
- locale:
Override the default `en_US` locale value
- ssl:
Toggle SSL, possible values are `t`, `f`
- ssl_options:
Arguments passed to :meth:`ssl.wrap_socket`
- retry_delay:
The number of seconds to sleep before attempting to connect on
connection failure.
- socket_timeout:
Override low level socket timeout value
:param str url: The AMQP URL to connect to
"""
def __init__(self, url):
"""Create a new URLParameters instance.
:param str url: The URL value
"""
super(URLParameters, self).__init__()
self._process_url(url)
def _process_url(self, url):
"""Take an AMQP URL and break it up into the various parameters.
:param str url: The URL to parse
"""
if url[0:4] == 'amqp':
url = 'http' + url[4:]
parts = urlparse.urlparse(url)
# Handle the Protocol scheme, changing to HTTPS so urlparse doesnt barf
if parts.scheme == 'https':
self.ssl = True
if self._validate_host(parts.hostname):
self.host = parts.hostname
if not parts.port:
if self.ssl:
self.port = self.DEFAULT_SSL_PORT if \
self.ssl else self.DEFAULT_PORT
elif self._validate_port(parts.port):
self.port = parts.port
self.credentials = pika_credentials.PlainCredentials(parts.username,
parts.password)
# Get the Virtual Host
if len(parts.path) <= 1:
self.virtual_host = self.DEFAULT_VIRTUAL_HOST
else:
path_parts = parts.path.split('/')
virtual_host = urllib.unquote(path_parts[1])
if self._validate_virtual_host(virtual_host):
self.virtual_host = virtual_host
# Handle query string values, validating and assigning them
values = urlparse.parse_qs(parts.query)
# Cast the various numeric values to the appropriate values
for key in values.keys():
# Always reassign the first list item in query values
values[key] = values[key].pop(0)
if values[key].isdigit():
values[key] = int(values[key])
else:
try:
values[key] = float(values[key])
except ValueError:
pass
if 'backpressure_detection' in values:
if values['backpressure_detection'] == 't':
self.backpressure_detection = True
elif values['backpressure_detection'] == 'f':
self.backpressure_detection = False
else:
raise ValueError('Invalid backpressure_detection value: %s' %
values['backpressure_detection'])
if ('channel_max' in values and
self._validate_channel_max(values['channel_max'])):
self.channel_max = values['channel_max']
if ('connection_attempts' in values and
self._validate_connection_attempts(values['connection_attempts'])):
self.connection_attempts = values['connection_attempts']
if ('frame_max' in values and
self._validate_frame_max(values['frame_max'])):
self.frame_max = values['frame_max']
if ('heartbeat_interval' in values and
self._validate_heartbeat_interval(values['heartbeat_interval'])):
self.heartbeat = values['heartbeat_interval']
if ('locale' in values and
self._validate_locale(values['locale'])):
self.locale = values['locale']
if ('retry_delay' in values and
self._validate_retry_delay(values['retry_delay'])):
self.retry_delay = values['retry_delay']
if ('socket_timeout' in values and
self._validate_socket_timeout(values['socket_timeout'])):
self.socket_timeout = values['socket_timeout']
if 'ssl_options' in values:
options = ast.literal_eval(values['ssl_options'])
if self._validate_ssl_options(options):
self.ssl_options = options
class Connection(object):
"""This is the core class that implements communication with RabbitMQ. This
class should not be invoked directly but rather through the use of an
adapter such as SelectConnection or BlockingConnection.
:param pika.connection.Parameters parameters: Connection parameters
:param method on_open_callback: Called when the connection is opened
:param method on_open_error_callback: Called if the connection cant
be opened
:param method on_close_callback: Called when the connection is closed
"""
ON_CONNECTION_BACKPRESSURE = '_on_connection_backpressure'
ON_CONNECTION_CLOSED = '_on_connection_closed'
ON_CONNECTION_ERROR = '_on_connection_error'
ON_CONNECTION_OPEN = '_on_connection_open'
CONNECTION_CLOSED = 0
CONNECTION_INIT = 1
CONNECTION_PROTOCOL = 2
CONNECTION_START = 3
CONNECTION_TUNE = 4
CONNECTION_OPEN = 5
CONNECTION_CLOSING = 6
def __init__(self,
parameters=None,
on_open_callback=None,
on_open_error_callback=None,
on_close_callback=None):
"""Connection initialization expects an object that has implemented the
Parameters class and a callback function to notify when we have
successfully connected to the AMQP Broker.
Available Parameters classes are the ConnectionParameters class and
URLParameters class.
:param pika.connection.Parameters parameters: Connection parameters
:param method on_open_callback: Called when the connection is opened
:param method on_open_error_callback: Called if the connection cant
be opened
:param method on_close_callback: Called when the connection is closed
"""
# Define our callback dictionary
self.callbacks = callback.CallbackManager()
# Add the on connection error callback
self.callbacks.add(0, self.ON_CONNECTION_ERROR,
on_open_error_callback or self._on_connection_error,
False)
# On connection callback
if on_open_callback:
self.add_on_open_callback(on_open_callback)
# On connection callback
if on_close_callback:
self.add_on_close_callback(on_close_callback)
# Set our configuration options
self.params = parameters or ConnectionParameters()
# Initialize the connection state and connect
self._init_connection_state()
self.connect()
def add_backpressure_callback(self, callback_method):
"""Call method "callback" when pika believes backpressure is being
applied.
:param method callback_method: The method to call
"""
self.callbacks.add(0, self.ON_CONNECTION_BACKPRESSURE,
callback_method, False)
def add_on_close_callback(self, callback_method):
"""Add a callback notification when the connection has closed. The
callback will be passed the connection, the reply_code (int) and the
reply_text (str), if sent by the remote server.
:param method callback_method: Callback to call on close
"""
self.callbacks.add(0, self.ON_CONNECTION_CLOSED, callback_method, False)
def add_on_open_callback(self, callback_method):
"""Add a callback notification when the connection has opened.
:param method callback_method: Callback to call when open
"""
self.callbacks.add(0, self.ON_CONNECTION_OPEN, callback_method, False)
def add_on_open_error_callback(self, callback_method, remove_default=True):
"""Add a callback notification when the connection can not be opened.
The callback method should accept the connection object that could not
connect, and an optional error message.
:param method callback_method: Callback to call when can't connect
:param bool remove_default: Remove default exception raising callback
"""
if remove_default:
self.callbacks.remove(0, self.ON_CONNECTION_ERROR,
self._on_connection_error)
self.callbacks.add(0, self.ON_CONNECTION_ERROR, callback_method, False)
def add_timeout(self, deadline, callback_method):
"""Adapters should override to call the callback after the
specified number of seconds have elapsed, using a timer, or a
thread, or similar.
:param int deadline: The number of seconds to wait to call callback
:param method callback_method: The callback method
"""
raise NotImplementedError
def channel(self, on_open_callback, channel_number=None, force_binary=False):
"""Create a new channel with the next available channel number or pass
in a channel number to use. Must be non-zero if you would like to
specify but it is recommended that you let Pika manage the channel
numbers.
:param method on_open_callback: The callback when the channel is opened
:param int channel_number: The channel number to use, defaults to the
next available.
:param bool force_binary: Prevents channel from autodetecting unicode
:rtype: pika.channel.Channel
"""
if not channel_number:
channel_number = self._next_channel_number()
self._channels[channel_number] = self._create_channel(channel_number,
on_open_callback,
force_binary)
self._add_channel_callbacks(channel_number)
self._channels[channel_number].open()
return self._channels[channel_number]
def close(self, reply_code=200, reply_text='Normal shutdown'):
"""Disconnect from RabbitMQ. If there are any open channels, it will
attempt to close them prior to fully disconnecting. Channels which
have active consumers will attempt to send a Basic.Cancel to RabbitMQ
to cleanly stop the delivery of messages prior to closing the channel.
:param int reply_code: The code number for the close
:param str reply_text: The text reason for the close
"""
if self.is_closing or self.is_closed:
return
if self._has_open_channels:
self._close_channels(reply_code, reply_text)
# Set our connection state
self._set_connection_state(self.CONNECTION_CLOSING)
LOGGER.info("Closing connection (%s): %s", reply_code, reply_text)
self.closing = reply_code, reply_text
if not self._has_open_channels:
# if there are open channels then _on_close_ready will finally be
# called in _on_channel_closeok once all channels have been closed
self._on_close_ready()
def connect(self):
"""Invoke if trying to reconnect to a RabbitMQ server. Constructing the
Connection object should connect on its own.
"""
self._set_connection_state(self.CONNECTION_INIT)
error = self._adapter_connect()
if not error:
return self._on_connected()
self.remaining_connection_attempts -= 1
LOGGER.warning('Could not connect, %i attempts left',
self.remaining_connection_attempts)
if self.remaining_connection_attempts:
LOGGER.info('Retrying in %i seconds', self.params.retry_delay)
self.add_timeout(self.params.retry_delay, self.connect)
else:
self.callbacks.process(0, self.ON_CONNECTION_ERROR, self, self, error)
self.remaining_connection_attempts = self.params.connection_attempts
self._set_connection_state(self.CONNECTION_CLOSED)
def remove_timeout(self, callback_method):
"""Adapters should override to call the callback after the
specified number of seconds have elapsed, using a timer, or a
thread, or similar.
:param method callback_method: The callback to remove a timeout for
"""
raise NotImplementedError
def set_backpressure_multiplier(self, value=10):
"""Alter the backpressure multiplier value. We set this to 10 by default.
This value is used to raise warnings and trigger the backpressure
callback.
:param int value: The multiplier value to set
"""
self._backpressure = value
#
# Connections state properties
#
@property
def is_closed(self):
"""
Returns a boolean reporting the current connection state.
"""
return self.connection_state == self.CONNECTION_CLOSED
@property
def is_closing(self):
"""
Returns a boolean reporting the current connection state.
"""
return self.connection_state == self.CONNECTION_CLOSING
@property
def is_open(self):
"""
Returns a boolean reporting the current connection state.
"""
return self.connection_state == self.CONNECTION_OPEN
#
# Properties that reflect server capabilities for the current connection
#
@property
def basic_nack(self):
"""Specifies if the server supports basic.nack on the active connection.
:rtype: bool
"""
return self.server_capabilities.get('basic.nack', False)
@property
def consumer_cancel_notify(self):
"""Specifies if the server supports consumer cancel notification on the
active connection.
:rtype: bool
"""
return self.server_capabilities.get('consumer_cancel_notify', False)
@property
def exchange_exchange_bindings(self):
"""Specifies if the active connection supports exchange to exchange
bindings.
:rtype: bool
"""
return self.server_capabilities.get('exchange_exchange_bindings',
False)
@property
def publisher_confirms(self):
"""Specifies if the active connection can use publisher confirmations.
:rtype: bool
"""
return self.server_capabilities.get('publisher_confirms', False)
#
# Internal methods for managing the communication process
#
def _adapter_connect(self):
"""Subclasses should override to set up the outbound socket connection.
:raises: NotImplementedError
"""
raise NotImplementedError
def _adapter_disconnect(self):
"""Subclasses should override this to cause the underlying transport
(socket) to close.
:raises: NotImplementedError
"""
raise NotImplementedError
def _add_channel_callbacks(self, channel_number):
"""Add the appropriate callbacks for the specified channel number.
:param int channel_number: The channel number for the callbacks
"""
self.callbacks.add(channel_number,
spec.Channel.CloseOk,
self._on_channel_closeok)
def _add_connection_start_callback(self):
"""Add a callback for when a Connection.Start frame is received from
the broker.
"""
self.callbacks.add(0, spec.Connection.Start, self._on_connection_start)
def _add_connection_tune_callback(self):
"""Add a callback for when a Connection.Tune frame is received."""
self.callbacks.add(0, spec.Connection.Tune, self._on_connection_tune)
def _append_frame_buffer(self, value):
"""Append the bytes to the frame buffer.
:param str value: The bytes to append to the frame buffer
"""
self._frame_buffer += value
@property
def _buffer_size(self):
"""Return the suggested buffer size from the connection state/tune or
the default if that is None.
:rtype: int
"""
return self.params.frame_max or spec.FRAME_MAX_SIZE
def _check_for_protocol_mismatch(self, value):
"""Invoked when starting a connection to make sure it's a supported
protocol.
:param pika.frame.Method value: The frame to check
:raises: ProtocolVersionMismatch
"""
if (value.method.version_major,
value.method.version_minor) != spec.PROTOCOL_VERSION[0:2]:
raise exceptions.ProtocolVersionMismatch(frame.ProtocolHeader(),
value)
@property
def _client_properties(self):
"""Return the client properties dictionary.
:rtype: dict
"""
return {'product': PRODUCT,
'platform': 'Python %s' % platform.python_version(),
'capabilities': {'authentication_failure_close': True,
'basic.nack': True,
'connection.blocked': True,
'consumer_cancel_notify': True,
'publisher_confirms': True},
'information': 'See http://pika.rtfd.org',
'version': __version__}
def _close_channels(self, reply_code, reply_text):
"""Close the open channels with the specified reply_code and reply_text.
:param int reply_code: The code for why the channels are being closed
:param str reply_text: The text reason for why the channels are closing
"""
if self.is_open:
for channel_number in self._channels.keys():
if self._channels[channel_number].is_open:
self._channels[channel_number].close(reply_code, reply_text)
else:
del self._channels[channel_number]
# Force any lingering callbacks to be removed
# moved inside else block since _on_channel_closeok removes
# callbacks
self.callbacks.cleanup(channel_number)
else:
self._channels = dict()
def _combine(self, a, b):
"""Pass in two values, if a is 0, return b otherwise if b is 0,
return a. If neither case matches return the smallest value.
:param int a: The first value
:param int b: The second value
:rtype: int
"""
return min(a, b) or (a or b)
def _connect(self):
"""Attempt to connect to RabbitMQ
:rtype: bool
"""
warnings.warn('This method is deprecated, use Connection.connect',
DeprecationWarning)
def _create_channel(self, channel_number, on_open_callback, force_binary):
"""Create a new channel using the specified channel number and calling
back the method specified by on_open_callback
:param int channel_number: The channel number to use
:param method on_open_callback: The callback when the channel is opened
:param bool force_binary: Prevents channel from autodetecting unicode
"""
return channel.Channel(self, channel_number, on_open_callback, force_binary)
def _create_heartbeat_checker(self):
"""Create a heartbeat checker instance if there is a heartbeat interval
set.
:rtype: pika.heartbeat.Heartbeat
"""
if self.params.heartbeat is not None and self.params.heartbeat > 0:
LOGGER.debug('Creating a HeartbeatChecker: %r',
self.params.heartbeat)
return heartbeat.HeartbeatChecker(self, self.params.heartbeat)
def _deliver_frame_to_channel(self, value):
"""Deliver the frame to the channel specified in the frame.
:param pika.frame.Method value: The frame to deliver
"""
if not value.channel_number in self._channels:
if self._is_basic_deliver_frame(value):
self._reject_out_of_band_delivery(value.channel_number,
value.method.delivery_tag)
else:
LOGGER.warning("Received %r for non-existing channel %i",
value, value.channel_number)
return
return self._channels[value.channel_number]._handle_content_frame(value)
def _detect_backpressure(self):
"""Attempt to calculate if TCP backpressure is being applied due to
our outbound buffer being larger than the average frame size over
a window of frames.
"""
avg_frame_size = self.bytes_sent / self.frames_sent
buffer_size = sum([len(frame) for frame in self.outbound_buffer])
if buffer_size > (avg_frame_size * self._backpressure):
LOGGER.warning(BACKPRESSURE_WARNING, buffer_size,
int(buffer_size / avg_frame_size))
self.callbacks.process(0, self.ON_CONNECTION_BACKPRESSURE, self)
def _ensure_closed(self):
"""If the connection is not closed, close it."""
if self.is_open:
self.close()
def _flush_outbound(self):
"""Adapters should override to flush the contents of outbound_buffer
out along the socket.
:raises: NotImplementedError
"""
raise NotImplementedError