Merge pull request #186 from zopefoundation/fix_blob#150

fix #150

CHANGES.rst

@@ -4,6 +4,9 @@ Changelog
 5.2.4 (unreleased)
 ------------------
 
+- Fix bug related to blobs stored by ``ZEO``
+  `#150 <https://github.com/zopefoundation/ZEO/issues/150>`_.
+
 
 5.2.3 (2021-08-09)
 ------------------

src/ZEO/asyncio/base.py

@@ -77,6 +77,9 @@ class Protocol(asyncio.Protocol):
             # will be used with blobs, in which case, the individual
             # messages will be big to begin with.
             data = iter(data)
+            if paused:
+                append(data)
+                return
             for message in data:
                 writelines((pack(">I", len(message)), message))
                 if paused:

src/ZEO/asyncio/tests.py

@@ -2,8 +2,12 @@ from .._compat import PY3
 
 if PY3:
     import asyncio
+    def to_byte(i):
+        return bytes([i])
 else:
     import trollius as asyncio
+    def to_byte(b):
+        return b
 
 from zope.testing import setupstack
 from concurrent.futures import Future
@@ -18,6 +22,7 @@ import unittest
 
 from ..Exceptions import ClientDisconnected, ProtocolError
 
+from .base import Protocol
 from .testing import Loop
 from .client import ClientRunner, Fallback
 from .server import new_connection, best_protocol_version
@@ -869,10 +874,33 @@ class Logging(object):
         logging.getLogger().setLevel(logging.NOTSET)
 
 
+class ProtocolTests(setupstack.TestCase):
+
+    def setUp(self):
+        self.loop = loop = Loop()
+        loop.create_connection(lambda: Protocol(loop, None), sock=True)
+
+    def test_writeit(self):
+        """test https://github.com/zopefoundation/ZEO/issues/150."""
+        loop = self.loop
+        protocol, transport = loop.protocol, loop.transport
+        transport.capacity = 1  # single message
+        def it(tag):
+            yield tag
+            yield tag
+        protocol._writeit(it(b"0"))
+        protocol._writeit(it(b"1"))
+        for b in b"0011":
+            l, t = transport.pop(2)
+            self.assertEqual(l, b"\x00\x00\x00\x01")
+            self.assertEqual(t, to_byte(b))
+
+
 def test_suite():
     suite = unittest.TestSuite()
     suite.addTest(unittest.makeSuite(ClientTests))
     suite.addTest(unittest.makeSuite(ServerTests))
     suite.addTest(unittest.makeSuite(MsgpackClientTests))
     suite.addTest(unittest.makeSuite(MsgpackServerTests))
+    suite.addTest(unittest.makeSuite(ProtocolTests))
     return suite

src/ZEO/tests/drop_cache_rather_than_verify.txt

@@ -75,6 +75,10 @@ Now, we'll restart the server on the original address:
 
     >>> wait_connected(db.storage)
 
+##### debugging only ########
+    >>> print(db.storage._server.client.verify_result)
+    cache too old, clearing
+
 Now, let's verify our assertions above:
 
 - Publishes a stale-cache event.
@@ -141,7 +145,11 @@ another client:
 (When a database is created, it checks to make sure the root object is
 in the database, which is why we get 1, rather than 0 objects in the cache.)
 
-- Publishes a stake-cache event.
+##### debugging only ########
+    >>> print(db.storage._server.client.verify_result)
+    cache too old, clearing
+
+- Publishes a stale-cache event.
 
     >>> for e in events:
     ...     print(e)