[Zodb-checkins] SVN: ZODB/trunk/src/ZEO/zrpc/connection.py Added support for message iterators. This allows one, for example, to

[Jim Fulton]

Log message for revision 75839:
  Added support for message iterators.  This allows one, for example, to
  use an iterator to send a large file without loading it in memory.
  
  Updated the ZEO protocol to reflect the new Blob-support methods.
  

Changed:
  U   ZODB/trunk/src/ZEO/zrpc/connection.py

-=-
Modified: ZODB/trunk/src/ZEO/zrpc/connection.py
===================================================================
--- ZODB/trunk/src/ZEO/zrpc/connection.py	2007-05-18 18:01:59 UTC (rev 75838)
+++ ZODB/trunk/src/ZEO/zrpc/connection.py	2007-05-18 18:02:05 UTC (rev 75839)
@@ -295,17 +295,31 @@
     # Z303 -- named after the ZODB release 3.3
     #         Added methods for MVCC:
     #             loadBefore()
-    #             loadEx()
     #         A Z303 client cannot talk to a Z201 server, because the latter
     #         doesn't support MVCC.  A Z201 client can talk to a Z303 server,
     #         but because (at least) the type of the root object changed
     #         from ZODB.PersistentMapping to persistent.mapping, the older
     #         client can't actually make progress if a Z303 client created,
     #         or ever modified, the root.
+    #
+    # Z308 -- named after the ZODB release 3.8
+    #         Added blob-support server methods:
+    #             sendBlob
+    #             storeEmptyBlob
+    #             storeBlobStart
+    #             storeBlobChunk
+    #             storeBlobEnd
+    #             storeBlobShared
+    #         Added blob-support client methods:
+    #             recieveBlobStart
+    #             recieveBlobChunk
+    #             recieveBlobStop
+    
+    # XXX add blob methods
 
     # Protocol variables:
     # Our preferred protocol.
-    current_protocol = "Z303"
+    current_protocol = "Z308"
 
     # If we're a client, an exhaustive list of the server protocols we
     # can accept.
@@ -313,7 +327,7 @@
 
     # If we're a server, an exhaustive list of the client protocols we
     # can accept.
-    clients_we_can_talk_to = ["Z200", "Z201", current_protocol]
+    clients_we_can_talk_to = ["Z200", "Z201", "Z303", current_protocol]
 
     # This is pretty excruciating.  Details:
     #
@@ -619,17 +633,29 @@
     # The next two public methods (call and callAsync) are used by
     # clients to invoke methods on remote objects
 
-    def send_call(self, method, args, flags):
-        # send a message and return its msgid
+    def __new_msgid(self):
         self.msgid_lock.acquire()
         try:
             msgid = self.msgid
             self.msgid = self.msgid + 1
+            return msgid
         finally:
             self.msgid_lock.release()
+
+    def __call_message(self, method, args, flags):
+        # compute a message and return it
+        msgid = self.__new_msgid()
         if __debug__:
             self.log("send msg: %d, %d, %s, ..." % (msgid, flags, method),
                      level=TRACE)
+        return self.marshal.encode(msgid, flags, method, args)
+
+    def send_call(self, method, args, flags):
+        # send a message and return its msgid
+        msgid = self.__new_msgid()
+        if __debug__:
+            self.log("send msg: %d, %d, %s, ..." % (msgid, flags, method),
+                     level=TRACE)
         buf = self.marshal.encode(msgid, flags, method, args)
         self.message_output(buf)
         return msgid
@@ -684,6 +710,18 @@
             raise DisconnectedError()
         self.send_call(method, args, ASYNC)
 
+    def callAsyncIterator(self, iterator):
+        """Queue a sequence of calls using an iterator
+
+        The calls will not be interleaved with other calls from the same
+        client.
+        """
+        self.message_output(self.__outputIterator(iterator))
+
+    def __outputIterator(self, iterator):
+        for method, args in iterator:
+            yield self.__call_message(method, args, ASYNC)
+
     # handle IO, possibly in async mode
 
     def _prepare_async(self):