Am/paste async button (#614)

* Add 'paste async' button to manage content

* Changes

* Add upgrade profile and sample blob solution

* Fix broken blobs in documents

* Add loadBlob monkeypatch for async paste operations

* Add original loadblob behavior if env variable not enabled

castle/cms/blobmissing.py

@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+from pkg_resources import resource_filename
+from shutil import copyfile
+from ZEO import ClientStorage
+from ZODB.blob import BlobFile
+from ZODB.POSException import POSKeyError
+from ZODB.POSException import Unsupported
+
+import logging
+import os
+
+
+logger = logging.getLogger(__name__)
+
+
+def patched_blob_init(self, name, mode, blob):
+    allow_experimental = os.getenv("CASTLE_ALLOW_EXPERIMENTAL_BLOB_REPLACEMENT", None)
+    if not os.path.exists(name) and allow_experimental is not None:
+        create_empty_blob(name)
+    super(BlobFile, self).__init__(name, mode + 'b')
+    self.blob = blob
+
+
+def patched_loadBlob(self, oid, serial):
+    # Load a blob.  If it isn't present and we have a shared blob
+    # directory, then assume that it doesn't exist on the server
+    # and return None.
+
+    blob_filename = self.fshelper.getBlobFilename(oid, serial)
+    allow_experimental = os.getenv("CASTLE_ALLOW_EXPERIMENTAL_BLOB_REPLACEMENT", None)
+
+    if allow_experimental is None:
+        # Exhibit original behavior if blob replacement is not enabled
+        if not os.path.exists(blob_filename):
+            logger.error("No blob file at path: %s", blob_filename)
+            raise POSKeyError("No blob file", oid, serial)
+        return blob_filename
+
+    if self.shared_blob_dir:
+        if os.path.exists(blob_filename):
+            return blob_filename
+        else:
+            # create empty file
+            create_empty_blob(blob_filename)
+        if os.path.exists(blob_filename):
+            return blob_filename
+        else:
+            # We're using a server shared cache.  If the file isn't
+            # here, it's not anywhere.
+            raise POSKeyError("No blob file", oid, serial)
+
+    if os.path.exists(blob_filename):
+        return ClientStorage._accessed(blob_filename)
+    else:
+        # create empty file
+        create_empty_blob(blob_filename)
+
+    if os.path.exists(blob_filename):
+        return ClientStorage._accessed(blob_filename)
+
+    # First, we'll create the directory for this oid, if it doesn't exist.
+    self.fshelper.createPathForOID(oid)
+
+    # OK, it's not here and we (or someone) needs to get it.  We
+    # want to avoid getting it multiple times.  We want to avoid
+    # getting it multiple times even accross separate client
+    # processes on the same machine. We'll use file locking.
+
+    lock = ClientStorage._lock_blob(blob_filename)
+    try:
+        # We got the lock, so it's our job to download it.  First,
+        # we'll double check that someone didn't download it while we
+        # were getting the lock:
+
+        if os.path.exists(blob_filename):
+            return ClientStorage._accessed(blob_filename)
+
+        # Ask the server to send it to us.  When this function
+        # returns, it will have been sent. (The recieving will
+        # have been handled by the asyncore thread.)
+
+        self._server.sendBlob(oid, serial)
+
+        if os.path.exists(blob_filename):
+            return ClientStorage._accessed(blob_filename)
+
+        raise POSKeyError("No blob file", oid, serial)
+
+    finally:
+        lock.close()
+
+
+def create_empty_blob(filename):
+    dirname = os.path.split(filename)[0]
+    if not os.path.isdir(dirname):
+        os.makedirs(dirname, 0o700)
+
+    logger.info("Broken blob reference detected for: %s", filename)
+
+    source = resource_filename(
+        'castle.cms',
+        'static/images/placeholder.png',
+    )
+
+    copyfile(source, filename)
+    logger.info("Placeholder blob created for: %s", filename)

castle/cms/blobmissing.zcml

@@ -0,0 +1,29 @@
+<configure
+    xmlns="http://namespaces.zope.org/zope"
+    xmlns:five="http://namespaces.zope.org/five"
+    xmlns:i18n="http://namespaces.zope.org/i18n"
+    xmlns:plone="http://namespaces.plone.org/plone"
+    xmlns:monkey="http://namespaces.plone.org/monkey"
+    i18n_domain="castle.cms">
+
+    <include package="collective.monkeypatcher" />
+    <include package="collective.monkeypatcher" file="meta.zcml" />
+
+    <monkey:patch
+        description="Create the blob folder path and create (touch) an empty file for each blob file if it's missing."
+        class="ZODB.blob.BlobFile"
+        original="__init__"
+        replacement=".blobmissing.patched_blob_init"
+        docstringWarning="true"
+        />
+
+    <monkey:patch
+        description="Create the blob-file if it's missing."
+        class="ZEO.ClientStorage.ClientStorage"
+        original="loadBlob"
+        replacement=".blobmissing.patched_loadBlob"
+        docstringWarning="true"
+        />
+
+
+</configure>

castle/cms/browser/content/fc.py

@@ -33,14 +33,6 @@
 from zope.interface import implementer
 from zope.lifecycleevent import ObjectModifiedEvent
 from zope.component.hooks import getSite
-from plone.app.contenttypes.interfaces import IDocument
-from zope.annotation.interfaces import IAnnotations
-from persistent.mapping import PersistentMapping
-from plone.namedfile.file import NamedBlobImage
-from ZODB.blob import FilesystemHelper
-from pkg_resources import resource_filename
-from shutil import copyfile
-import os
 from re import sub
 
 from plone.app.content.browser.vocabulary import (
@@ -227,39 +219,10 @@ def __call__(self):
             except Exception as e:
                 logger.error("Error retrieving object: {}".format(e))
                 return
-
-        # Set up FilesystemHelper to handle broken blob references, if any
-        conn = self.context._p_jar
-        storage = conn._storage
-        fshelper = storage.fshelper
-        base_dir = fshelper.base_dir
-        zeofshelper = FilesystemHelper(base_dir)
-        broken_items = []
-
-        if IDocument.providedBy(obj):
-            annotations = IAnnotations(obj)
-            for key in annotations.keys():
-                data = annotations[key]
-
-                # PersistentMapping tiles contain the blobs we need
-                if isinstance(data, PersistentMapping):
-                    for item in data.values():
-                        if isinstance(item, dict) and isinstance(item.get('data'), NamedBlobImage):
-                            blobfile = item.get('data')
-                            blob = blobfile._blob
-                            try:
-                                # Check blob
-                                blob._p_activate()
-                            except Exception as e:
-                                # Broken blob reference
-                                broken_items.append('/'.join(obj.getPhysicalPath()))
-                                filename = zeofshelper.getBlobFilename(blob._p_oid, blob._p_serial)
-                                if not os.path.exists(filename):
-                                    self.create_empty_blob(filename)     
 
         tasks.paste_items.delay(
             self.request.form['folder'], paste_data['op'],
-            paste_data['mdatas'], broken_items)
+            paste_data['mdatas'], fix_blobs=True)
 
         return self.do_redirect(
             self.canonical_object_url,

castle/cms/configure.zcml

@@ -86,6 +86,7 @@
   <include file="static.zcml" />
   <include file="theming.zcml" />
   <include file="meta.zcml" />
+  <include file="blobmissing.zcml" />
 
   <include zcml:condition="installed collective.easyform" package=".easyform" />
 

castle/cms/tasks/content.py

@@ -1,3 +1,4 @@
+
 from AccessControl import Unauthorized
 from Acquisition import aq_parent
 from castle.cms import cache
@@ -12,6 +13,7 @@
 
 import logging
 import transaction
+import os
 
 
 logger = logging.getLogger('castle.cms')
@@ -39,11 +41,14 @@ def paste_error_handle(where, op, mdatas):
 
 
 @retriable(on_retry_exhausted=paste_error_handle)
-def _paste_items(where, op, mdatas, broken_items):
+def _paste_items(where, op, mdatas, fix_blobs):
     logger.info('Copying a bunch of items')
     portal = api.portal.get()
     dest = portal.restrictedTraverse(str(where.lstrip('/')))
 
+    if fix_blobs:
+        os.environ["CASTLE_ALLOW_EXPERIMENTAL_BLOB_REPLACEMENT"] = 'True'
+
     if not getCelery().conf.task_always_eager:
         portal._p_jar.sync()
 
@@ -102,11 +107,13 @@ def _paste_items(where, op, mdatas, broken_items):
 
         except Exception:
             logger.warn('Could not send status email ', exc_info=True)
+
+    os.environ.pop("CASTLE_ALLOW_EXPERIMENTAL_BLOB_REPLACEMENT", None)
 
 
 @task()
-def paste_items(where, op, mdatas, broken_items=[]):
-    _paste_items(where, op, mdatas, broken_items)
+def paste_items(where, op, mdatas, fix_blobs=None):
+    _paste_items(where, op, mdatas, fix_blobs)
 
 
 def delete_error_handle(where, op, mdatas):