Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2000-2011 Regents of the University of California and the

 *              National Center for Ecological Analysis and Synthesis

 *

 *   '$Author:  $'

 *     '$Date:  $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat.dataone;

import java.io.IOException;

import java.io.InputStream;

import java.math.BigInteger;

import java.util.Calendar;

import java.util.Date;

import java.util.List;

import java.util.Set;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.io.IOUtils;

import org.apache.log4j.Logger;

import org.dataone.client.CNode;

import org.dataone.client.D1Client;

import org.dataone.service.cn.v1.CNAuthorization;

import org.dataone.service.cn.v1.CNCore;

import org.dataone.service.cn.v1.CNRead;

import org.dataone.service.cn.v1.CNReplication;

import org.dataone.service.exceptions.IdentifierNotUnique;

import org.dataone.service.exceptions.InsufficientResources;

import org.dataone.service.exceptions.InvalidRequest;

import org.dataone.service.exceptions.InvalidSystemMetadata;

import org.dataone.service.exceptions.InvalidToken;

import org.dataone.service.exceptions.NotAuthorized;

import org.dataone.service.exceptions.NotFound;

import org.dataone.service.exceptions.NotImplemented;

import org.dataone.service.exceptions.ServiceFailure;

import org.dataone.service.exceptions.UnsupportedType;

import org.dataone.service.types.v1.AccessPolicy;

import org.dataone.service.types.v1.Checksum;

import org.dataone.service.types.v1.Identifier;

import org.dataone.service.types.v1.Node;

import org.dataone.service.types.v1.NodeList;

import org.dataone.service.types.v1.NodeReference;

import org.dataone.service.types.v1.NodeType;

import org.dataone.service.types.v1.ObjectFormat;

import org.dataone.service.types.v1.ObjectFormatIdentifier;

import org.dataone.service.types.v1.ObjectFormatList;

import org.dataone.service.types.v1.ObjectList;

import org.dataone.service.types.v1.ObjectLocationList;

import org.dataone.service.types.v1.Permission;

import org.dataone.service.types.v1.Replica;

import org.dataone.service.types.v1.ReplicationPolicy;

import org.dataone.service.types.v1.ReplicationStatus;

import org.dataone.service.types.v1.Session;

import org.dataone.service.types.v1.Subject;

import org.dataone.service.types.v1.SystemMetadata;

import org.dataone.service.types.v1.util.ChecksumUtil;

import org.dataone.service.util.Constants;

import com.hazelcast.query.SqlPredicate;

import edu.ucsb.nceas.metacat.EventLog;

import edu.ucsb.nceas.metacat.IdentifierManager;

import edu.ucsb.nceas.metacat.McdbDocNotFoundException;

import edu.ucsb.nceas.metacat.dataone.hazelcast.HazelcastService;

/**

 * Represents Metacat's implementation of the DataONE Coordinating Node 

 * service API. Methods implement the various CN* interfaces, and methods common

 * to both Member Node and Coordinating Node interfaces are found in the

 * D1NodeService super class.

 *

 */

