#!/usr/bin/env python
"""A remote data store using HTTP."""


import base64
import binascii
import httplib
import logging
import random
import re
import socket
import threading
import time
import urlparse

from grr import config
from grr.lib import rdfvalue
from grr.lib import utils
from grr.lib.rdfvalues import data_server as rdf_data_server
from grr.lib.rdfvalues import data_store as rdf_data_store
from grr.lib.rdfvalues import protodict as rdf_protodict
from grr.server import access_control
from grr.server import data_store
from grr.server.data_server import auth

from grr.server.data_server import constants
from grr.server.data_server import utils as sutils
from grr.server.data_stores import common


def CheckResponseStatus(response):
  """Catch error conditions from the response and raise them."""
  # Common case exit early.
  if response.status == rdf_data_store.DataStoreResponse.Status.OK:
    return response

  elif (response.status ==
        rdf_data_store.DataStoreResponse.Status.AUTHORIZATION_DENIED):
    raise access_control.UnauthorizedAccess(response.status_desc,
                                            response.failed_subject)

  elif response.status == rdf_data_store.DataStoreResponse.Status.TIMEOUT_ERROR:
    raise data_store.TimeoutError(response.status_desc)

  elif (response.status ==
        rdf_data_store.DataStoreResponse.Status.DATA_STORE_ERROR):
    raise data_store.Error(response.status_desc)

  raise data_store.Error("Unknown error %s" % response.status_desc)


class Error(data_store.Error):
  """Base class for remote data store errors."""
  pass


class HTTPDataStoreError(Error):
  """Raised when there is a critical error in the remote data store."""
  pass


