- Fixed deprecation warning for connection.txt

 - added clean(er?) handling on cleaning up blobs on transaction boundaries and
   beeing more careful not to cause harmful states when leaking BlobFiles out
   of threads/transactions (e.g. to Medusa)

branches/ctheune-blobsupport/src/ZEO/runzeo.py

@@ -235,7 +235,7 @@ class ZEOServer:
                 storage.close()
             except: # Keep going
                 log("failed to close storage %r" % name,
-                    level=logging.EXCEPTION, exc_info=True)
+                    level=logging.exception, exc_info=True)
 
 
 # Signal names

branches/ctheune-blobsupport/src/ZODB/Blobs/Blob.py

 
 import os
+import time
 import tempfile
 
 from zope.interface import implements
@@ -7,6 +8,8 @@ from zope.interface import implements
 from ZODB.Blobs.interfaces import IBlob
 from ZODB.Blobs.exceptions import BlobError
 from ZODB import utils
+import transaction
+from transaction.interfaces import IDataManager
 from persistent import Persistent
 
 try:
@@ -25,6 +28,8 @@ class Blob(Persistent):
 
     def open(self, mode):
         """Returns a file(-like) object for handling the blob data."""
+        result = None
+
         if mode == "r":
             if self._current_filename() is None:
                 raise BlobError, "Blob does not exist."
@@ -33,7 +38,7 @@ class Blob(Persistent):
                 raise BlobError, "Already opened for writing."
 
             self._p_blob_readers += 1
-            return BlobFile(self._current_filename(), "rb", self)
+            result = BlobFile(self._current_filename(), "rb", self)
 
         if mode == "w":
             if self._p_blob_readers != 0:
@@ -43,7 +48,7 @@ class Blob(Persistent):
                 self._p_blob_uncommitted = utils.mktemp()
 
             self._p_blob_writers += 1
-            return BlobFile(self._p_blob_uncommitted, "wb", self)
+            result = BlobFile(self._p_blob_uncommitted, "wb", self)
 
         if mode =="a":
             if self._current_filename() is None:
@@ -63,13 +68,75 @@ class Blob(Persistent):
                 uncommitted = BlobFile(self._p_blob_uncommitted, "ab", self)
             
             self._p_blob_writers +=1
-            return uncommitted
+            result = uncommitted
+
+        if result is not None:
+            dm = BlobDataManager(self, result)
+            transaction.get().register(dm)
+            return result
 
     # utility methods
 
     def _current_filename(self):
         return self._p_blob_uncommitted or self._p_blob_data
 
+class BlobDataManager:
+    """Special data manager to handle transaction boundaries for blobs.
+
+    Blobs need some special care taking on transaction boundaries. As 
+    a) the ghost objects might get reused, the _p_ attributes must be
+       set to a consistent state
+    b) the file objects might get passed out of the thread/transaction
+       and must deny any relationship to the original blob.
+    """
+
+    implements(IDataManager)
+
+    def __init__(self, blob, filehandle):
+        self.blob = blob
+        self.filehandle = filehandle
+        self.isSub = False
+        self._sortkey = time.time()
+
+    def _cleanUpBlob(self):
+        self.blob._p_blob_readers = 0
+        self.blob._p_blob_writers = 0
+        self.filehandle.cleanTransaction()
+
+    def abort_sub(self, transaction):
+        pass
+
+    def commit_sub(self, transaction):
+        pass
+
+    def tpc_begin(self, transaction, subtransaction=False):
+        self.isSub = subtransaction
+
+    def tpc_abort(self, transaction):
+        self._cleanUpBlob()
+
+    def tpc_finish(self, transaction):
+        self.isSub = False
+
+    def tpc_vote(self, transaction):
+        if not self.isSub:
+            self._cleanUpBlob()
+                
+    def commit(self, object, transaction):
+        pass
+
+    def abort(self, object, transaction):
+        self._cleanUpBlob()
+
+    def sortKey(self):
+        return self._sortkey
+
+    def beforeCompletion(self, transaction):
+        pass
+
+    def afterCompletion(self, transaction):
+        pass
+
 class BlobFile(file):
 
     # XXX those files should be created in the same partition as
