Metacat

/**

 *  '$RCSfile$'

 *    Purpose: A class that gets Accession Number, check for uniqueness

 *             and register it into db

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

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Jivka Bojilova, Matt Jones

 *

 *   '$Author: leinfelder $'

 *     '$Date: 2011-11-02 20:40:12 -0700 (Wed, 02 Nov 2011) $'

 * '$Revision: 6595 $'

 *

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

import java.io.FileInputStream;

import java.io.FileNotFoundException;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dataone.configuration.Settings;

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

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

import com.hazelcast.client.ClientConfig;

import com.hazelcast.client.HazelcastClient;

import com.hazelcast.config.Config;

import com.hazelcast.config.FileSystemXmlConfig;

import com.hazelcast.core.EntryEvent;

import com.hazelcast.core.EntryListener;

import com.hazelcast.core.IMap;

public class SystemMetadataEventListener implements EntryListener<Identifier, SystemMetadata> {

	private static Log log = LogFactory.getLog(SystemMetadataEventListener.class);

	private SolrIndex solrIndex = null;

	private HazelcastClient hzClient;

    private IMap<Identifier, SystemMetadata> systemMetadataMap;

    private IMap<Identifier, String> objectPathMap;

    /**

     * Default constructor - caller needs to initialize manually

     * @see setSolrIndex()

     * @see start()

     */

    public SystemMetadataEventListener() {

    }

    /**

     * Sets the SolrIndex instance and initializes the listener 

     * @param solrIndex

     */

    public SystemMetadataEventListener(SolrIndex solrIndex) {

    	this.solrIndex = solrIndex;

    	start();

    }

    /**

     * Get the SolrIndex that this listener communicates with

     * @return a SolrIndex instance that indexes the content being listened to

     */

    public SolrIndex getSolrIndex() {

		return solrIndex;

	}

    /**

     * Set the SolrIndex instance that the listener modifies

     * @param solrIndex

     */

	public void setSolrIndex(SolrIndex solrIndex) {

		this.solrIndex = solrIndex;

	}

	/**

     * Registers this instance as a system metadata map event listener

     */

    public void start() {

    	// get config values

        String hzSystemMetadata = Settings.getConfiguration().getString(

                "dataone.hazelcast.storageCluster.systemMetadataMap");

        String hzObjectPath = Settings.getConfiguration().getString(

                "dataone.hazelcast.storageCluster.objectPathMap");

        String configFileName = Settings.getConfiguration().getString(

        		"dataone.hazelcast.configFilePath");;

    	Config hzConfig = null;

		try {

    		hzConfig = new FileSystemXmlConfig(configFileName);

    	} catch (FileNotFoundException e) {

    		log.error("could not load hazelcast configuration file from: " + configFileName, e);

			//e.printStackTrace();

    	}

        String hzGroupName = hzConfig.getGroupConfig().getName();

        String hzGroupPassword = hzConfig.getGroupConfig().getPassword();

        String hzAddress = hzConfig.getNetworkConfig().getInterfaces().getInterfaces().iterator().next() + ":" + hzConfig.getNetworkConfig().getPort();

    	log.info("starting index entry listener...");

    	log.info("System Metadata value: " + hzSystemMetadata);

    	log.info("Object path value: " + hzObjectPath);

    	log.info("Group Name: " + hzGroupName);

    	log.info("Group Password: " + "*****"); // don't show value

    	log.info("HZ Address: " + hzAddress);

    	// connect to the HZ cluster

        ClientConfig cc = new ClientConfig();

		cc.getGroupConfig().setName(hzGroupName);

		cc.getGroupConfig().setPassword(hzGroupPassword);

		cc.addAddress(hzAddress);

        try {

        	this.hzClient = HazelcastClient.newHazelcastClient(cc);

        } catch (Exception e) {

            log.error("Unable to create hazelcast client: ", e);

            e.printStackTrace();

        }

        // get shared structures and add listener

        this.systemMetadataMap = this.hzClient.getMap(hzSystemMetadata);

        this.objectPathMap = this.hzClient.getMap(hzObjectPath);

        this.systemMetadataMap.addEntryListener(this, true);

        log.info("System Metadata size: " + this.systemMetadataMap.size());

        log.info("Object path size:" + this.objectPathMap.size());

    }

    /**

     * Removes this instance as a system metadata map event listener

     */

    public void stop() {

    	log.info("stopping index entry listener...");

        this.systemMetadataMap.removeEntryListener(this);

    }

    /**

     * Get the SystemMetadata for the specified id from the distributed Map.

     * The null maybe is returned if there is no system metadata found.

     * @param id  the specified id.

     * @return the SystemMetadata associated with the id.

     */

    public SystemMetadata getSystemMetadata(String id) {

        SystemMetadata metadata = null;

        if(systemMetadataMap != null && id != null) {

            Identifier identifier = new Identifier();

            identifier.setValue(id);

            metadata = systemMetadataMap.get(identifier);

        }

        return metadata;

    }

    /**

     * Get the obsoletes chain of the specified id. The returned list doesn't include

     * the specified id itself. The newer version has the lower index number in the list.

     * Empty list will be returned if there is no document to be obsoleted by this id.

     * @param id

     * @return

     */

    public List<String> getObsoletes(String id) {

        List<String> obsoletes = new ArrayList<String>();

        while (id != null) {

            SystemMetadata metadata = getSystemMetadata(id);

            id = null;//set it to be null in order to stop the while loop if the id can't be assinged to a new value in the following code.

            if(metadata != null) {

                Identifier identifier = metadata.getObsoletes();

                if(identifier != null && identifier.getValue() != null && !identifier.getValue().trim().equals("")) {

                    obsoletes.add(identifier.getValue());

                    id = identifier.getValue();

                } 

            } 

        }

        return obsoletes;

    }

    /**

     * Get an InputStream as the data object for the specific pid.

     * @param pid

     * @return

     * @throws FileNotFoundException

     */

    public InputStream getDataObject(String pid) throws FileNotFoundException {

        Identifier identifier = new Identifier();

        identifier.setValue(pid);

        String objectPath = objectPathMap.get(identifier);

        InputStream data = null;

        data = new FileInputStream(objectPath);

        return data;

    }

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

	    //System.out.println("===================================calling entryAdded method ");

	    log.info("===================================calling entryAdded method ");

		// use the same implementation for insert/update for now

		this.entryUpdated(entryEvent);

	}

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

		// remove from the index for now, this may be a temporary eviction

		this.entryRemoved(entryEvent);

	}

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

		// remove from the index

		Identifier pid = entryEvent.getKey();		

		try {

			solrIndex.remove(pid.getValue());

		} catch (Exception e) {

			// TODO: need to track errors, retry later

			log.error(e.getMessage(), e);

		}

	}

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

	    //System.out.println("===================================calling entryUpdated method ");

	    log.info("===================================calling entryUpdated method ");

		// add to the index

		Identifier pid = entryEvent.getKey();

		//System.out.println("===================================update the document "+pid.getValue());

		log.info("===================================update the document "+pid.getValue());

		SystemMetadata systemMetadata = entryEvent.getValue();

		Identifier obsoletes = systemMetadata.getObsoletes();

		List<String> obsoletesChain = null;

		if(obsoletes != null) {

		    obsoletesChain= getObsoletes(pid.getValue());

		}

		String objectPath = objectPathMap.get(pid);

		InputStream data = null;

		try {

			data = new FileInputStream(objectPath);

			solrIndex.update(pid.getValue(), obsoletesChain, systemMetadata, data);

		} catch (Exception e) {

			// TODO: need to track errors, retry later

			log.error(e.getMessage(), e);

		}

	}

}