class DataServerConnection(object):
  """Represents one connection to a data server."""

  def __init__(self, server):
    self.conn = None
    self.sock = None
    self.lock = threading.Lock()
    self.server = server
    # Mark pending requests and subjects that are scheduled to change on
    # the database.
    self.requests = []
    self._DoConnection()

  def Address(self):
    return self.server.Address()

  def Port(self):
    return self.server.Port()

  def _ReadExactly(self, n):
    ret = ""
    left = n
    while left:
      data = self.sock.recv(left)
      if not data:
        raise IOError("Expected %d bytes, got EOF after %d" % (n, len(ret)))
      ret += data
      left = n - len(ret)
    return ret

  def _ReadReply(self):
    try:
      replylen_str = self._ReadExactly(sutils.SIZE_PACKER.size)
      replylen = sutils.SIZE_PACKER.unpack(replylen_str)[0]
      reply = self._ReadExactly(replylen)
      response = rdf_data_store.DataStoreResponse.FromSerializedString(reply)
      CheckResponseStatus(response)
      return response
    except (socket.error, socket.timeout, IOError) as e:
      logging.warning("Cannot read reply from server %s:%d : %s",
                      self.Address(), self.Port(), e)
      return None

  def _Sync(self):
    """Read responses from the pending requests."""
    self.sock.settimeout(config.CONFIG["HTTPDataStore.read_timeout"])
    while self.requests:
      response = self._ReadReply()
      if not response:
        # Could not read response. Let's exit and force a reconnection
        # followed by a replay.
        # TODO(user): Maybe we need to assign an unique ID for each
        # request so that the server knows which ones were already applied.
        return False
      self.requests.pop()
    return True

  def _SendRequest(self, command):
    request_str = command.SerializeToString()
    request_body = sutils.SIZE_PACKER.pack(len(request_str)) + request_str
    self.sock.settimeout(config.CONFIG["HTTPDataStore.send_timeout"])
    try:
      self.sock.sendall(request_body)
      return True
    except (socket.error, socket.timeout):
      logging.warning("Could not send request to server %s:%d",
                      self.Address(), self.Port())
      return False

  def _Reconnect(self):
    """Reconnect to the data server."""
    try:
      if self.sock:
        self.sock.close()
    except socket.error:
      pass
    try:
      if self.conn:
        self.conn.close()
    except httplib.HTTPException:
      pass
    try:
      self.conn = httplib.HTTPConnection(self.Address(), self.Port())
      username = config.CONFIG.Get("HTTPDataStore.username")
      password = config.CONFIG.Get("HTTPDataStore.password")
      if not username:
        raise HTTPDataStoreError("HTTPDataStore.username not provided")
      if not password:
        raise HTTPDataStoreError("HTTPDataStore.password not provided")
      # We ask the server for a nonce.
      self.conn.request("POST", "/client/handshake", "", {})
      response = self.conn.getresponse()
      if response.status != constants.RESPONSE_OK:
        logging.warning("Could not handshake the server %s:%d",
                        self.Address(), self.Port())
        return False
      # Generate the authentication token.
      size_nonce = int(response.getheader("Content-Length"))
      nonce = response.read(size_nonce)
      rdf_token = auth.NonceStore.GenerateAuthToken(nonce, username, password)
      token = rdf_token.SerializeToString()
      # We trick HTTP here and use the underlying socket to pipeline requests.
      headers = {"Content-Length": len(token)}
      self.conn.request("POST", "/client/start", token, headers)
      self.sock = self.conn.sock
      # Confirm handshake.
      self.sock.setblocking(1)
      self.sock.settimeout(config.CONFIG["HTTPDataStore.login_timeout"])
      ack = self._ReadExactly(3)
      if ack == "IP\n":
        raise HTTPDataStoreError("Invalid data server username/password.")
      if ack != "OK\n":
        return False
      logging.info("Connected to data server %s:%d", self.Address(),
                   self.Port())
      return True
    except httplib.HTTPException as e:
      logging.warning("Httplib problem when connecting to %s:%d: %s",
                      self.Address(), self.Port(), str(e))
      return False
    except (socket.error, socket.timeout, IOError) as e:
      logging.warning("Socket problem when connecting to %s:%d: %s",
                      self.Address(), self.Port(), str(e))
      return False
    return False

  def _ReplaySync(self):
    """Send all the requests again."""
    if self.requests:
      logging.info("Replaying the failed requests")
    while self.requests:
      req = self.requests[-1]
      if not self._SendRequest(req):
        return False
      self.sock.settimeout(config.CONFIG["HTTPDataStore.replay_timeout"])
      response = self._ReadReply()
      if not response:
        # Could not read response. Let's exit and force a reconnection
        # followed by a replay.
        # TODO(user): Maybe we need to assign an unique ID for each
        # request so that the server knows which ones were already applied.
        return False
      self.requests.pop()
    return True

  def _DoConnection(self):
    """Cleanups the current connection and creates another one."""
    started = time.time()
    while True:
      if self._Reconnect() and self._ReplaySync():
        break
      else:
        logging.warning("Had to connect to %s:%d but failed. Trying again...",
                        self.Address(), self.Port())
        # Sleep for some time before trying again.
        time.sleep(config.CONFIG["HTTPDataStore.retry_time"])
      if time.time(
      ) - started >= config.CONFIG["HTTPDataStore.reconnect_timeout"]:
        raise HTTPDataStoreError("Could not connect to %s:%d. Giving up." %
                                 (self.Address(), self.Port()))

  def _RedoConnection(self):
    logging.warning("Attempt to reconnect with %s:%d",
                    self.Address(), self.Port())
    self._DoConnection()

  @utils.Synchronized
  def MakeRequestAndContinue(self, command, unused_subject):
    """Make request but do not sync with the data server."""
    while not self._SendRequest(command):
      self._RedoConnection()
    self.requests.insert(0, command)
    return None

  @utils.Synchronized
  def SyncAndMakeRequest(self, command):
    """Make a request to the data server and return the response."""
    if not self._Sync():
      # Must reconnect and resend requests.
      self._RedoConnection()
    # At this point, we have a synchronized connection.
    while not self._SendRequest(command):
      self._RedoConnection()
    self.sock.settimeout(config.CONFIG["HTTPDataStore.read_timeout"])
    response = self._ReadReply()
    if not response:
      # Must reconnect and resend the request.
      while True:
        self._RedoConnection()
        if not self._SendRequest(command):
          continue
        response = self._ReadReply()
        if response:
          break
    return response

  @utils.Synchronized
  def Sync(self):
    if self._Sync():
      return True
    self._RedoConnection()
    return self._Sync()

  def NumPendingRequests(self):
    return len(self.requests)

  def Close(self):
    self.conn.close()


