Merge pull request #3 from docker-hub/rename_maxsize_param_name

Roberto Gandolfo Hashioka · Roberto Gandolfo Hashioka · commit b48c71462dae · 2015-05-13T12:43:37.000-07:00
Renamed the maxsize param name to match the potential upstream version
diff --git a/kafka/producer/base.py b/kafka/producer/base.py
@@ -101,7 +101,7 @@ class Producer(object):
         batch_send: If True, messages are send in batches
         batch_send_every_n: If set, messages are send in batches of this size
         batch_send_every_t: If set, messages are send after this timeout
-        maxsize: sets the upper-bound limit on the number of items that can be placed in the queue
+        async_queue_maxsize: sets the upper-bound limit on the number of items that can be placed in the queue
     """
 
     ACK_NOT_REQUIRED = 0            # No ack is required
@@ -117,13 +117,13 @@ def __init__(self, client, async=False,
                  batch_send=False,
                  batch_send_every_n=BATCH_SEND_MSG_COUNT,
                  batch_send_every_t=BATCH_SEND_DEFAULT_INTERVAL,
-                 maxsize=None):
+                 async_queue_maxsize=None):
 
         if batch_send:
             async = True
             assert batch_send_every_n > 0
             assert batch_send_every_t > 0
-            assert maxsize >= 0
+            assert async_queue_maxsize >= 0
         else:
             batch_send_every_n = 1
             batch_send_every_t = 3600
@@ -134,8 +134,8 @@ def __init__(self, client, async=False,
         self.ack_timeout = ack_timeout
         self.stopped = False
 
-        if maxsize is None:
-            maxsize = ASYNC_QUEUE_MAXSIZE
+        if async_queue_maxsize is None:
+            async_queue_maxsize = ASYNC_QUEUE_MAXSIZE
 
         if codec is None:
             codec = CODEC_NONE
@@ -148,7 +148,7 @@ def __init__(self, client, async=False,
             log.warning("async producer does not guarantee message delivery!")
             log.warning("Current implementation does not retry Failed messages")
             log.warning("Use at your own risk! (or help improve with a PR!)")
-            self.queue = Queue(maxsize)  # Messages are sent through this queue
+            self.queue = Queue(async_queue_maxsize)  # Messages are sent through this queue
             self.thread_stop_event = Event()
             self.thread = Thread(target=_send_upstream,
                                  args=(self.queue,
diff --git a/kafka/producer/simple.py b/kafka/producer/simple.py
@@ -24,7 +24,7 @@ class SimpleProducer(Producer):
         client: The Kafka client instance to use
 
     Keyword Arguments:
-        maxsize: sets the upper-bound limit on the number of items that can be placed in the queue
+        async_queue_maxsize: sets the upper-bound limit on the number of items that can be placed in the queue
         async: If True, the messages are sent asynchronously via another
             thread (process). We will not wait for a response to these
         req_acks: A value indicating the acknowledgements that the server must
@@ -47,14 +47,14 @@ def __init__(self, client, async=False,
                  batch_send_every_n=BATCH_SEND_MSG_COUNT,
                  batch_send_every_t=BATCH_SEND_DEFAULT_INTERVAL,
                  random_start=True,
-                 maxsize=None):
+                 async_queue_maxsize=None):
         self.partition_cycles = {}
         self.random_start = random_start
         super(SimpleProducer, self).__init__(client, async, req_acks,
                                              ack_timeout, codec, batch_send,
                                              batch_send_every_n,
                                              batch_send_every_t,
-                                             maxsize)
+                                             async_queue_maxsize)
 
     def _next_partition(self, topic):
         if topic not in self.partition_cycles:
diff --git a/test/test_producer.py b/test/test_producer.py
@@ -48,7 +48,7 @@ def partitions(topic):
     @patch('kafka.producer.base._send_upstream')
     def test_producer_async_queue_overfilled_batch_send(self, mock):
         queue_size = 2
-        producer = Producer(MagicMock(), batch_send=True, maxsize=queue_size)
+        producer = Producer(MagicMock(), batch_send=True, async_queue_maxsize=queue_size)
 
         topic = b'test-topic'
         partition = 0
@@ -64,7 +64,7 @@ def test_producer_async_queue_overfilled_batch_send(self, mock):
     @patch('kafka.producer.base._send_upstream')
     def test_producer_async_queue_overfilled(self, mock):
         queue_size = 2
-        producer = Producer(MagicMock(), async=True, maxsize=queue_size)
+        producer = Producer(MagicMock(), async=True, async_queue_maxsize=queue_size)
 
         topic = b'test-topic'
         partition = 0
@@ -79,7 +79,7 @@ def test_producer_async_queue_overfilled(self, mock):
 
     def test_producer_async_queue_normal(self):
         queue_size = 4
-        producer = Producer(MagicMock(), async=True, maxsize=queue_size)
+        producer = Producer(MagicMock(), async=True, async_queue_maxsize=queue_size)
 
         topic = b'test-topic'
         partition = 0