@@ -83,30 +150,39 @@ class BlobFile(file):
         self.blob = blob
         self.streamsize = 1<<16
 
+    def _p_changed(self):
+        if self.blob is not None:
+            self.blob._p_changed = 1
+
     def write(self, data):
         super(BlobFile, self).write(data)
-        self.blob._p_changed = 1
+        self._p_changed()
 
     def writelines(self, lines):
         super(BlobFile, self).writelines(lines)
-        self.blob._p_changed = 1
+        self._p_changed()
 
     def truncate(self, size):
         super(BlobFile, self).truncate(size)
-        self.blob._p_changed = 1
+        self._p_changed()
         
     def close(self):
-        if (self.mode.startswith("w") or
-            self.mode.startswith("a")):
-            self.blob._p_blob_writers -= 1
-        else:
-            self.blob._p_blob_readers -= 1
+        if self.blob is not None:
+            if (self.mode.startswith("w") or
+                self.mode.startswith("a")):
+                self.blob._p_blob_writers -= 1
+            else:
+                self.blob._p_blob_readers -= 1
         super(BlobFile, self).close()
 
+    def cleanTransaction(self):
+        self.blob = None
+
     def next(self):
         data = self.read(self.streamsize)
         if not data:
-            self.blob._p_blob_readers -= 1
+            if self.blob is not None:
+                self.blob._p_blob_readers -= 1
             raise StopIteration
         return data
 

branches/ctheune-blobsupport/src/ZODB/Blobs/tests/connection.txt

@@ -49,7 +49,7 @@ Getting stuff out of there works similar:
     >>> connection2 = database.open()
     >>> root = connection2.root()
     >>> blob2 = root['myblob']
-    >>> IBlob.isImplementedBy(blob2)
+    >>> IBlob.providedBy(blob2)
     True
     >>> blob2.open("r").read()
     "I'm a happy Blob."

branches/ctheune-blobsupport/src/ZODB/component.xml

@@ -164,14 +164,17 @@
     <key name="version-cache-size" datatype="integer" default="100"/>
   </sectiontype>
 
-  <sectiontype name="blobfilestorage" datatype=".BlobFileStorage"
-    implements="ZODB.storage" extends="filestorage">
+  <sectiontype name="blobstorage" datatype=".BlobStorage"
+    implements="ZODB.storage">
     <key name="blob-dir" required="yes">
       <description>
         Path name to the blob storage directory.
       </description>
     </key>
+    <section type="ZODB.storage" name="*" attribute="base"/>
   </sectiontype>
+
+
     
 
 

branches/ctheune-blobsupport/src/ZODB/config.py

@@ -132,13 +132,12 @@ class FileStorage(BaseConfig):
                            read_only=self.config.read_only,
                            quota=self.config.quota)
 
-class BlobFileStorage(FileStorage):
+class BlobStorage(BaseConfig):
 
     def open(self):
         from ZODB.Blobs.BlobStorage import BlobStorage
-        base_storage = FileStorage.open(self)
-        return BlobStorage(self.config.blob_dir, base_storage)
-
+        base = self.config.base.open()
+        return BlobStorage(self.config.blob_dir, base)
 
         
 class ZEOClient(BaseConfig):

branches/ctheune-blobsupport/src/transaction/interfaces.py

@@ -124,7 +124,7 @@ class IDataManager(zope.interface.Interface):
         transaction being committed.
         """
 
-    def commit(transaction):
+    def commit(object, transaction):
         """Commit modifications to registered objects.
 
         Save the object as part of the data to be made persistent if
@@ -134,7 +134,7 @@ class IDataManager(zope.interface.Interface):
         errors occur it saves the objects in the storage. 
         """
 
-    def abort(transaction):
+    def abort(object, transaction):
         """Abort a transaction and forget all changes.
 
         Abort must be called outside of a two-phase commit.