class DataServer(object):
  """A DataServer object contains connections a data server."""

  def __init__(self, addr, port):
    self.addr = addr
    self.port = port
    self.conn = httplib.HTTPConnection(self.Address(), self.Port())
    self.lock = threading.Lock()
    self.max_connections = config.CONFIG["Dataserver.max_connections"]
    # Start with a single connection.
    self.connections = [DataServerConnection(self)]

  def Port(self):
    return self.port

  def Address(self):
    return self.addr

  def Close(self):
    for conn in self.connections:
      conn.Close()
    self.connections = []
    if self.conn:
      self.conn.close()
      self.conn = None

  @utils.Synchronized
  def Sync(self):
    for conn in self.connections:
      # TODO(user): Consider adding error handling here.
      conn.Sync()

  @utils.Synchronized
  def GetConnection(self):
    """Return a connection to the data server."""
    best = min(self.connections, key=lambda x: x.NumPendingRequests())
    if best.NumPendingRequests():
      if len(self.connections) == self.max_connections:
        # Too many connections, use this one.
        return best
      new = DataServerConnection(self)
      self.connections.append(new)
      return new
    else:
      return best

    # Attempt to get one connection with no pending requests.
    return self.connections[0]

  def _FetchMapping(self):
    """Attempt to fetch mapping from the data server."""
    try:
      self.conn.request("POST", "/client/mapping")
      res = self.conn.getresponse()
      if res.status != constants.RESPONSE_OK:
        return None
      return res.read()
    except httplib.HTTPException:
      logging.warning("Could not connect server %s:%d",
                      self.Address(), self.Port())
      return None

  def LoadMapping(self):
    """Load mapping from the data server."""
    started = time.time()
    while True:
      data = self._FetchMapping()
      if data:
        mapping = rdf_data_server.DataServerMapping.FromSerializedString(data)
        return mapping

      if time.time(
      ) - started > config.CONFIG["HTTPDataStore.reconnect_timeout"]:
        raise HTTPDataStoreError("Could not get server mapping from data "
                                 "server at %s:%d." % (self.Address(),
                                                       self.Port()))
      time.sleep(config.CONFIG["HTTPDataStore.retry_time"])


class RemoteInquirer(object):
  """Class that holds connections to all data servers."""

  mapping = None

  def __init__(self):
    # Create a connection to all data servers
    server_list = config.CONFIG["Dataserver.server_list"]
    if not server_list:
      raise HTTPDataStoreError("List of data servers is not available.")
    self.servers = []
    for location in server_list:
      loc = urlparse.urlparse(location, scheme="http")
      self.servers.append(DataServer(loc.hostname, loc.port))
    self.mapping_server = random.choice(self.servers)
    self.mapping = self.mapping_server.LoadMapping()

    if len(self.mapping.servers) != len(server_list):
      logging.warning("There is a mismatch between the data "
                      "servers and the configuration file. '%s' != '%s'",
                      self.mapping.servers, server_list)
      raise HTTPDataStoreError("There is a mismatch between the data "
                               "servers and the configuration file.")
    for i, serv in enumerate(self.mapping.servers):
      target = self.servers[i]
      if target.Port() != serv.port:
        logging.warning("There is a mismatch between the data "
                        "servers and the configuration file. '%s' != '%s'",
                        self.mapping.servers, server_list)
        raise HTTPDataStoreError("There is a mismatch between the data "
                                 "servers and the configuration file.")

  def MapKey(self, key):
    """Return the data server responsible for a given key."""
    sid = sutils.MapKeyToServer(self.mapping, key)
    return self.servers[sid]

  def GetPathing(self):
    return self.GetMapping().pathing

  def RenewMapping(self):
    self.mapping = self.mapping_server.LoadMapping()
    return self.mapping

  def GetMapping(self):
    return self.mapping

  def Flush(self):
    for serv in self.servers:
      serv.Sync()

  def CloseConnections(self):
    for serv in self.servers:
      serv.Close()


