def _MultiNotifyQueue(self, queue, notifications, timestamp=None, sync=True,
                        mutation_pool=None):
    serialized_notifications = {}
    now = rdfvalue.RDFDatetime().Now()
    expiry_time = config_lib.CONFIG["Worker.notification_expiry_time"]
    for notification in notifications:
      if not notification.first_queued:
        notification.first_queued = (self.frozen_timestamp or
                                     rdfvalue.RDFDatetime().Now())
      else:
        diff = now - notification.first_queued
        if diff.seconds >= expiry_time:
          # This notification has been around for too long, we drop it.
          logging.debug("Dropping notification: %s", str(notification))
          continue
      session_id = notification.session_id
      # Don't serialize session ids to save some bytes.
      notification.session_id = None
      notification.timestamp = None
      serialized_notifications[session_id] = notification.SerializeToString()

    values = {}
    for session_id, data in serialized_notifications.iteritems():
      values[self.NOTIFY_PREDICATE_TEMPLATE % session_id] = [(data, timestamp)]

    if mutation_pool:
      mutation_pool.MultiSet(self.GetNotificationShard(queue),
                             values, replace=False)
    else:
      self.data_store.MultiSet(
          self.GetNotificationShard(queue),
          values, sync=sync, replace=False, token=self.token)
Does the actual queuing.
dataset/ETHPy150Open google/grr/grr/lib/queue_manager.py/QueueManager._MultiNotifyQueue