fixed a bug when the job size was bigger than the buffer size.

added tests to check big jobs
changed buffer value

pystalkd/Beanstalkd.py

@@ -114,19 +114,21 @@ def _recv(self):
         :rtype: bytes
 
         """
-        buffer_size = 8192
+        buffer_size = 4096
         recvbuff = bytearray(buffer_size)
+        byte_list = []
         mem_view = memoryview(recvbuff)
 
         while True:
 
             n_bytes = SocketError.wrap(self._socket.recv_into, mem_view)
 
             mem_view = mem_view[0:n_bytes]
+            byte_list.append(mem_view.tobytes())
             if mem_view[-2:] == b'\r\n':
                 break
 
-        return mem_view.tobytes()
+        return b''.join(byte_list)
 
     def send(self, command, *args):
         """

setup.py

@@ -13,7 +13,7 @@
 
 setup(
     name='pystalkd',
-    version='1.2.0',
+    version='1.2.1',
     packages=['pystalkd'],
     url='https://github.com/menezes-/pystalkd',
     download_url='https://github.com/menezes-/pystalkd/archive/1.2.zip',

test.py

@@ -1,6 +1,9 @@
 from datetime import timedelta
 from pystalkd import Beanstalkd
+from os import urandom
 import json
+import random
+import string
 import unittest
 
 __author__ = 'Gabriel'
@@ -177,7 +180,6 @@ def test_bytes(self):
         """
 
         # test with random bytes
-        from os import urandom
         test_bytes = urandom(50)
         job_id = self.conn.put_bytes(test_bytes)
         self.assertIsInstance(job_id, int, "where's my job id?!")
@@ -186,6 +188,39 @@ def test_bytes(self):
         self.assertEquals(job.body, test_bytes)
         job.delete()
 
+    def test_big(self):
+        """
+        Test a job with the max allowed size for job
+        """
+        if self.conn.parse_yaml:
+            max_size = self.conn.stats()['max-job-size']
+        else:
+            # use the default max size
+            max_size = 65535  # bytes
+
+        test_str = ''.join(random.choice(string.ascii_uppercase) for _ in range(max_size))
+        job_id = self.conn.put(test_str)
+        self.assertIsInstance(job_id, int, "where's my job id?!")
+        job = self.conn.reserve(0)
+        body = job.body
+        job.delete()
+        self.assertEqual(test_str, body)
+
+    def test_big_bytes(self):
+        if self.conn.parse_yaml:
+            max_size = self.conn.stats()['max-job-size']
+        else:
+            # use the default max size
+            max_size = 65535  # bytes
+
+        test_bytes = urandom(max_size)
+        job_id = self.conn.put_bytes(test_bytes)
+        self.assertIsInstance(job_id, int, "where's my job id?!")
+        job = self.conn.reserve_bytes(0)
+        body = job.body
+        job.delete()
+        self.assertEqual(test_bytes, body)
+
     # http://stackoverflow.com/a/5387956/482238
 
     def steps(self):
@@ -227,4 +262,6 @@ def tearDown(self):
     suite.addTest(TestBeanstalkd("test_temporary_watch", host_arg, port_arg))
     suite.addTest(TestBeanstalkd("test_chinese_word", host_arg, port_arg))
     suite.addTest(TestBeanstalkd("test_bytes", host_arg, port_arg))
+    suite.addTest(TestBeanstalkd("test_big", host_arg, port_arg))
+    suite.addTest(TestBeanstalkd("test_big_bytes", host_arg, port_arg))
     unittest.TextTestRunner().run(suite)