class RemoteMappingCache(utils.FastStore):
  """A local cache for mappings between paths and data servers."""

  def __init__(self, size):
    super(RemoteMappingCache, self).__init__(size)
    self.inquirer = RemoteInquirer()
    self.path_regexes = [re.compile(x) for x in self.inquirer.GetPathing()]

  def KillObject(self, obj):
    pass

  def GetInquirer(self):
    return self.inquirer

  @utils.Synchronized
  def Get(self, subject):
    """This will create the object if needed so should not fail."""
    filename, directory = common.ResolveSubjectDestination(
        subject, self.path_regexes)
    key = common.MakeDestinationKey(directory, filename)
    try:
      return super(RemoteMappingCache, self).Get(key)
    except KeyError:
      data_server = self.inquirer.MapKey(key)

      super(RemoteMappingCache, self).Put(key, data_server)

      return data_server

  def AllDatabases(self):
    for server in self.inquirer.servers:
      yield server

  def GetPrefix(self, prefix):
    """Yields all databases which could contain records begining with prefix."""
    components = common.Components(prefix)

    components = [common.ConvertStringToFilename(x) for x in components]
    path_prefix = utils.JoinPath(*components)
    if path_prefix == "/":
      path_prefix = ""

    for regex in self.path_regexes:
      result = common.EvaluatePrefix(path_prefix, regex)
      if result == "MATCH":
        yield self.Get(prefix)
        return
      if result == "POSSIBLE":
        for data_server in self.AllDatabases():
          yield data_server
        return
    yield self.Get(prefix)


