Metacat

/**

 * This work was created by participants in the DataONE project, and is

 * jointly copyrighted by participating institutions in DataONE. For

 * more information on DataONE, see our web site at http://dataone.org.

 *

 *   Copyright ${year}

 *

 * 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.

 */

package edu.ucsb.nceas.metacat.dataone.hazelcast;

import java.io.InputStream;

import java.io.Serializable;

import java.util.concurrent.Callable;

import org.dataone.client.D1Client;

import org.dataone.configuration.Settings;

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

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

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

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

import com.hazelcast.core.Hazelcast;

import com.hazelcast.core.IMap;

/**

 * A single CN replication task to be executed by the CN Replication Service. This 

 * applies to replication of system metadata, science metadata, and resource maps 

 * across CNs.

 * 

 * The task is built when a change occurs in the hzSystemMetadata map in the

 * storage cluster initiated by Metacat for a given PID. If the change involves

 * only modifications to system metadata, those changes will occur in the shared

 * map and in the backing store of the member that owns the map entry. Other 

 * members must be updated using this task.  Likewise, changes to the system

 * metadata map that involve science metadata creation or changes will be

 * distributed to all members via this task.  The same is for OAI-ORE resource maps.

 * 

 * @author cjones

 *

 */

public class CNReplicationTask implements Serializable, Callable<Identifier> {

  /* The identifier of this task */

  private String taskid;

  /* The identifier of the object to replicate */

  private Identifier pid;

  /* The object format type stated in the system metadata (DATA/METADATA/RESOURCE) */

  private String formatType;

  /* The target Node object */

  private Node targetNode;

  /* The originating Node object */

  private Node originatingNode;

  /* The subject of the target node, extracted from the Node object */

  private String targetNodeSubject;

  /* The subject of the originating node, extracted from the Node object */

  private String originatingNodeSubject;

  /* The permission to be executed (in this case, always 'replicate') */

  String permission;

  /**

   * Constructor - create an empty replication task instance

   */

  public CNReplicationTask() {

  }

  /**

   * Constructor - create a replication task instance

   * 

   * @param taskid

   * @param pid

   * @param targetNode

   */

  public CNReplicationTask(String taskid, Identifier pid, String formatType,

    Node originatingNode, Node targetNode,

    Permission replicatePermission) {

    this.taskid = taskid;

    this.pid = pid;

    this.formatType = formatType;

    this.originatingNode = originatingNode;

    this.targetNode = targetNode;

    this.originatingNodeSubject = originatingNode.getSubject(0).getValue();

    this.targetNodeSubject = targetNode.getSubject(0).getValue();

    this.permission = replicatePermission.name();

  }

  /**

   * Get the task identifier for this task

   * @return the taskid

   */

  public String getTaskid() {

    return taskid;

  }

  /**

   * Set the task identifier for this task

   * @param taskid the taskid to set

   */

  public void setTaskid(String taskid) {

    this.taskid = taskid;

  }

  /**

   * Get the object identifier to be replicated

   * @return the pid

   */

  public Identifier getPid() {

    return pid;

  }

  /**

   * Set the object identifier to be replicated

   * @param pid the pid to set

   */

  public void setPid(Identifier pid) {

    this.pid = pid;

  }

  /**

   * Get the format type of the object to be replicated

   * @return the pid

   */

  public String getFormatType() {

    return this.formatType;

  }

  /**

   * Set the format type of the object to be replicated

   * @param pid the pid to set

   */

  public void setFormatType(String formatType) {

    this.formatType = formatType;

  }

  /**

   * Get the target node

   * @return the targetNode

   */

  public Node getTargetNode() {

    return targetNode;

  }

  /**

   * Set the target node

   * @param targetNode the targetNode to set

   */

  public void setTargetNode(Node targetNode) {

    this.targetNode = targetNode;

  }

  /**

   * Get the originating node

   * @return the originatingNode

   */

  public Node getOriginatingNode() {

    return originatingNode;

  }

  /**

   * Set the originating node

   * @param originatingNode the originatingNode to set

   */

  public void setOriginatingNode(Node originatingNode) {

    this.originatingNode = originatingNode;

  }

  /**

   * For the given Replication task, return the Subject listed in the target

   * node.  Usually used in authorizing a replication event.

   * 

   * @return subject - the subject listed in the target Node object as a string

   */

  public String getTargetNodeSubject() {

    return this.targetNodeSubject;

  }

  /**

   * Set the target node subject identifying the node

   * @param subject the targetNode subject

   */

  public void setTargetNodeSubject(String subject) {

    this.targetNodeSubject = subject;

  }

  /**

   * For the given Replication task, return the Subject listed in the target

   * node.  Usually used in authorizing a replication event.

   * 

   * @return subject - the subject listed in the target Node object as a string

   */

  public String getOriginatingNodeSubject() {

    return this.originatingNodeSubject;

  }

  /**

   * Set the target node subject identifying the node

   * @param subject the targetNode subject

   */

  public void setOriginatingNodeSubject(String subject) {

    this.originatingNodeSubject = subject;

  }

  /**

   * Get the permission being allowed for this task

   * 

   * @return subject - the subject listed in the target Node object

   */

  public String getPermission() {

    return this.permission;

  }

  /**

   * Set the permission being allowed for this task

   * @param subject the targetNode subject

   */

  public void setPermission(Permission permission) {

    this.permission = permission.name();

  }

  /**

   * Implement the Callable interface, providing code that initiates replication.

   * 

   * @return pid - the identifier of the replicated object upon success

   */

  public Identifier call() {

	// Get the D1 Hazelcast configuration parameters

	String hzSystemMetadata = 

		Settings.getConfiguration().getString("dataone.hazelcast.systemMetadata");

	// get the system metadata for the pid	

	IMap<Identifier, SystemMetadata> sysMetaMap = Hazelcast.getMap(hzSystemMetadata);

	// get the systemMetadata

	SystemMetadata sm = sysMetaMap.get(pid);

	// TODO: need any info from the system metadata?

	// get the object from source

	InputStream sciMetaORE = null;

	try {

		// get from the MN

		sciMetaORE = D1Client.getMN(originatingNode.getIdentifier()).get(null, pid);

		// save it to target

		D1Client.getMN(targetNode.getIdentifier()).create(null, pid, sciMetaORE, sm);

	} catch (Exception e) {

		// report up the stack, even though we can't declare it

		throw new RuntimeException(e.getMessage(), e);

	}

    return pid;

  }

}