Merge "Small fixes in storage drivers for Python 3"

marconi/queues/storage/mongodb/messages.py

@@ -573,7 +573,7 @@ class MessageController(storage.Message):
                                       {'$set': {'tx': None}},
                                       upsert=False, multi=True)
 
-                return map(str, ids)
+                return [str(id_) for id_ in ids]
 
             except pymongo.errors.DuplicateKeyError as ex:
                 # TODO(kgriffs): Record stats of how often retries happen,

marconi/queues/storage/mongodb/utils.py

@@ -237,7 +237,7 @@ def get_partition(num_partitions, queue, project=None):
     # NOTE(kgriffs): For small numbers of partitions, crc32 will
     # provide a uniform distribution. This was verified experimentally
     # with up to 100 partitions.
-    return binascii.crc32(name) % num_partitions
+    return binascii.crc32(name.encode('utf-8')) % num_partitions
 
 
 def raises_conn_error(func):
@@ -279,6 +279,7 @@ def retries_on_autoreconnect(func):
         max_attemps = self.driver.mongodb_conf.max_reconnect_attempts
         sleep_sec = self.driver.mongodb_conf.reconnect_sleep
 
+        last_ex = None
         for attempt in range(max_attemps):
             try:
                 return func(self, *args, **kwargs)
@@ -288,12 +289,13 @@ def retries_on_autoreconnect(func):
                 LOG.warn(_(u'Caught AutoReconnect, retrying the '
                            'call to {0}').format(func))
 
+                last_ex = ex
                 time.sleep(sleep_sec * (2 ** attempt))
         else:
             LOG.error(_(u'Caught AutoReconnect, maximum attempts '
                         'to {0} exceeded.').format(func))
 
-            raise ex
+            raise last_ex
 
     return wrapper
 
@@ -317,3 +319,6 @@ class HookedCursor(object):
     def next(self):
         item = next(self.cursor)
         return self.denormalizer(item)
+
+    def __next__(self):
+        return self.next()

marconi/queues/storage/sqlalchemy/messages.py

@@ -238,7 +238,7 @@ class MessageController(storage.Message):
             statement = statement.order_by(tables.Messages.c.id.desc())
             result = trans.execute(statement).fetchall()
 
-        return map(utils.msgid_encode, [i[0] for i in reversed(result)])
+        return [utils.msgid_encode(i[0]) for i in reversed(result)]
 
     def delete(self, queue, message_id, project, claim=None):
         if project is None:

marconi/queues/storage/sqlalchemy/utils.py

@@ -88,7 +88,9 @@ def msgid_decode(id):
 
 
 def marker_encode(id):
-    return oct(id ^ 0x3c96a355)[1:]
+    # NOTE(AAzza): cannot use oct(id) here, because on Python 3 it returns
+    # string with prefix '0o', whereas on Python 2 prefix is just '0'
+    return '{0:o}'.format(id ^ 0x3c96a355)
 
 
 def marker_decode(id):