class HTTPDataStore(data_store.DataStore):
  """A data store which calls a remote server."""

  cache = None
  inquirer = None

  def __init__(self):
    super(HTTPDataStore, self).__init__()
    self.cache = RemoteMappingCache(1000)
    self.inquirer = self.cache.GetInquirer()
    self._ComputeNewSize(self.inquirer.GetMapping(), time.time())

  def GetServer(self, subject):
    return self.cache.Get(subject).GetConnection()

  def GetServersForPrefix(self, prefix):
    for s in self.cache.GetPrefix(prefix):
      yield s.GetConnection()

  def TimestampSpecFromTimestamp(self, timestamp):
    """Create a timestamp spec from a timestamp value.

    Args:
      timestamp: A range of times for consideration (In
          microseconds). Can be a constant such as ALL_TIMESTAMPS or
          NEWEST_TIMESTAMP or a tuple of ints (start, end).

    Returns:
       An rdfvalue.TimestampSpec() instance.
    """
    if timestamp is None:
      all_ts = rdf_data_store.TimestampSpec.Type.ALL_TIMESTAMPS
      return rdf_data_store.TimestampSpec(type=all_ts)

    if timestamp in (rdf_data_store.TimestampSpec.Type.ALL_TIMESTAMPS,
                     rdf_data_store.TimestampSpec.Type.NEWEST_TIMESTAMP):
      return rdf_data_store.TimestampSpec(type=timestamp)

    if timestamp == self.NEWEST_TIMESTAMP:
      newest = rdf_data_store.TimestampSpec.Type.NEWEST_TIMESTAMP
      return rdf_data_store.TimestampSpec(type=newest)

    if isinstance(timestamp, (list, tuple)):
      start, end = timestamp
      return rdf_data_store.TimestampSpec(
          start=start,
          end=end,
          type=rdf_data_store.TimestampSpec.Type.RANGED_TIME)

    return rdf_data_store.TimestampSpec(
        start=timestamp, type=rdf_data_store.TimestampSpec.Type.SPECIFIC_TIME)

  def _MakeSyncRequest(self, request, typ):
    return self._MakeRequestSyncOrAsync(request, typ, True)

  def _MakeRequestSyncOrAsync(self, request, typ, sync):
    subject = request.subject[0]
    server = self.GetServer(subject)
    cmd = rdf_data_server.DataStoreCommand(command=typ, request=request)
    if sync:
      return server.SyncAndMakeRequest(cmd)
    else:
      return server.MakeRequestAndContinue(cmd, subject)

  def _MakeRequestsForPrefix(self, prefix, typ, request):
    cmd = rdf_data_server.DataStoreCommand(command=typ, request=request)
    for server in self.GetServersForPrefix(prefix):
      yield server.SyncAndMakeRequest(cmd)

  def DeleteAttributes(self,
                       subject,
                       attributes,
                       start=None,
                       end=None,
                       sync=True,
                       token=None):
    request = rdf_data_store.DataStoreRequest(subject=[subject])

    if isinstance(attributes, basestring):
      raise ValueError(
          "String passed to DeleteAttributes (non string iterable expected).")

    # Set timestamp.
    start = start or 0
    if end is None:
      end = (2**63) - 1  # sys.maxint

    request.timestamp = rdf_data_store.TimestampSpec(
        start=start,
        end=end,
        type=rdf_data_store.TimestampSpec.Type.RANGED_TIME)

    if token:
      request.token = token
    if sync:
      request.sync = sync

    for attr in attributes:
      request.values.Append(attribute=attr)

    typ = rdf_data_server.DataStoreCommand.Command.DELETE_ATTRIBUTES
    self._MakeRequestSyncOrAsync(request, typ, sync)

  def DeleteSubject(self, subject, sync=False, token=None):
    request = rdf_data_store.DataStoreRequest(subject=[subject])
    if token:
      request.token = token

    typ = rdf_data_server.DataStoreCommand.Command.DELETE_SUBJECT
    self._MakeRequestSyncOrAsync(request, typ, sync)

  def _MakeRequest(self,
                   subjects,
                   attributes,
                   timestamp=None,
                   token=None,
                   limit=None):
    if isinstance(attributes, basestring):
      attributes = [attributes]

    request = rdf_data_store.DataStoreRequest(subject=subjects)
    if limit:
      request.limit = limit

    token = token or data_store.default_token
    if token:
      request.token = token

    if timestamp is not None:
      request.timestamp = self.TimestampSpecFromTimestamp(timestamp)

    for attribute in attributes:
      request.values.Append(attribute=attribute)

    return request

  def MultiResolvePrefix(self,
                         subjects,
                         attribute_prefix,
                         timestamp=None,
                         limit=None,
                         token=None):
    """MultiResolvePrefix."""
    typ = rdf_data_server.DataStoreCommand.Command.MULTI_RESOLVE_PREFIX
    results = {}
    remaining_limit = limit
    for subject in subjects:
      request = self._MakeRequest(
          [subject],
          attribute_prefix,
          timestamp=timestamp,
          token=token,
          limit=remaining_limit)

      response = self._MakeSyncRequest(request, typ)

      if response.results:
        result_set = response.results[0]
        values = [(pred, self._Decode(value), ts)
                  for (pred, value, ts) in result_set.payload]
        if limit:
          if len(values) >= remaining_limit:
            results[subject] = values[:remaining_limit]
            return results.iteritems()
          remaining_limit -= len(values)

        results[subject] = values
    return results.iteritems()

  def ScanAttributes(self,
                     subject_prefix,
                     attributes,
                     after_urn=None,
                     max_records=None,
                     token=None,
                     relaxed_order=False):
    """ScanAttribute."""

    subject_prefix = utils.SmartStr(rdfvalue.RDFURN(subject_prefix))
    if subject_prefix[-1] != "/":
      subject_prefix += "/"

    typ = rdf_data_server.DataStoreCommand.Command.SCAN_ATTRIBUTES
    subjects = [subject_prefix]
    if after_urn:
      subjects.append(after_urn)
    request = self._MakeRequest(
        subjects, attributes, token=token, limit=max_records)
    if relaxed_order:
      for response in self._MakeRequestsForPrefix(subject_prefix, typ, request):
        for result in response.results:
          values = {}
          for attribute, (ts, value) in result.payload:
            values[attribute] = (ts, self._Decode(value))
          yield (result.subject, values)
    else:
      results = []
      for response in self._MakeRequestsForPrefix(subject_prefix, typ, request):
        for result in response.results:
          values = {}
          for attribute, (ts, value) in result.payload:
            values[attribute] = (ts, self._Decode(value))
          results.append((result.subject, values))
      for r in sorted(results, key=lambda x: x[0]):
        yield r

  def MultiSet(self,
               subject,
               values,
               timestamp=None,
               replace=True,
               sync=True,
               to_delete=None,
               token=None):
    """MultiSet."""
    request = rdf_data_store.DataStoreRequest(sync=sync)
    token = token or data_store.default_token
    if token:
      request.token = token

    request.subject.Append(subject)
    now = time.time() * 1000000

    if timestamp is None or timestamp == self.NEWEST_TIMESTAMP:
      timestamp = now

    to_delete = set(to_delete or [])
    for attribute in to_delete:
      if attribute not in values:
        values[attribute] = [(None, 0)]

    for k, seq in values.items():
      for v in seq:
        if isinstance(v, basestring):
          element_timestamp = timestamp
        else:
          try:
            v, element_timestamp = v
          except (TypeError, ValueError):
            element_timestamp = timestamp

        option = rdf_data_store.DataStoreValue.Option.DEFAULT
        if replace or k in to_delete:
          option = rdf_data_store.DataStoreValue.Option.REPLACE

        new_value = request.values.Append(
            attribute=utils.SmartUnicode(k), option=option)

        if element_timestamp is None:
          element_timestamp = now

        new_value.timestamp = self.TimestampSpecFromTimestamp(element_timestamp)

        if v is not None:
          new_value.value.SetValue(v)

    typ = rdf_data_server.DataStoreCommand.Command.MULTI_SET
    self._MakeRequestSyncOrAsync(request, typ, sync)

  def ResolveMulti(self,
                   subject,
                   attributes,
                   timestamp=None,
                   limit=None,
                   token=None):
    """ResolveMulti."""
    request = self._MakeRequest(
        [subject], attributes, timestamp=timestamp, limit=limit, token=token)

    typ = rdf_data_server.DataStoreCommand.Command.RESOLVE_MULTI
    response = self._MakeSyncRequest(request, typ)

    results = []
    for result in response.results:
      for (attribute, value, timestamp) in result.payload:
        results.append((attribute, self._Decode(value), timestamp))
    return results

  def _Decode(self, value):
    """Decodes strings from serialized responses."""
    result = value

    if isinstance(value, (tuple, list)):
      try:
        base64value, one = value
        if one == 1:
          result = base64.decodestring(base64value)
      except (ValueError, binascii.Error):
        pass

    return result

  def DBSubjectLock(self, subject, lease_time=None, token=None):
    """We do not support locks directly."""
    return HTTPDBSubjectLock(self, subject, lease_time=lease_time, token=token)

  def LockSubject(self, subject, lease_time, token):
    """Locks a specific subject."""
    request = rdf_data_store.DataStoreRequest(subject=[subject])
    specific = rdf_data_store.TimestampSpec.Type.SPECIFIC_TIME
    request.timestamp = rdf_data_store.TimestampSpec(
        start=lease_time, type=specific)

    if token:
      request.token = token

    typ = rdf_data_server.DataStoreCommand.Command.LOCK_SUBJECT
    response = self._MakeSyncRequest(request, typ)

    if not response.results:
      return None
    result = response.results[0]
    if not result.values:
      return None
    return result.values[0].value.string

  def ExtendSubjectLock(self, subject, transid, lease_time, token):
    """Extends lock of subject."""
    request = rdf_data_store.DataStoreRequest(subject=[subject])
    specific = rdf_data_store.TimestampSpec.Type.SPECIFIC_TIME
    request.timestamp = rdf_data_store.TimestampSpec(
        start=lease_time, type=specific)
    if token:
      request.token = token
    blob = rdf_protodict.DataBlob(string=transid)
    value = rdf_data_store.DataStoreValue(value=blob)
    request.values.Append(value)

    typ = rdf_data_server.DataStoreCommand.Command.EXTEND_SUBJECT
    response = self._MakeSyncRequest(request, typ)

    if not response.results:
      return None
    result = response.results[0]
    if not result.values:
      return None
    value = result.values[0].value.string
    return transid if transid == value else None

  def UnlockSubject(self, subject, transid, token):
    """Unlocks subject using lock id."""
    request = rdf_data_store.DataStoreRequest(subject=[subject])
    if token:
      request.token = token
    blob = rdf_protodict.DataBlob(string=transid)
    value = rdf_data_store.DataStoreValue(value=blob)
    request.values.Append(value)

    # We do not care about the server response.
    typ = rdf_data_server.DataStoreCommand.Command.UNLOCK_SUBJECT
    self._MakeSyncRequest(request, typ)

    return transid

  def Flush(self):
    super(HTTPDataStore, self).Flush()
    if self.inquirer:
      self.inquirer.Flush()

  def CloseConnections(self):
    if self.inquirer:
      self.inquirer.CloseConnections()

  def _ComputeNewSize(self, mapping, new_time):
    self.last_size = 0
    for serv in mapping.servers:
      self.last_size += serv.state.size
    self.last_size_update = new_time

  def Size(self):
    """Get size of data store."""
    now = time.time()
    if now < self.last_size_update + 60:
      return self.last_size
    mapping = self.inquirer.RenewMapping()
    self._ComputeNewSize(mapping, now)
    return self.last_size


