Peler

: / opt / cpanel-ccs / txdav / common / datastore /
Filename : sql_notification.py
back
# -*- test-case-name: twext.enterprise.dal.test.test_record -*-
##
# Copyright (c) 2015-2017 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##

from twext.enterprise.dal.record import SerializableRecord, fromTable
from twext.enterprise.dal.syntax import Select, Parameter, Insert, \
    SavepointAction, Delete, Max, Len, Update
from twext.enterprise.util import parseSQLTimestamp
from twext.internet.decorate import memoizedKey
from twext.python.clsprop import classproperty
from twext.python.log import Logger
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.python.util import FancyEqMixin
from twistedcaldav.dateops import datetimeMktime
from txdav.base.propertystore.sql import PropertyStore
from txdav.common.datastore.sql_tables import schema, _HOME_STATUS_NORMAL, \
    _HOME_STATUS_EXTERNAL, _HOME_STATUS_DISABLED, _HOME_STATUS_MIGRATING
from txdav.common.datastore.sql_util import _SharedSyncLogic
from txdav.common.icommondatastore import RecordNotAllowedError, ConcurrentModification
from txdav.common.idirectoryservice import DirectoryRecordNotFoundError
from txdav.common.inotifications import INotificationCollection, \
    INotificationObject
from txdav.idav import ChangeCategory
from txweb2.dav.noneprops import NonePropertyStore
from txweb2.http_headers import MimeType
from zope.interface.declarations import implements
import hashlib
import json

"""
Classes and methods that relate to the Notification collection in the SQL store.
"""


