Metacat

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

import com.hazelcast.client.HazelcastClient;

import com.hazelcast.core.EntryEvent;

import com.hazelcast.core.EntryListener;

import com.hazelcast.core.Hazelcast;

import com.hazelcast.core.IMap;

import com.hazelcast.core.Member;

import com.hazelcast.partition.Partition;

import com.hazelcast.partition.PartitionService;

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

    CNCore, CNRead, CNReplication, EntryListener<Identifier, SystemMetadata> {

  /* The instance of the Hazelcast client */

  private HazelcastClient hzClient;

  /* The name of the DataONE Hazelcast cluster group */

  private String groupName;

  /* The name of the DataONE Hazelcast cluster password */

  private String groupPassword;

  /* The name of the DataONE Hazelcast cluster IP addresses */

  private String addressList;

  /* The name of the node map */

  private String nodeMap;

  /* The name of the system metadata map */

  private String systemMetadataMap;

  /* The Hazelcast distributed task id generator namespace */

  private String taskIds;

  /* The Hazelcast distributed system metadata map */

  private IMap<NodeReference, Node> nodes;

  /* The Hazelcast distributed system metadata map */

  private IMap<Identifier, SystemMetadata> systemMetadata;

  /* The Hazelcast distributed pending replication tasks map*/

  private IMap<String, CNReplicationTask> pendingReplicationTasks;

    // Get configuration properties on instantiation

    try {

      groupName = 

        PropertyService.getProperty("dataone.hazelcast.processCluster.groupName");

      groupPassword = 

        PropertyService.getProperty("dataone.hazelcast.processCluster.password");

      addressList = 

        PropertyService.getProperty("dataone.hazelcast.processCluster.instances");

      nodeMap = 

        PropertyService.getProperty("dataone.hazelcast.processCluster.nodesMap");

      systemMetadataMap = 

        PropertyService.getProperty("dataone.hazelcast.storageCluster.systemMetadataMap");

      // Become a DataONE-process cluster client

      //TODO: where should this be?

//      String[] addresses = addressList.split(",");

//      hzClient = 

//        HazelcastClient.newHazelcastClient(this.groupName, this.groupPassword, addresses);

//      nodes = hzClient.getMap(nodeMap);

//      pendingReplicationTasks = hzClient.getMap(pendingTasksQueue);

      // Get a reference to the shared system metadata map as a cluster member

      systemMetadata = Hazelcast.getMap(systemMetadataMap);

      // Listen for changes to the system metadata map

      systemMetadata.addEntryListener(this, true);

    } catch (PropertyNotFoundException e) {

      String msg = "Couldn't find Hazelcast properties for the DataONE clusters. " +

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

      logMetacat.error(msg);

    }

	      (Set<CNReplicationTask>) this.pendingReplicationTasks.values(new SqlPredicate(query));

  /**

	 * Implement the EntryListener interface for Hazelcast, reponding to entry

	 * added events in the hzSystemMetadata map. Evaluate the entry and create

	 * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)

	 * 

	 * @param event - The EntryEvent that occurred

	 */

	@Override

	public void entryAdded(EntryEvent<Identifier, SystemMetadata> event) {

		// handle as update - that method will create if necessary

		entryUpdated(event);

	}

  /**

   * Implement the EntryListener interface for Hazelcast, reponding to entry

   * evicted events in the hzSystemMetadata map.  Evaluate the entry and create

   * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)

   * 

   * @param event - The EntryEvent that occurred

   */

  @Override

  public void entryEvicted(EntryEvent<Identifier, SystemMetadata> event) {

    // nothing to do, entries are still in the backing store

  }

  /**

   * Implement the EntryListener interface for Hazelcast, reponding to entry

   * removed events in the hzSystemMetadata map.  Evaluate the entry and create

   * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)

   * 

   * @param event - The EntryEvent that occurred

   */

  @Override

  public void entryRemoved(EntryEvent<Identifier, SystemMetadata> event) {

    // we don't remove objects

  }

  /**

   * Implement the EntryListener interface for Hazelcast, reponding to entry

   * updated events in the hzSystemMetadata map.  Evaluate the entry and create

   * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)

   * 

   * @param event - The EntryEvent that occurred

   */

  @Override

  public void entryUpdated(EntryEvent<Identifier, SystemMetadata> event) {

		logMetacat.debug("Entry added/updated to System Metadata map: " + event.getKey().getValue());

		PartitionService partitionService = Hazelcast.getPartitionService();

		Partition partition = partitionService.getPartition(event.getKey());

		Member ownerMember = partition.getOwner();

		if (!ownerMember.localMember()) {

			// need to pull the entry into the local store

			logMetacat.debug("Saving entry locally: " + event.getKey().getValue());

			try {

				if (!IdentifierManager.getInstance().identifierExists(event.getKey().getValue())) {

					IdentifierManager.getInstance().createSystemMetadata(event.getValue());

				} else {

					IdentifierManager.getInstance().updateSystemMetadata(event.getValue());

				}

			} catch (McdbDocNotFoundException e) {

				logMetacat.error(

						"Could not save System Metadata to local store.", e);

			}

		}

		// TODO evaluate the type of system metadata change, decide if it

		// warrants a replication event, what type (DATA, METADATA, RESOURCE),

		// iteratively lock the PID, create and submit the tasks, and expect a

		// result back. Deal with exceptions.

		boolean isMetadata = D1NodeService.isScienceMetadata(event.getValue());

		// TODO: do we need to do anything explicit here?

  }