class HTTPDBSubjectLock(data_store.DBSubjectLock):
  """The opensource remote data store subject lock.

  We only ensure that two simultaneous locks can not be held on the
  same subject.

  This means that the first thread which grabs the lock is considered the owner
  of the lock. Any subsequent locks on the same subject will fail
  immediately with data_store.DBSubjectLockError. NOTE that it is still possible
  to manipulate the row without a transaction - this is a design feature!

  A lock is considered expired after a certain time.
  """
  locked = False

  def _Acquire(self, lease_time):
    self.transid = self.store.LockSubject(self.subject, lease_time * 1e6,
                                          self.token)
    if not self.transid:
      raise data_store.DBSubjectLockError(
          "Unable to lock subject %s" % self.subject)
    self.expires = int((time.time() + lease_time) * 1e6)
    self.locked = True

  def UpdateLease(self, duration):
    ret = self.store.ExtendSubjectLock(self.subject, self.transid,
                                       duration * 1e6, self.token)
    if ret != self.transid:
      raise data_store.DBSubjectLockError(
          "Unable to update the lease on %s" % self.subject)
    self.expires = int((time.time() + duration) * 1e6)

  def Release(self):
    if self.locked:
      self.store.UnlockSubject(self.subject, self.transid, self.token)
      self.locked = False