public class CNodeService extends D1NodeService implements CNAuthorization,

    CNCore, CNRead, CNReplication {

  /* the logger instance */

  private Logger logMetacat = null;

  /**

   * singleton accessor

   */

  public static CNodeService getInstance(HttpServletRequest request) { 

    return new CNodeService(request);

  }

  /**

   * Constructor, private for singleton access

   */

  private CNodeService(HttpServletRequest request) {

    super(request);

    logMetacat = Logger.getLogger(CNodeService.class);

  }

  /**

   * Set the replication policy for an object given the object identifier

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * @param policy - the replication policy to be applied

   * 

   * @return true or false

   * 

   * @throws NotImplemented

   * @throws NotAuthorized

   * @throws ServiceFailure

   * @throws InvalidRequest

   * 

   */

  @Override

  public boolean setReplicationPolicy(Session session, Identifier pid,

      ReplicationPolicy policy, long serialVersion) 

      throws NotImplemented, NotFound, NotAuthorized, ServiceFailure, 

      InvalidRequest, InvalidToken {

      // get the subject

      Subject subject = session.getSubject();

      // are we allowed to do this?

      if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {

        throw new NotAuthorized("4881", Permission.CHANGE_PERMISSION + 

            " not allowed by " + subject.getValue() + " on " + pid.getValue());  

      }

      SystemMetadata systemMetadata = null;

      try {

          HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

          systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

          // does the request have the most current system metadata?

          if ( systemMetadata.getSerialVersion().longValue() != serialVersion ) {

             String msg = "The requested system metadata version number " + 

                 serialVersion + "differs from the current version at " +

                 systemMetadata.getSerialVersion().longValue() +

                 " Please get the latest copy in order to modify it.";

             throw new InvalidRequest("4883", msg);

          }

      } catch (Exception e) { // Catch is generic since HZ throws RuntimeException

        throw new NotFound("4884", "No record found for: " + pid.getValue());

      }

      // set the new policy

      systemMetadata.setReplicationPolicy(policy);

      // update the metadata

      try {

        systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

        systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

	      HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);

	      HazelcastService.getInstance().getSystemMetadataMap().unlock(systemMetadata.getIdentifier());

      } catch (Exception e) {

		      throw new ServiceFailure("4882", e.getMessage());

	    } finally {

	        HazelcastService.getInstance().getSystemMetadataMap().unlock(systemMetadata.getIdentifier());

	    }

      return true;

  }

  /**

   * Set the replication status for an object given the object identifier

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * @param status - the replication status to be applied

   * 

   * @return true or false

   * 

   * @throws NotImplemented

   * @throws NotAuthorized

   * @throws ServiceFailure

   * @throws InvalidRequest

   * @throws InvalidToken

   * @throws NotFound

   * 

   */

  @Override

  public boolean setReplicationStatus(Session session, Identifier pid,

    NodeReference targetNode, ReplicationStatus status, long serialVersion) 

    throws ServiceFailure, NotImplemented, InvalidToken, NotAuthorized, 

    InvalidRequest, NotFound {

    // get the subject

    Subject subject = session.getSubject();

    // are we allowed to do this?

    if (!isAuthorized(session, pid, Permission.WRITE)) {

      throw new NotAuthorized("4720", Permission.WRITE + " not allowed by " + 

          subject.getValue() + " on " + pid.getValue());  

    }

    SystemMetadata systemMetadata = null;

    try {      

        HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

        systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

        // does the request have the most current system metadata?

        if ( systemMetadata.getSerialVersion().longValue() != serialVersion ) {

           String msg = "The requested system metadata version number " + 

               serialVersion + "differs from the current version at " +

               systemMetadata.getSerialVersion().longValue() +

               " Please get the latest copy in order to modify it.";

           throw new InvalidRequest("4730", msg);

        }

    } catch (Exception e) { // Catch is generic since HZ throws RuntimeException

      throw new NotFound("4740", "No record found for: " + pid.getValue() +

          " : " + e.getMessage());

    }

    // set the status for the replica

    List<Replica> replicas = systemMetadata.getReplicaList();

    for (Replica replica: replicas) {

        if (replica.getReplicaMemberNode().getValue().equals(targetNode.getValue())) {

            replica.setReplicationStatus(status);

        }

    }

    // [re]set the list -- redundant?

    systemMetadata.setReplicaList(replicas);

    // update the metadata

    try {

        systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

        systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

	      HazelcastService.getInstance().getSystemMetadataMap().put(systemMetadata.getIdentifier(), systemMetadata);

	      HazelcastService.getInstance().getSystemMetadataMap().unlock(systemMetadata.getIdentifier());

    } catch (Exception e) {

		    throw new ServiceFailure("4700", e.getMessage());

	  } finally {

	      HazelcastService.getInstance().getSystemMetadataMap().unlock(systemMetadata.getIdentifier());

	  }

    return true;

  }

  /**

   * Test that the specified relationship between pidOfSubject and pidOfObject exists

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param node - the node information for the given node be modified

   * 

   * @return true if the relationship exists

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  @Override

  public boolean assertRelation(Session session, Identifier pidOfSubject, 

    String relationship, Identifier pidOfObject) 

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, 

    InvalidRequest, NotImplemented {

    boolean asserted = false;

    // are we allowed to do this?

    if (!isAuthorized(session, pidOfSubject, Permission.READ)) {

      throw new NotAuthorized("4881", Permission.READ + " not allowed on " + pidOfSubject.getValue());  

    }

    SystemMetadata systemMetadata = null;

    try {

        HazelcastService.getInstance().getSystemMetadataMap().lock(pidOfSubject);

        systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pidOfSubject);

        // check relationships

        // TODO: use ORE map

        if (relationship.equalsIgnoreCase("describes")) {

        }

        if (relationship.equalsIgnoreCase("describedBy")) {

        }

        if (relationship.equalsIgnoreCase("derivedFrom")) {

        }

        if (relationship.equalsIgnoreCase("obsoletes")) {

            Identifier pid = systemMetadata.getObsoletes();

            if (pid.getValue().equals(pidOfObject.getValue())) {

              asserted = true;

        }

          //return systemMetadata.getObsoleteList().contains(pidOfObject);

        }

        if (relationship.equalsIgnoreCase("obsoletedBy")) {

            Identifier pid = systemMetadata.getObsoletedBy();

            if (pid.getValue().equals(pidOfObject.getValue())) {

              asserted = true;

        }

          //return systemMetadata.getObsoletedByList().contains(pidOfObject);

        }

        HazelcastService.getInstance().getSystemMetadataMap().unlock(pidOfSubject);

    } catch (Exception e) {

        throw new ServiceFailure("4270", "Could not assert relation for: " + 

            pidOfSubject.getValue() +

            ". The error message was: " + e.getMessage());

    } finally {

        HazelcastService.getInstance().getSystemMetadataMap().unlock(pidOfSubject);

    }

    return asserted;

  }

  /**

   * Return the checksum of the object given the identifier 

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * @return checksum - the checksum of the object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   */

  @Override

  public Checksum getChecksum(Session session, Identifier pid)

    throws InvalidToken, ServiceFailure, NotAuthorized, NotFound, 

    NotImplemented {

    if (!isAuthorized(session, pid, Permission.READ)) {

        throw new NotAuthorized("1400", Permission.READ + " not allowed on " + pid.getValue());  

    }

    SystemMetadata systemMetadata = null;

    Checksum checksum = null;

    try {

        HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

        systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

        checksum = systemMetadata.getChecksum();

        HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

    } catch (Exception e) {

        throw new ServiceFailure("1410", "An error occurred getting the checksum for " + 

            pid.getValue() + ". The error message was: " + e.getMessage());

    } finally {

        HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

    }

    return checksum;

  }

  /**

   * Resolve the location of a given object

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object

   * 

   * @return objectLocationList - the list of nodes known to contain the object

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws NotFound

   * @throws NotImplemented

   */

  @Override

  public ObjectLocationList resolve(Session session, Identifier pid)

    throws InvalidToken, ServiceFailure, NotAuthorized,

    NotFound, NotImplemented {

    throw new NotImplemented("4131", "resolve not implemented");

  }

  /**

   * Search the metadata catalog for identifiers that match the criteria

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param queryType - An identifier for the type of query expression 

   *                    provided in the query

   * @param query -  The criteria for matching the characteristics of the 

   *                 metadata objects of interest

   * 

   * @return objectList - the list of objects matching the criteria

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotImplemented

   */

  @Override

  public ObjectList search(Session session, String queryType, String query)

    throws InvalidToken, ServiceFailure, NotAuthorized, InvalidRequest,

    NotImplemented {

    ObjectList objectList = null;

    try {

        objectList = 

          IdentifierManager.getInstance().querySystemMetadata(

              null, //startTime, 

              null, //endTime,

              null, //objectFormat, 

              false, //replicaStatus, 

              0, //start, 

              -1 //count

              );

    } catch (Exception e) {

      throw new ServiceFailure("4310", "Error querying system metadata: " + e.getMessage());

    }

      return objectList;

    //throw new NotImplemented("4281", "search not implemented");

    // the code block below is from an older implementation

    /*  This block commented out because of the EcoGrid circular dependency.

         *  For now, query will not be supported until the circularity can be

         *  resolved, probably by moving the ecogrid query syntax transformers

         *  directly into the Metacat codebase.  MBJ 2010-02-03

        try {

            EcogridQueryParser parser = new EcogridQueryParser(request

                    .getReader());

            parser.parseXML();

            QueryType queryType = parser.getEcogridQuery();

            EcogridJavaToMetacatJavaQueryTransformer queryTransformer = 

                new EcogridJavaToMetacatJavaQueryTransformer();

            QuerySpecification metacatQuery = queryTransformer

                    .transform(queryType);

            DBQuery metacat = new DBQuery();

            boolean useXMLIndex = (new Boolean(PropertyService

                    .getProperty("database.usexmlindex"))).booleanValue();

            String xmlquery = "query"; // we don't care the query in resultset,

            // the query can be anything

            PrintWriter out = null; // we don't want metacat result, so set out null

            // parameter: queryspecification, user, group, usingIndexOrNot

            StringBuffer result = metacat.createResultDocument(xmlquery,

                    metacatQuery, out, username, groupNames, useXMLIndex);

            // create result set transfer       

            String saxparser = PropertyService.getProperty("xml.saxparser");

            MetacatResultsetParser metacatResultsetParser = new MetacatResultsetParser(

                    new StringReader(result.toString()), saxparser, queryType

                            .getNamespace().get_value());

            ResultsetType records = metacatResultsetParser.getEcogridResult();

            System.out

                    .println(EcogridResultsetTransformer.toXMLString(records));

            response.setContentType("text/xml");

            out = response.getWriter();

            out.print(EcogridResultsetTransformer.toXMLString(records));

        } catch (Exception e) {

            e.printStackTrace();

        }*/

  }

  /**

   * Returns the object format registered in the DataONE Object Format 

   * Vocabulary for the given format identifier

   * 

   * @param fmtid - the identifier of the format requested

   * 

   * @return objectFormat - the object format requested

   * 

   * @throws ServiceFailure

   * @throws NotFound

   * @throws InsufficientResources

   * @throws NotImplemented

   */

  @Override

  public ObjectFormat getFormat(ObjectFormatIdentifier fmtid)

    throws ServiceFailure, NotFound, InsufficientResources,

    NotImplemented {

      return ObjectFormatService.getInstance().getFormat(fmtid);

  }

  /**

   * Returns a list of all object formats registered in the DataONE Object 

   * Format Vocabulary

    * 

   * @return objectFormatList - The list of object formats registered in 

   *                            the DataONE Object Format Vocabulary

   * 

   * @throws ServiceFailure

   * @throws NotImplemented

   * @throws InsufficientResources

   */

  @Override

  public ObjectFormatList listFormats() 

    throws ServiceFailure, InsufficientResources, 

    NotImplemented {

    return ObjectFormatService.getInstance().listFormats();

  }

  /**

   * Returns a list of nodes that have been registered with the DataONE infrastructure

    * 

   * @return nodeList - List of nodes from the registry

   * 

   * @throws ServiceFailure

   * @throws NotImplemented

   */

  @Override

  public NodeList listNodes() 

    throws NotImplemented, ServiceFailure {

    throw new NotImplemented("4800", "listNodes not implemented");

  }

  /**

   * Provides a mechanism for adding system metadata independently of its 

   * associated object, such as when adding system metadata for data objects.

    * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The identifier of the object to register the system metadata against

   * @param sysmeta - The system metadata to be registered

   * 

   * @return true if the registration succeeds

   * 

   * @throws NotImplemented

   * @throws NotAuthorized

   * @throws ServiceFailure

   * @throws InvalidRequest

   * @throws InvalidSystemMetadata

   */

  @Override

  public Identifier registerSystemMetadata(Session session, Identifier pid,

      SystemMetadata sysmeta) 

      throws NotImplemented, NotAuthorized, ServiceFailure, InvalidRequest, 

      InvalidSystemMetadata {

      // TODO: control who can call this?

      if (session == null) {

          //TODO: many of the thrown exceptions do not use the correct error codes

          //check these against the docs and correct them

          throw new NotAuthorized("4861", "No Session - could not authorize for registration." +

                  "  If you are not logged in, please do so and retry the request.");

      }

      // verify that guid == SystemMetadata.getIdentifier()

      logMetacat.debug("Comparing guid|sysmeta_guid: " + pid.getValue() + 

          "|" + sysmeta.getIdentifier().getValue());

      if (!pid.getValue().equals(sysmeta.getIdentifier().getValue())) {

          throw new InvalidRequest("4863", 

              "The identifier in method call (" + pid.getValue() + 

              ") does not match identifier in system metadata (" +

              sysmeta.getIdentifier().getValue() + ").");

      }

      logMetacat.debug("Checking if identifier exists...");

      // Check that the identifier does not already exist

      if (HazelcastService.getInstance().getSystemMetadataMap().containsKey(pid)) {

          throw new InvalidRequest("4863", 

              "The identifier is already in use by an existing object.");

      }

      // insert the system metadata into the object store

      logMetacat.debug("Starting to insert SystemMetadata...");

      try {

          HazelcastService.getInstance().getSystemMetadataMap().lock(sysmeta.getIdentifier());

          sysmeta.setSerialVersion(BigInteger.ONE);

          sysmeta.setDateSysMetadataModified(Calendar.getInstance().getTime());

          HazelcastService.getInstance().getSystemMetadataMap().put(sysmeta.getIdentifier(), sysmeta);

          HazelcastService.getInstance().getSystemMetadataMap().unlock(sysmeta.getIdentifier());

      } catch (Exception e) {

      	logMetacat.error("Problem registering system metadata: " + pid.getValue(), e);

          throw new ServiceFailure("4862", "Error inserting system metadata: " + 

              e.getClass() + ": " + e.getMessage());

      } finally {

        HazelcastService.getInstance().getSystemMetadataMap().unlock(sysmeta.getIdentifier());

      }

      logMetacat.debug("Returning from registerSystemMetadata");

      EventLog.getInstance().log(request.getRemoteAddr(), 

          request.getHeader("User-Agent"), session.getSubject().getValue(), 

          pid.getValue(), "registerSystemMetadata");

      return pid;

  }

  /**

   * Given an optional scope and format, reserves and returns an identifier 

   * within that scope and format that is unique and will not be 

   * used by any other sessions. 

    * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The identifier of the object to register the system metadata against

   * @param scope - An optional string to be used to qualify the scope of 

   *                the identifier namespace, which is applied differently 

   *                depending on the format requested. If scope is not 

   *                supplied, a default scope will be used.

   * @param format - The optional name of the identifier format to be used, 

   *                  drawn from a DataONE-specific vocabulary of identifier 

   *                 format names, including several common syntaxes such 

   *                 as DOI, LSID, UUID, and LSRN, among others. If the 

   *                 format is not supplied by the caller, the CN service 

   *                 will use a default identifier format, which may change 

   *                 over time.

   * 

   * @return true if the registration succeeds

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws IdentifierNotUnique

   * @throws NotImplemented

   */

  @Override

  public Identifier reserveIdentifier(Session session, Identifier pid)

  throws InvalidToken, ServiceFailure,

        NotAuthorized, IdentifierNotUnique, NotImplemented, InvalidRequest {

    throw new NotImplemented("4191", "reserveIdentifier not implemented on this node");

  }

  @Override

  public Identifier generateIdentifier(Session session, String scheme, String fragment)

  throws InvalidToken, ServiceFailure,

        NotAuthorized, NotImplemented, InvalidRequest {

    throw new NotImplemented("4191", "generateIdentifier not implemented on this node");

  }

  /**

    * Checks whether the pid is reserved by the subject in the session param

    * If the reservation is held on the pid by the subject, we return true.

    * 

   * @param session - the Session object containing the Subject

   * @param pid - The identifier to check

   * 

   * @return true if the reservation exists for the subject/pid

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotFound - when the pid is not found (in use or in reservation)

   * @throws NotAuthorized - when the subject does not hold a reservation on the pid

   * @throws IdentifierNotUnique - when the pid is in use

   * @throws NotImplemented

   */

  @Override

  public boolean hasReservation(Session session, Identifier pid) 

      throws InvalidToken, ServiceFailure, NotFound, NotAuthorized, IdentifierNotUnique, 

      NotImplemented, InvalidRequest {

      throw new NotImplemented("4191", "hasReservation not implemented on this node");

  }

  /**

   * Changes ownership (RightsHolder) of the specified object to the 

   * subject specified by userId

    * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - Identifier of the object to be modified

   * @param userId - The subject that will be taking ownership of the specified object.

   *

   * @return pid - the identifier of the modified object

   * 

   * @throws ServiceFailure

   * @throws InvalidToken

   * @throws NotFound

   * @throws NotAuthorized

   * @throws NotImplemented

   * @throws InvalidRequest

   */  

  @Override

  public Identifier setOwner(Session session, Identifier pid, Subject userId,

      long serialVersion)

      throws InvalidToken, ServiceFailure, NotFound, NotAuthorized,

      NotImplemented, InvalidRequest {

      // get the subject

      Subject subject = session.getSubject();

      // are we allowed to do this?

      if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {

        throw new NotAuthorized("4440", "not allowed by " + subject.getValue() + " on " + pid.getValue());  

      }

      SystemMetadata systemMetadata = null;

      try {

          HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

          systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

          // does the request have the most current system metadata?

          if ( systemMetadata.getSerialVersion().longValue() != serialVersion ) {

             String msg = "The requested system metadata version number " + 

                 serialVersion + "differs from the current version at " +

                 systemMetadata.getSerialVersion().longValue() +

                 " Please get the latest copy in order to modify it.";

             throw new InvalidRequest("4442", msg);

          }

      } catch (Exception e) { // Catch is generic since HZ throws RuntimeException

          throw new NotFound("4460", "No record found for: " + pid.getValue());

      }

      // set the new rights holder

      systemMetadata.setRightsHolder(userId);

      // update the metadata

      try {

          systemMetadata.setSerialVersion(systemMetadata.getSerialVersion().add(BigInteger.ONE));

          systemMetadata.setDateSysMetadataModified(Calendar.getInstance().getTime());

	        HazelcastService.getInstance().getSystemMetadataMap().put(pid, systemMetadata);

	        HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

      } catch (Exception e) {

		  throw new ServiceFailure("4490", e.getMessage());

	    } finally {

	        HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

	    }

      return pid;

  }

  /**

   * Verify that a replication task is authorized by comparing the target node's

   * Subject (from the X.509 certificate-derived Session) with the list of 

   * subjects in the known, pending replication tasks map.

   * 

   * @param originatingNodeSession - Session information that contains the 

   *                                 identity of the calling user

   * @param targetNodeSubject - Subject identifying the target node

   * @param pid - the identifier of the object to be replicated

   * @param replicatePermission - the execute permission to be granted

   * 

   * @throws ServiceFailure

   * @throws NotImplemented

   * @throws InvalidToken

   * @throws NotAuthorized

   * @throws InvalidRequest

   * @throws NotFound

   */

  @Override

  public boolean isNodeAuthorized(Session originatingNodeSession, 

    Subject targetNodeSubject, Identifier pid, Permission replicatePermission) 

    throws NotImplemented, NotAuthorized, InvalidToken, ServiceFailure, 

    NotFound, InvalidRequest {

	  boolean isAllowed = false;

	  SystemMetadata sysmeta = null;

    NodeReference targetNode = null;

	  try {

	    // get the target node reference from the nodes list

	    CNode cn = D1Client.getCN();

	    List<Node> nodes = (List<Node>) cn.listNodes();

	    for ( Node node : nodes ) {

	        Subject nodeSubject = node.getSubject(0);

	        if (nodeSubject.getValue().equals(targetNodeSubject)) {

	            targetNode = node.getIdentifier();

	        }

	    }

	    //lock, get, and unlock the pid

	    HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

	    sysmeta = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

	    List<Replica> replicaList = sysmeta.getReplicaList();

      // find the replica with the status set to 'requested'

      for (Replica replica : replicaList) {

          ReplicationStatus status = replica.getReplicationStatus();

          NodeReference listedNode = replica.getReplicaMemberNode();

          if (listedNode.equals(targetNode)

                  && status.equals(ReplicationStatus.REQUESTED)) {

              isAllowed = true;

              break;

          }

      }

      HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

	  } catch(Exception e) {

	      // Catch Hazelcast RuntimeExceptions

	      throw new ServiceFailure("4872", 

	          "Couldn't determine if node is allowed: " + e.getMessage());

	  } finally {

	    // always unlock the pid

      HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

	  }

	  return isAllowed;

  }

  /**

   * Adds a new object to the Node, where the object is a science metadata object.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - The object identifier to be created

   * @param object - the object bytes

   * @param sysmeta - the system metadata that describes the object  

   * 

   * @return pid - the object identifier created

   * 

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotAuthorized

   * @throws IdentifierNotUnique

   * @throws UnsupportedType

   * @throws InsufficientResources

   * @throws InvalidSystemMetadata

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public Identifier create(Session session, Identifier pid, InputStream object,

    SystemMetadata sysmeta) 

    throws InvalidToken, ServiceFailure, NotAuthorized, IdentifierNotUnique, 

    UnsupportedType, InsufficientResources, InvalidSystemMetadata, 

    NotImplemented, InvalidRequest {

      try {

        // are we allowed?

          boolean isAllowed = false;

          CNode cn = D1Client.getCN();

          List<Node> nodes = (List<Node>) cn.listNodes();

          for (Node node : nodes) {

              if ( node.getType().equals(NodeType.CN) ) {

                  List<Subject> subjects = node.getSubjectList();

                  for (Subject subject : subjects) {

                     if (subject.getValue().equals(session.getSubject().getValue())) {

                         isAllowed = true;

                         break;

                     }

                  }

              }

          }

          // proceed if we're called by a CN

          if ( isAllowed ) {

              // create the coordinating node version of the document      

              HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

              sysmeta.setSerialVersion(BigInteger.ONE);

              sysmeta.setDateSysMetadataModified(Calendar.getInstance().getTime());

              pid = super.create(session, pid, object, sysmeta);

              HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

          } else {

              String msg = "The subject listed as " + session.getSubject().getValue() + 

                  " isn't allowed to call create() on a Coordinating Node.";

              logMetacat.info(msg);

              throw new NotAuthorized("1100", msg);

          }

      } catch (Exception e) {

          // Convert Hazelcast runtime exceptions to service failures

          String msg = "There was a problem creating the object identified by " +

              pid.getValue() + ". There error message was: " + e.getMessage();

      } finally {

          HazelcastService.getInstance().getSystemMetadataMap().unlock(pid);

      }

      return pid;

  }

  /**

   * Set access for a given object using the object identifier and a Subject

   * under a given Session.

   * 

   * @param session - the Session object containing the credentials for the Subject

   * @param pid - the object identifier for the given object to apply the policy

   * @param policy - the access policy to be applied

   * 

   * @return true if the application of the policy succeeds

   * @throws InvalidToken

   * @throws ServiceFailure

   * @throws NotFound

   * @throws NotAuthorized

   * @throws NotImplemented

   * @throws InvalidRequest

   */

  public boolean setAccessPolicy(Session session, Identifier pid, 

      AccessPolicy accessPolicy, long serialVersion) 

      throws InvalidToken, ServiceFailure, NotFound, NotAuthorized, 

      NotImplemented, InvalidRequest {

      boolean success = false;

      // get the subject

      Subject subject = session.getSubject();

      // are we allowed to do this?

      if (!isAuthorized(session, pid, Permission.CHANGE_PERMISSION)) {

          throw new NotAuthorized("4420", "not allowed by " + subject.getValue() + 

          " on " + pid.getValue());  

      }

      SystemMetadata systemMetadata = null;

      try {

          HazelcastService.getInstance().getSystemMetadataMap().lock(pid);

          systemMetadata = HazelcastService.getInstance().getSystemMetadataMap().get(pid);

          // does the request have the most current system metadata?

          if ( systemMetadata.getSerialVersion().longValue() != serialVersion ) {

             String msg = "The requested system metadata version number " + 

                 serialVersion + "differs from the current version at " +

                 systemMetadata.getSerialVersion().longValue() +

                 " Please get the latest copy in order to modify it.";

             throw new InvalidRequest("4402", msg);

          }

      } catch (Exception e) {

          // convert Hazelcast RuntimeException to NotFound

          throw new NotFound("4400", "No record found for: " + pid);

      }

      // set the access policy