|
|
@@ -64,25 +64,25 @@ MAX_RECURSION_DEPTH = 50
|
|
|
|
|
|
|
|
|
class LifoQueue(queue.Queue):
|
|
|
- '''
|
|
|
- Variant of Queue that retrieves most recently added entries first.
|
|
|
+ '''
|
|
|
+ Variant of Queue that retrieves most recently added entries first.
|
|
|
|
|
|
- This LIFO Queue is included in python2.7 (or 2.6) and later,
|
|
|
- but it's a simple subclass, so we "backport" it here.
|
|
|
- '''
|
|
|
+ This LIFO Queue is included in python2.7 (or 2.6) and later,
|
|
|
+ but it's a simple subclass, so we "backport" it here.
|
|
|
+ '''
|
|
|
|
|
|
- def _init(self, maxsize):
|
|
|
- self.queue = []
|
|
|
- self.maxsize = maxsize
|
|
|
+ def _init(self, maxsize):
|
|
|
+ self.queue = []
|
|
|
+ self.maxsize = maxsize
|
|
|
|
|
|
- def _qsize(self, len=len):
|
|
|
- return len(self.queue)
|
|
|
+ def _qsize(self, len=len):
|
|
|
+ return len(self.queue)
|
|
|
|
|
|
- def _put(self, item):
|
|
|
- self.queue.append(item)
|
|
|
+ def _put(self, item):
|
|
|
+ self.queue.append(item)
|
|
|
|
|
|
- def _get(self):
|
|
|
- return self.queue.pop()
|
|
|
+ def _get(self):
|
|
|
+ return self.queue.pop()
|
|
|
|
|
|
|
|
|
class ConnectionConfig(object):
|
|
|
@@ -150,8 +150,9 @@ class ConnectionConfig(object):
|
|
|
self.coordinator_host = coordinator_host
|
|
|
|
|
|
def __str__(self):
|
|
|
- return ', '.join(map(str, [self.klass, self.host, self.port, self.service_name, self.use_sasl, self.kerberos_principal, self.timeout_seconds,
|
|
|
- self.mechanism, self.username, self.use_ssl, self.ca_certs, self.keyfile, self.certfile, self.validate, self.transport,
|
|
|
+ return ', '.join(map(str, [self.klass, self.host, self.port, self.service_name, self.use_sasl, self.kerberos_principal,
|
|
|
+ self.timeout_seconds, self.mechanism, self.username, self.use_ssl, self.ca_certs, self.keyfile,
|
|
|
+ self.certfile, self.validate, self.transport,
|
|
|
self.multiple, self.transport_mode, self.http_url, self.coordinator_host]))
|
|
|
|
|
|
def update_coordinator_host(self, coordinator_host):
|
|
|
@@ -259,7 +260,8 @@ class ConnectionPooler(object):
|
|
|
raise socket.timeout(
|
|
|
("Timed out after %.2f seconds waiting to retrieve a %s client from the pool.") % (has_waited_for, conf.service_name))
|
|
|
else:
|
|
|
- message = "Waited %d seconds for a Thrift client to %s:%d %s" % (has_waited_for, conf.host, conf.port, conf.get_coordinator_host())
|
|
|
+ message = "Waited %d seconds for a Thrift client to %s:%d %s" % (has_waited_for,
|
|
|
+ conf.host, conf.port, conf.get_coordinator_host())
|
|
|
log_if_slow_call(duration=has_waited_for, message=message)
|
|
|
|
|
|
return connection
|
|
|
@@ -305,7 +307,17 @@ def connect_to_thrift(conf):
|
|
|
mode.set_verify(conf.validate)
|
|
|
else:
|
|
|
if conf.use_ssl:
|
|
|
- mode = TSSLSocketWithWildcardSAN(conf.host, conf.port, validate=conf.validate, ca_certs=conf.ca_certs, keyfile=conf.keyfile, certfile=conf.certfile)
|
|
|
+ try:
|
|
|
+ from ssl import PROTOCOL_TLS
|
|
|
+ PROTOCOL_SSLv23 = PROTOCOL_TLS
|
|
|
+ except ImportError:
|
|
|
+ try:
|
|
|
+ from ssl import PROTOCOL_SSLv23 as PROTOCOL_TLS
|
|
|
+ PROTOCOL_SSLv23 = PROTOCOL_TLS
|
|
|
+ except ImportError:
|
|
|
+ PROTOCOL_SSLv23 = PROTOCOL_TLS = 2
|
|
|
+ mode = TSSLSocketWithWildcardSAN(conf.host, conf.port, validate=conf.validate, ca_certs=conf.ca_certs,
|
|
|
+ keyfile=conf.keyfile, certfile=conf.certfile, ssl_version=PROTOCOL_SSLv23)
|
|
|
else:
|
|
|
mode = TSocket(conf.host, conf.port)
|
|
|
|
|
|
@@ -623,26 +635,26 @@ def thrift2json(tft):
|
|
|
JSON protocol only supports key types that are base types.
|
|
|
I believe this ought to be true for sets, as well.
|
|
|
"""
|
|
|
- if isinstance(tft,type(None)):
|
|
|
+ if isinstance(tft, type(None)):
|
|
|
return None
|
|
|
- if isinstance(tft,(float,int,complex,basestring)):
|
|
|
+ if isinstance(tft, (float, int, complex, basestring)):
|
|
|
return tft
|
|
|
- if isinstance(tft,dict):
|
|
|
+ if isinstance(tft, dict):
|
|
|
d = {}
|
|
|
for key, val in tft.items():
|
|
|
d[key] = thrift2json(val)
|
|
|
return d
|
|
|
- if isinstance(tft,list):
|
|
|
+ if isinstance(tft, list):
|
|
|
return [thrift2json(x) for x in tft]
|
|
|
if isinstance(tft, set):
|
|
|
- return dict( (x, True) for x in tft )
|
|
|
+ return dict((x, True) for x in tft)
|
|
|
|
|
|
json = {}
|
|
|
d = {}
|
|
|
- if hasattr(tft,"__dict__"):
|
|
|
+ if hasattr(tft, "__dict__"):
|
|
|
d = tft.__dict__
|
|
|
else:
|
|
|
- if hasattr(tft,"__slots__"):
|
|
|
+ if hasattr(tft, "__slots__"):
|
|
|
d = tft.__slots__
|
|
|
else:
|
|
|
return {}
|
|
|
@@ -810,7 +822,7 @@ def fixup_enums(obj, name_class_map, suffix="AsString"):
|
|
|
"""
|
|
|
for n in list(name_class_map.keys()):
|
|
|
c = name_class_map[n]
|
|
|
- setattr(obj, n + suffix, c._VALUES_TO_NAMES[getattr(obj,n)])
|
|
|
+ setattr(obj, n + suffix, c._VALUES_TO_NAMES[getattr(obj, n)])
|
|
|
return obj
|
|
|
|
|
|
def is_thrift_struct(o):
|