class NotificationCollection(FancyEqMixin, _SharedSyncLogic):
    log = Logger()

    implements(INotificationCollection)

    compareAttributes = (
        "_ownerUID",
        "_resourceID",
    )

    _revisionsSchema = schema.NOTIFICATION_OBJECT_REVISIONS
    _homeSchema = schema.NOTIFICATION_HOME

    _externalClass = None

    @classmethod
    def makeClass(cls, transaction, homeData):
        """
        Build the actual home class taking into account the possibility that we might need to
        switch in the external version of the class.

        @param transaction: transaction
        @type transaction: L{CommonStoreTransaction}
        @param homeData: home table column data
        @type homeData: C{list}
        """

        status = homeData[cls.homeColumns().index(cls._homeSchema.STATUS)]
        if status == _HOME_STATUS_EXTERNAL:
            home = cls._externalClass(transaction, homeData)
        else:
            home = cls(transaction, homeData)
        return home.initFromStore()

    @classmethod
    def homeColumns(cls):
        """
        Return a list of column names to retrieve when doing an ownerUID->home lookup.
        """

        # Common behavior is to have created and modified

        return (
            cls._homeSchema.RESOURCE_ID,
            cls._homeSchema.OWNER_UID,
            cls._homeSchema.STATUS,
        )

    @classmethod
    def homeAttributes(cls):
        """
        Return a list of attributes names to map L{homeColumns} to.
        """

        # Common behavior is to have created and modified

        return (
            "_resourceID",
            "_ownerUID",
            "_status",
        )

    def __init__(self, txn, homeData):

        self._txn = txn

        for attr, value in zip(self.homeAttributes(), homeData):
            setattr(self, attr, value)

        self._txn = txn
        self._dataVersion = None
        self._notifications = {}
        self._notificationNames = None
        self._syncTokenRevision = None

        # Make sure we have push notifications setup to push on this collection
        # as well as the home it is in
        self._notifiers = dict([(factory_name, factory.newNotifier(self),) for factory_name, factory in txn._notifierFactories.items()])

    @inlineCallbacks
    def initFromStore(self):
        """
        Initialize this object from the store.
        """

        yield self._loadPropertyStore()
        returnValue(self)

    @property
    def _home(self):
        """
        L{NotificationCollection} serves as its own C{_home} for the purposes of
        working with L{_SharedSyncLogic}.
        """
        return self

    @classmethod
    def notificationsWithUID(cls, txn, uid, status=None, create=False):
        return cls.notificationsWith(txn, None, uid, status=status, create=create)

    @classmethod
    def notificationsWithResourceID(cls, txn, rid):
        return cls.notificationsWith(txn, rid, None)

    @classmethod
    @inlineCallbacks
    def notificationsWith(cls, txn, rid, uid, status=None, create=False):
        """
        @param uid: I'm going to assume uid is utf-8 encoded bytes
        """
        if rid is not None:
            query = cls._homeSchema.RESOURCE_ID == rid
        elif uid is not None:
            query = cls._homeSchema.OWNER_UID == uid
            if status is not None:
                query = query.And(cls._homeSchema.STATUS == status)
            else:
                statusSet = (_HOME_STATUS_NORMAL, _HOME_STATUS_EXTERNAL,)
                if txn._allowDisabled:
                    statusSet += (_HOME_STATUS_DISABLED,)
                query = query.And(cls._homeSchema.STATUS.In(statusSet))
        else:
            raise AssertionError("One of rid or uid must be set")

        results = yield Select(
            cls.homeColumns(),
            From=cls._homeSchema,
            Where=query,
        ).on(txn)

        if len(results) > 1:
            # Pick the best one in order: normal, disabled and external
            byStatus = dict([(result[cls.homeColumns().index(cls._homeSchema.STATUS)], result) for result in results])
            result = byStatus.get(_HOME_STATUS_NORMAL)
            if result is None:
                result = byStatus.get(_HOME_STATUS_DISABLED)
            if result is None:
                result = byStatus.get(_HOME_STATUS_EXTERNAL)
        elif results:
            result = results[0]
        else:
            result = None

        if result:
            # Return object that already exists in the store
            homeObject = yield cls.makeClass(txn, result)
            returnValue(homeObject)
        else:
            # Can only create when uid is specified
            if not create or uid is None:
                returnValue(None)

            # Determine if the user is local or external
            record = yield txn.directoryService().recordWithUID(uid.decode("utf-8"))
            if record is None:
                raise DirectoryRecordNotFoundError("Cannot create home for UID since no directory record exists: {}".format(uid))

            if status is None:
                createStatus = _HOME_STATUS_NORMAL if record.thisServer() else _HOME_STATUS_EXTERNAL
            elif status == _HOME_STATUS_MIGRATING:
                if record.thisServer():
                    raise RecordNotAllowedError("Cannot migrate a user data for a user already hosted on this server")
                createStatus = status
            elif status in (_HOME_STATUS_NORMAL, _HOME_STATUS_EXTERNAL,):
                createStatus = status
            else:
                raise RecordNotAllowedError("Cannot create home with status {}: {}".format(status, uid))

            # Use savepoint so we can do a partial rollback if there is a race
            # condition where this row has already been inserted
            savepoint = SavepointAction("notificationsWithUID")
            yield savepoint.acquire(txn)

            try:
                resourceid = (yield Insert(
                    {
                        cls._homeSchema.OWNER_UID: uid,
                        cls._homeSchema.STATUS: createStatus,
                    },
                    Return=cls._homeSchema.RESOURCE_ID
                ).on(txn))[0][0]
            except Exception:
                # FIXME: Really want to trap the pg.DatabaseError but in a non-
                # DB specific manner
                yield savepoint.rollback(txn)

                # Retry the query - row may exist now, if not re-raise
                results = yield Select(
                    cls.homeColumns(),
                    From=cls._homeSchema,
                    Where=query,
                ).on(txn)
                if results:
                    homeObject = yield cls.makeClass(txn, results[0])
                    returnValue(homeObject)
                else:
                    raise
            else:
                yield savepoint.release(txn)

                # Note that we must not cache the owner_uid->resource_id
                # mapping in the query cacher when creating as we don't want that to appear
                # until AFTER the commit
                results = yield Select(
                    cls.homeColumns(),
                    From=cls._homeSchema,
                    Where=cls._homeSchema.RESOURCE_ID == resourceid,
                ).on(txn)
                homeObject = yield cls.makeClass(txn, results[0])
                if homeObject.normal():
                    yield homeObject._initSyncToken()
                    yield homeObject.notifyChanged()
                returnValue(homeObject)

    @inlineCallbacks
    def _loadPropertyStore(self):
        self._propertyStore = yield PropertyStore.load(
            self._ownerUID,
            self._ownerUID,
            None,
            self._txn,
            self._resourceID,
            notifyCallback=self.notifyChanged
        )

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self._resourceID)

    def id(self):
        """
        Retrieve the store identifier for this collection.

        @return: store identifier.
        @rtype: C{int}
        """
        return self._resourceID

    @classproperty
    def _dataVersionQuery(cls):
        nh = cls._homeSchema
        return Select(
            [nh.DATAVERSION], From=nh,
            Where=nh.RESOURCE_ID == Parameter("resourceID")
        )

    @inlineCallbacks
    def dataVersion(self):
        if self._dataVersion is None:
            self._dataVersion = (yield self._dataVersionQuery.on(
                self._txn, resourceID=self._resourceID))[0][0]
        returnValue(self._dataVersion)

    def name(self):
        return "notification"

    def uid(self):
        return self._ownerUID

    def status(self):
        return self._status

    @inlineCallbacks
    def setStatus(self, newStatus):
        """
        Mark this home as being purged.
        """
        # Only if different
        if self._status != newStatus:
            yield Update(
                {self._homeSchema.STATUS: newStatus},
                Where=(self._homeSchema.RESOURCE_ID == self._resourceID),
            ).on(self._txn)
            self._status = newStatus

    def normal(self):
        """
        Is this an normal (internal) home.

        @return: a L{bool}.
        """
        return self._status == _HOME_STATUS_NORMAL

    def external(self):
        """
        Is this an external home.

        @return: a L{bool}.
        """
        return self._status == _HOME_STATUS_EXTERNAL

    def owned(self):
        return True

    def ownerHome(self):
        return self._home

    def viewerHome(self):
        return self._home

    def notificationObjectRecords(self):
        return NotificationObjectRecord.querysimple(self._txn, notificationHomeResourceID=self.id())

    @inlineCallbacks
    def notificationObjects(self):
        results = (yield NotificationObject.loadAllObjects(self))
        for result in results:
            self._notifications[result.uid()] = result
        self._notificationNames = sorted([result.name() for result in results])
        returnValue(results)

    _notificationUIDsForHomeQuery = Select(
        [schema.NOTIFICATION.NOTIFICATION_UID], From=schema.NOTIFICATION,
        Where=schema.NOTIFICATION.NOTIFICATION_HOME_RESOURCE_ID ==
        Parameter("resourceID"))

    @inlineCallbacks
    def listNotificationObjects(self):
        """
        List the names of all notification objects in this collection. Note that the name
        is actually the UID value with ".xml" appended, as per L{NotificationObject.name}.
        """
        if self._notificationNames is None:
            rows = yield self._notificationUIDsForHomeQuery.on(
                self._txn, resourceID=self._resourceID)
            self._notificationNames = sorted([row[0] + ".xml" for row in rows])
        returnValue(self._notificationNames)

    # used by _SharedSyncLogic.resourceNamesSinceRevision()
    def listObjectResources(self):
        return self.listNotificationObjects()

    def _nameToUID(self, name):
        """
        Based on the file-backed implementation, the 'name' is just uid +
        ".xml".
        """
        return name.rsplit(".", 1)[0]

    def notificationObjectWithName(self, name):
        return self.notificationObjectWithUID(self._nameToUID(name))

    @memoizedKey("uid", "_notifications")
    @inlineCallbacks
    def notificationObjectWithUID(self, uid):
        """
        Create an empty notification object first then have it initialize itself
        from the store.
        """
        no = NotificationObject(self, uid)
        no = (yield no.initFromStore())
        returnValue(no)

    @inlineCallbacks
    def writeNotificationObject(self, uid, notificationtype, notificationdata):

        inserting = False
        notificationObject = yield self.notificationObjectWithUID(uid)
        if notificationObject is None:
            notificationObject = NotificationObject(self, uid)
            inserting = True
        yield notificationObject.setData(uid, notificationtype, notificationdata, inserting=inserting)
        if inserting:
            yield self._insertRevision(notificationObject.name())
            if self._notificationNames is not None:
                self._notificationNames.append(notificationObject.name())
        else:
            yield self._updateRevision(notificationObject.name())
        yield self.notifyChanged()
        returnValue(notificationObject)

    def removeNotificationObjectWithName(self, name):
        if self._notificationNames is not None:
            self._notificationNames.remove(name)
        return self.removeNotificationObjectWithUID(self._nameToUID(name))

    _removeByUIDQuery = Delete(
        From=schema.NOTIFICATION,
        Where=(schema.NOTIFICATION.NOTIFICATION_UID == Parameter("uid")).And(
            schema.NOTIFICATION.NOTIFICATION_HOME_RESOURCE_ID == Parameter("resourceID"))
    )

    @inlineCallbacks
    def removeNotificationObjectWithUID(self, uid):
        yield self._removeByUIDQuery.on(
            self._txn, uid=uid, resourceID=self._resourceID)
        self._notifications.pop(uid, None)
        yield self._deleteRevision("%s.xml" % (uid,))
        yield self.notifyChanged()

    _initSyncTokenQuery = Insert(
        {
            _revisionsSchema.HOME_RESOURCE_ID: Parameter("resourceID"),
            _revisionsSchema.RESOURCE_NAME: None,
            _revisionsSchema.REVISION: schema.REVISION_SEQ,
            _revisionsSchema.DELETED: False
        }, Return=_revisionsSchema.REVISION
    )

    @inlineCallbacks
    def _initSyncToken(self):
        self._syncTokenRevision = (yield self._initSyncTokenQuery.on(
            self._txn, resourceID=self._resourceID))[0][0]

    _syncTokenQuery = Select(
        [Max(_revisionsSchema.REVISION)], From=_revisionsSchema,
        Where=_revisionsSchema.HOME_RESOURCE_ID == Parameter("resourceID")
    )

    @inlineCallbacks
    def syncToken(self):
        if self._syncTokenRevision is None:
            self._syncTokenRevision = yield self.syncTokenRevision()
        returnValue("%s_%s" % (self._resourceID, self._syncTokenRevision))

    @inlineCallbacks
    def syncTokenRevision(self):
        revision = (yield self._syncTokenQuery.on(self._txn, resourceID=self._resourceID))[0][0]
        if revision is None:
            revision = int((yield self._txn.calendarserverValue("MIN-VALID-REVISION")))
        returnValue(revision)

    def properties(self):
        return self._propertyStore

    def addNotifier(self, factory_name, notifier):
        self._notifiers[factory_name] = notifier

    def getNotifier(self, factory_name):
        return self._notifiers.get(factory_name)

    def notifierID(self):
        return (self._txn._homeClass[self._txn._primaryHomeType]._notifierPrefix, "%s/notification" % (self.ownerHome().uid(),),)

    def parentNotifierID(self):
        return (self._txn._homeClass[self._txn._primaryHomeType]._notifierPrefix, "%s" % (self.ownerHome().uid(),),)

    @inlineCallbacks
    def notifyChanged(self, category=ChangeCategory.default):
        """
        Send notifications, change sync token and bump last modified because
        the resource has changed.  We ensure we only do this once per object
        per transaction.
        """
        if self._txn.isNotifiedAlready(self):
            returnValue(None)
        self._txn.notificationAddedForObject(self)

        # Send notifications
        if self._notifiers:
            # cache notifiers run in post commit
            notifier = self._notifiers.get("cache", None)
            if notifier:
                self._txn.postCommit(notifier.notify)
            # push notifiers add their work items immediately
            notifier = self._notifiers.get("push", None)
            if notifier:
                yield notifier.notify(self._txn, priority=category.value)

        returnValue(None)

    @classproperty
    def _completelyNewRevisionQuery(cls):
        rev = cls._revisionsSchema
        return Insert({rev.HOME_RESOURCE_ID: Parameter("homeID"),
                       # rev.RESOURCE_ID: Parameter("resourceID"),
                       rev.RESOURCE_NAME: Parameter("name"),
                       rev.REVISION: schema.REVISION_SEQ,
                       rev.DELETED: False},
                      Return=rev.REVISION)

    def _maybeNotify(self):
        """
        Emit a push notification after C{_changeRevision}.
        """
        return self.notifyChanged()

    @inlineCallbacks
    def remove(self):
        """
        Remove DB rows corresponding to this notification home.
        """
        # Delete NOTIFICATION rows
        no = schema.NOTIFICATION
        kwds = {"ResourceID": self._resourceID}
        yield Delete(
            From=no,
            Where=(
                no.NOTIFICATION_HOME_RESOURCE_ID == Parameter("ResourceID")
            ),
        ).on(self._txn, **kwds)

        # Delete NOTIFICATION_HOME (will cascade to NOTIFICATION_OBJECT_REVISIONS)
        nh = schema.NOTIFICATION_HOME
        yield Delete(
            From=nh,
            Where=(
                nh.RESOURCE_ID == Parameter("ResourceID")
            ),
        ).on(self._txn, **kwds)

    purge = remove


