def Notify(self, message_type, subject, msg, source):
    """Send an AFF4-based notification to the user in the UI.

    Args:
      message_type: One of aff4_grr.Notification.notification_types e.g.
        "ViewObject", "HostInformation", "GrantAccess" or
        the same with an added ":[new-style notification type] suffix, e.g.
        "ViewObject:TYPE_CLIENT_INTERROGATED".
      subject: The subject to use, normally a URN.
      msg: The message to display.
      source: The class doing the notification.

    Raises:
      TypeError: On invalid message_type.
    """
    pending = self.Get(self.Schema.PENDING_NOTIFICATIONS)
    if pending is None:
      pending = self.Schema.PENDING_NOTIFICATIONS()

    # This is a legacy code that should go away after RELDB migration is done.
    # RELDB notifications have an explicitly stored notification type (that
    # is then mapped to ApiNotification.notification_type). However, legacy
    # AFF4-based data model notification types are different from the newly
    # introduced ones (new types are very concrete, i.e. "vfs file download
    # completed", legacy ones are very generic, i.e. "ViewObject").
    #
    # In order to temporarily keep the legacy code working and keep the tests
    # passing, we "pack" new-style notification type into an old style
    # "message_type". I.e. what was ViewObject before, will become
    # ViewObject:TYPE_CLIENT_INTERROGATED.
    #
    # Notifications without the ":[notification type]" suffix are also still
    # supported for backwards compatibility reasonds. They will be treated
    # as notifications with an unknown new-style type.
    if message_type.split(
        ":", 2)[0] not in rdf_flows.Notification.notification_types:
      raise TypeError("Invalid notification type %s" % message_type)

    pending.Append(
        type=message_type,
        subject=subject,
        message=msg,
        source=source,
        timestamp=int(time.time() * 1e6))

    # Limit the notification to 50, expiring older notifications.
    while len(pending) > 50:
      pending.Pop(0)

    self.Set(self.Schema.PENDING_NOTIFICATIONS, pending)