class NotificationObjectRecord(SerializableRecord, fromTable(schema.NOTIFICATION)):
    """
    @DynamicAttrs
    L{Record} for L{schema.NOTIFICATION}.
    """
    pass


class NotificationObject(FancyEqMixin, object):
    """
    This used to store XML data and an XML element for the type. But we are now switching it
    to use JSON internally. The app layer will convert that to XML and fill in the "blanks" as
    needed for the app.
    """
    log = Logger()

    implements(INotificationObject)

    compareAttributes = (
        "_resourceID",
        "_home",
    )

    _objectSchema = schema.NOTIFICATION

    def __init__(self, home, uid):
        self._home = home
        self._resourceID = None
        self._uid = uid
        self._md5 = None
        self._size = None
        self._created = None
        self._modified = None
        self._notificationType = None
        self._notificationData = None

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self._resourceID)

    @classproperty
    def _allColumnsByHomeIDQuery(cls):
        """
        DAL query to load all columns by home ID.
        """
        obj = cls._objectSchema
        return Select(
            [obj.RESOURCE_ID, obj.NOTIFICATION_UID, obj.MD5,
             Len(obj.NOTIFICATION_DATA), obj.NOTIFICATION_TYPE, obj.CREATED, obj.MODIFIED],
            From=obj,
            Where=(obj.NOTIFICATION_HOME_RESOURCE_ID == Parameter("homeID"))
        )

    @classmethod
    @inlineCallbacks
    def loadAllObjects(cls, parent):
        """
        Load all child objects and return a list of them. This must create the
        child classes and initialize them using "batched" SQL operations to keep
        this constant wrt the number of children. This is an optimization for
        Depth:1 operations on the collection.
        """

        results = []

        # Load from the main table first
        dataRows = (
            yield cls._allColumnsByHomeIDQuery.on(parent._txn,
                                                  homeID=parent._resourceID))

        if dataRows:
            # Get property stores for all these child resources (if any found)
            propertyStores = (yield PropertyStore.forMultipleResources(
                parent.uid(),
                None,
                None,
                parent._txn,
                schema.NOTIFICATION.RESOURCE_ID,
                schema.NOTIFICATION.NOTIFICATION_HOME_RESOURCE_ID,
                parent._resourceID,
            ))

        # Create the actual objects merging in properties
        for row in dataRows:
            child = cls(parent, None)
            (child._resourceID,
             child._uid,
             child._md5,
             child._size,
             child._notificationType,
             child._created,
             child._modified,) = tuple(row)
            child._created = parseSQLTimestamp(child._created)
            child._modified = parseSQLTimestamp(child._modified)
            try:
                child._notificationType = json.loads(child._notificationType)
            except ValueError:
                pass
            if isinstance(child._notificationType, unicode):
                child._notificationType = child._notificationType.encode("utf-8")
            child._loadPropertyStore(
                props=propertyStores.get(child._resourceID, None)
            )
            results.append(child)

        returnValue(results)

    @classproperty
    def _oneNotificationQuery(cls):
        no = cls._objectSchema
        return Select(
            [
                no.RESOURCE_ID,
                no.MD5,
                Len(no.NOTIFICATION_DATA),
                no.NOTIFICATION_TYPE,
                no.CREATED,
                no.MODIFIED
            ],
            From=no,
            Where=(no.NOTIFICATION_UID ==
                   Parameter("uid")).And(no.NOTIFICATION_HOME_RESOURCE_ID ==
                                         Parameter("homeID")))

    @inlineCallbacks
    def initFromStore(self):
        """
        Initialise this object from the store, based on its UID and home
        resource ID. We read in and cache all the extra metadata from the DB to
        avoid having to do DB queries for those individually later.

        @return: L{self} if object exists in the DB, else C{None}
        """
        rows = (yield self._oneNotificationQuery.on(
            self._txn, uid=self._uid, homeID=self._home._resourceID))
        if rows:
            (self._resourceID,
             self._md5,
             self._size,
             self._notificationType,
             self._created,
             self._modified,) = tuple(rows[0])
            self._created = parseSQLTimestamp(self._created)
            self._modified = parseSQLTimestamp(self._modified)
            try:
                self._notificationType = json.loads(self._notificationType)
            except ValueError:
                pass
            if isinstance(self._notificationType, unicode):
                self._notificationType = self._notificationType.encode("utf-8")
            self._loadPropertyStore()
            returnValue(self)
        else:
            returnValue(None)

    def _loadPropertyStore(self, props=None, created=False):
        if props is None:
            props = NonePropertyStore(self._home.uid())
        self._propertyStore = props

    def properties(self):
        return self._propertyStore

    def id(self):
        """
        Retrieve the store identifier for this object.

        @return: store identifier.
        @rtype: C{int}
        """
        return self._resourceID

    @property
    def _txn(self):
        return self._home._txn

    def notificationCollection(self):
        return self._home

    def uid(self):
        return self._uid

    def name(self):
        return self.uid() + ".xml"

    @classproperty
    def _newNotificationQuery(cls):
        no = cls._objectSchema
        return Insert(
            {
                no.NOTIFICATION_HOME_RESOURCE_ID: Parameter("homeID"),
                no.NOTIFICATION_UID: Parameter("uid"),
                no.NOTIFICATION_TYPE: Parameter("notificationType"),
                no.NOTIFICATION_DATA: Parameter("notificationData"),
                no.MD5: Parameter("md5"),
            },
            Return=[no.RESOURCE_ID, no.CREATED, no.MODIFIED]
        )

    @classproperty
    def _updateNotificationQuery(cls):
        no = cls._objectSchema
        return Update(
            {
                no.NOTIFICATION_TYPE: Parameter("notificationType"),
                no.NOTIFICATION_DATA: Parameter("notificationData"),
                no.MD5: Parameter("md5"),
            },
            Where=(no.NOTIFICATION_HOME_RESOURCE_ID == Parameter("homeID")).And(
                no.NOTIFICATION_UID == Parameter("uid")),
            Return=no.MODIFIED
        )

    @inlineCallbacks
    def setData(self, uid, notificationtype, notificationdata, inserting=False):
        """
        Set the object resource data and update and cached metadata.
        """

        notificationtext = json.dumps(notificationdata)
        self._notificationType = notificationtype
        self._md5 = hashlib.md5(notificationtext).hexdigest()
        self._size = len(notificationtext)
        if inserting:
            rows = yield self._newNotificationQuery.on(
                self._txn, homeID=self._home._resourceID, uid=uid,
                notificationType=json.dumps(self._notificationType),
                notificationData=notificationtext, md5=self._md5
            )
            self._resourceID, self._created, self._modified = (
                rows[0][0],
                parseSQLTimestamp(rows[0][1]),
                parseSQLTimestamp(rows[0][2]),
            )
            self._loadPropertyStore()
        else:
            rows = yield self._updateNotificationQuery.on(
                self._txn, homeID=self._home._resourceID, uid=uid,
                notificationType=json.dumps(self._notificationType),
                notificationData=notificationtext, md5=self._md5
            )
            self._modified = parseSQLTimestamp(rows[0][0])
        self._notificationData = notificationdata

    _notificationDataFromID = Select(
        [_objectSchema.NOTIFICATION_DATA], From=_objectSchema,
        Where=_objectSchema.RESOURCE_ID == Parameter("resourceID"))

    @inlineCallbacks
    def notificationData(self):
        if self._notificationData is None:
            self._notificationData = (yield self._notificationDataFromID.on(self._txn, resourceID=self._resourceID))
            if self._notificationData:
                self._notificationData = self._notificationData[0][0]
            else:
                raise ConcurrentModification()
            try:
                self._notificationData = json.loads(self._notificationData)
            except ValueError:
                pass
            if isinstance(self._notificationData, unicode):
                self._notificationData = self._notificationData.encode("utf-8")
        returnValue(self._notificationData)

    def contentType(self):
        """
        The content type of NotificationObjects is text/xml.
        """
        return MimeType.fromString("text/xml")

    def md5(self):
        return self._md5

    def size(self):
        return self._size

    def notificationType(self):
        return self._notificationType

    def created(self):
        return datetimeMktime(self._created)

    def modified(self):
        return datetimeMktime(self._modified)