D-Net

package eu.dnetlib.actionmanager.hbase;

import static org.junit.Assert.assertNotNull;

import static org.junit.Assert.assertTrue;

import java.io.StringReader;

import java.io.StringWriter;

import javax.annotation.Resource;

import javax.xml.transform.Transformer;

import org.apache.commons.io.IOUtils;

import org.dom4j.Document;

import org.dom4j.io.DocumentResult;

import org.dom4j.io.DocumentSource;

import org.dom4j.io.SAXReader;

import org.junit.Before;

import org.junit.Ignore;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.test.context.ContextConfiguration;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import eu.dnetlib.actionmanager.ActionManagerConstants.ACTION_TYPE;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Agent.AGENT_TYPE;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

@Ignore

@RunWith(SpringJUnit4ClassRunner.class)

@ContextConfiguration(locations = { "/eu/dnetlib/actionmanager/test/applicationContext-actionmanager-test.xml" })

public class HBaseActionManagerTest {

	@Resource

	private HBaseActionManagerCore core;

	private String dmfInfoPackage;

	private static final String SET = "acus";

	@Before

	public void setUp() throws Exception {

		StringWriter sw = new StringWriter();

		IOUtils.copy(getClass().getResourceAsStream("/eu/dnetlib/actionmanager/test/record.xml"), sw);

		dmfInfoPackage = sw.toString();

	}

	@Test

	public void testBeans() {

		assertNotNull(core);

		assertNotNull(core.getActionFactory());

		assertNotNull(core.getHbaseClient());

	}

	@Test

	public void testXSLT() throws Exception {

		Transformer transformer = core.getActionFactory().prepareXsltTransformer("dmf2actions");

		assertNotNull(transformer);

		Document doc = (new SAXReader()).read(new StringReader(dmfInfoPackage));

		final DocumentResult result = new DocumentResult();

		transformer.transform(new DocumentSource(doc), result);

		int count = result.getDocument().selectNodes("//ACTION").size();

		System.out.println("******************************************");

		System.out.println("N. Actions: " + count);

		System.out.println(result.getDocument().asXML());

		System.out.println("******************************************");

		assertTrue(count > 0);

	}

	@Test

	public void testInfoPackage() throws ActionManagerException {

		Agent a = new Agent("test", "test", AGENT_TYPE.algo);

		core.applyInfoPackageAction("dmf2actions", SET, a, Operation.INSERT, dmfInfoPackage, Provenance.user_claim_search, "0.5", "acus");

	}

	@Test

	public void testRetrieveActions() throws ActionManagerException {

		core.getHbaseClient().retrieveRows(ACTION_TYPE.aac + "|", null, 10);

	}

}

package eu.dnetlib.actionmanager.xslt;

import java.io.StringReader;

import java.io.StringWriter;

import javax.xml.transform.Source;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.stream.StreamResult;

import javax.xml.transform.stream.StreamSource;

import org.junit.Before;

import org.junit.Test;

public class RelationXsltFunctionsTest {

	private Transformer transformer;

	@Before

	public void setUp() throws Exception {

		Source source = new StreamSource(getClass().getResourceAsStream("/eu/dnetlib/actionmanager/xslt/rels2actions.xslt"));

		transformer = TransformerFactory.newInstance().newTransformer(source);

	}

	@Test

	public void test_01() throws Exception {

		StringBuffer xml = new StringBuffer();

		xml.append("<RELATIONS>");

		xml.append("</RELATIONS>");

		applyXslt(xml.toString());

	}

	@Test

	public void test_02() throws Exception {

		StringBuffer xml = new StringBuffer();

		xml.append("<RELATIONS>");

		xml.append("<RELATION type='resultProject' source='50|crossref____::7fb616c8cdd08f0f12a04ac0f18edf80' target='corda_______::281877' />");

		xml.append("<RELATION type='resultProject' source='50|crossref____::7fb616c8cdd08f0f12a04ac0f18edf80' target='40|corda_______::0001fc1603869410927ddbde3afb2403' />");

		xml.append("<RELATION type='personResult' source='30|crossref____::000178abc5a314472603976a1be68927' target='50|crossref____::7fb616c8cdd08f0f12a04ac0f18edf80' />");

		xml.append("</RELATIONS>");

		applyXslt(xml.toString());

	}

	public String applyXslt(String xml) throws Exception {

		final Source xmlSource = new StreamSource(new StringReader(xml));

		final StringWriter output = new StringWriter();

		transformer.transform(xmlSource, new StreamResult(output));

		System.out.println(output.toString());

		return output.toString();

	}

}

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jaxws="http://cxf.apache.org/jaxws"

	xmlns:sec="http://cxf.apache.org/configuration/security" xmlns:wsa="http://cxf.apache.org/ws/addressing"

	xmlns:p="http://www.springframework.org/schema/p" xmlns:http="http://cxf.apache.org/transports/http/configuration"

	xmlns:t="http://dnetlib.eu/springbeans/t" xmlns:template="http://dnetlib.eu/springbeans/template"

	xmlns:util="http://www.springframework.org/schema/util"

	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd

                                    http://cxf.apache.org/ws/addressing http://cxf.apache.org/schemas/ws-addr-conf.xsd

                                    http://cxf.apache.org/configuration/security http://cxf.apache.org/schemas/configuration/security.xsd

                                    http://cxf.apache.org/transports/http/configuration http://cxf.apache.org/schemas/configuration/http-conf.xsd

                            http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd

                            http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-2.0.xsd

                            http://dnetlib.eu/springbeans/template http://dnetlib.eu/springbeans/template.xsd">

	<bean id="hbaseActionManagerCore" class="eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore"

		p:actionFactory-ref="actionFactory"

		p:hbaseClient-ref="hbaseClient" />

	<bean id="hbaseClient" class="eu.dnetlib.actionmanager.hbase.HBaseClient"

		p:tableName="db_actions_michele"

		p:hbaseConf="${services.data.hbase.dm.properties}" 

		init-method="initWithEmptyTable" />		

		<!-- p:hbaseConf="classpath:/eu/dnetlib/actionmanager/test/hbase.properties" -->

	<bean id="actionFactory" class="eu.dnetlib.actionmanager.actions.ActionFactory">

		<property name="xslts">

			<util:map>

				<entry key="dmf2actions" value="classpath:/eu/dnetlib/actionmanager/xslt/dmf2actions.xslt" />

			</util:map>

		</property>

	</bean>

</beans>

#COMMON

hadoop.entry.node                               =       node1.hadoop.research-infrastructures.eu

#CORE-SITE

fs.defaultFS                                    =       hdfs://${hadoop.entry.node}:8020

io.file.buffer.size                     =       65536

io.compression.codecs                   =       org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.BZip2Codec,org.apache.hadoop.io.compress.DeflateCodec,org.apache.hadoop.io.compress.SnappyCodec

hadoop.security.authentication  =       simple

hadoop.security.auth_to_local   =       DEFAULT

#HBASE-SITE

hbase.rootdir                                   =       ${fs.defaultFS}/hbase

hbase.client.write.buffer               =       2097152

hbase.client.pause                              =       1000

hbase.client.retries.number             =       10

hbase.client.scanner.caching    =       1

hbase.client.keyvalue.maxsize   =       10485760

hbase.security.authentication   =       simple

zookeeper.session.timeout               =       60000

zookeeper.znode.parent                  =       /hbase

zookeeper.znode.rootserver              =       root-region-server

hbase.zookeeper.quorum                  =       ${hadoop.entry.node}

hbase.zookeeper.property.clientPort     =       2182

#HDFS-SITE

dfs.https.port                                  =       50470

dfs.namenode.https-address              =       ${hadoop.entry.node}:${dfs.https.port}

dfs.namenode.http-address               =       ${hadoop.entry.node}:50070

dfs.replication                                 =       3

dfs.blocksize                                   =       134217728

dfs.client.use.datanode.hostname=       false

#MAPRED-SITE

mapreduce.jobtracker.address    =       ${hadoop.entry.node}:8021

mapred.job.tracker                              =       ${hadoop.entry.node}:8021

mapred.mapper.new-api                   =       true

mapred.reducer.new-api                  =       true

mapred.output.compress                  =       false

mapred.output.compression.type  =       BLOCK

mapred.output.compression.codec =       org.apache.hadoop.io.compress.DefaultCodec

#mapred.map.output.compression.codec    =       org.apache.hadoop.io.compress.SnappyCodec

mapred.compress.map.output              =       true

#io.sort.factor                                 =       64

#io.sort.record.percent                 =       0.05

#io.sort.spill.percent                  =       0.8

#mapred.reduce.parallel.copies  =       10

#mapred.submit.replication              =       1

#mapred.reduce.tasks                            =       8

#io.sort.mb                                             =       56

#mapred.child.java.opts                 =        -Xmx238449165

#mapred.job.reuse.jvm.num.tasks =       1

#mapred.map.tasks                               =       6

#mapred.map.tasks.speculative.execution =       false

#mapred.reduce.tasks.speculative.execution      =       false

#mapred.reduce.slowstart.completed.maps =       0.8

##mapred.local.dir                              = /data/mapred/local,/data/mapred/jt

mapred.local.dir                                = /tmp

mapred.child.java.opts                  = -Xmx512973352

#dfs.client.use.legacy.blockreader=true

package eu.dnetlib.actionmanager.hbase;

import java.io.IOException;

import java.util.Collections;

import java.util.Iterator;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.NavigableMap;

import java.util.Set;

import java.util.concurrent.Executor;

import java.util.concurrent.Executors;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.hbase.client.Delete;

import org.apache.hadoop.hbase.client.Get;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.filter.Filter;

import org.apache.hadoop.hbase.filter.FilterList;

import org.apache.hadoop.hbase.filter.PageFilter;

import org.apache.hadoop.hbase.filter.PrefixFilter;

import org.apache.hadoop.hbase.util.Bytes;

import org.springframework.beans.factory.annotation.Required;

import com.google.common.base.Function;

import com.google.common.collect.Iterators;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import com.google.gson.Gson;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.actionmanager.ActionManagerConstants;

import eu.dnetlib.actionmanager.ActionManagerConstants.ACTION_TYPE;

import eu.dnetlib.actionmanager.ActionManagerConstants.COLUMN_FAMILIES;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.data.hadoop.config.ClusterName;

import eu.dnetlib.data.hadoop.config.ConfigurationEnumerator;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.miscutils.functional.hash.Hashing;

public class HBaseClient {

	private String tableName;

	@Resource

	private ConfigurationEnumerator configurationEnumerator;

	private Configuration config;

	private static final Log log = LogFactory.getLog(HBaseClient.class); // NOPMD by marko on 11/24/08 5:02 PM

	private final Executor executor = Executors.newSingleThreadExecutor();

	private final Function<Result, Map<String, String>> transformRow = new Function<Result, Map<String, String>>() {

		@Override

		public Map<String, String> apply(Result r) {

			Map<String, String> map = Maps.newHashMap();

			map.put("rowId", Bytes.toString(r.getRow()));

			for (Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> e1 : r.getMap().entrySet()) {

				String cf = Bytes.toString(e1.getKey());

				for (Entry<byte[], NavigableMap<Long, byte[]>> e2 : e1.getValue().entrySet()) {

					byte[] bb = e2.getValue().get(Collections.max(e2.getValue().keySet()));

					String key = cf + ":" + Bytes.toString(e2.getKey());

					String value = "";

					try {

						value = key.equals("target:content") ? Oaf.parseFrom(bb).toString() : Bytes.toString(bb);

					} catch (InvalidProtocolBufferException e) {

						log.error("Problem parsing protobuf !!!");

					}

					map.put(key, value);

				}

			}

			return map;

		}

	};

	public void init() {

		performInit(false);

	}

	public void initWithEmptyTable() {

		performInit(true);

	}

	private void performInit(final boolean delete) {

		executor.execute(new Runnable() {

			@Override

			public void run() {

				try {

					log.info("Initializing Action Manager");

					loadConfiguration();

					prepareTable(delete);

					log.info("ActionManager is ready");

				} catch (IOException e) {

					log.error("Error initializing action manager", e);

				}

			}

		});

	}

	private void prepareTable(boolean delete) throws IOException {

		final HBaseAdmin admin = new HBaseAdmin(config);

		try {

			if (delete && admin.tableExists(tableName)) {

				log.info("Deleting existing hbase table: " + tableName);

				admin.disableTable(tableName);

				admin.deleteTable(tableName);

			}

			if (!admin.tableExists(tableName)) {

				log.info("Creating missing hbase table: " + tableName);

				admin.createTable(new HTableDescriptor(tableName));

			}

			final HTableDescriptor desc = admin.getTableDescriptor(Bytes.toBytes(getTableName()));

			Set<String> currents = Sets.newHashSet();

			for (HColumnDescriptor hcd : desc.getColumnFamilies()) {

				currents.add(hcd.getNameAsString());

			}

			Set<String> missing = Sets.newHashSet();

			for (COLUMN_FAMILIES cf : ActionManagerConstants.COLUMN_FAMILIES.values()) {

				if (!currents.contains(cf.toString())) {

					missing.add(cf.toString());

				}

			}

			if (!missing.isEmpty()) {

				if (admin.isTableEnabled(getTableName())) {

					admin.disableTable(getTableName());

				}

				for (String column : missing) {

					log.info("hbase table: '" + getTableName() + "', adding columnFamily: " + column);

					admin.addColumn(getTableName(), new HColumnDescriptor(column));

				}

				admin.enableTable(getTableName());

			}

		} finally {

			admin.close();

		}

	}

	private void loadConfiguration() throws IOException {

		config = configurationEnumerator.get(ClusterName.DM);

		log.info(config.toString());

	}

	public Configuration getConfig() {

		return config;

	}

	public void write(List<Put> puts) throws ActionManagerException {

		try {

			final HTable table = new HTable(config, tableName);

			try {

				log.info("Adding " + puts.size() + " action(s) to " + tableName);

				table.put(puts);

			} finally {

				table.flushCommits();

				table.close();

			}

		} catch (IOException e) {

			throw new ActionManagerException(e);

		}

	}

	public void write(List<Put> puts, HTable table) throws ActionManagerException {

		try {

			log.info("Adding " + puts.size() + " action(s) to " + tableName);

			table.put(puts);

		} catch (IOException e) {

			throw new ActionManagerException(e);

		}

	}

	public void delete(List<Delete> deletes) throws ActionManagerException {

		try {

			final HTable table = new HTable(config, tableName);

			try {

				log.info("Deleting " + deletes.size() + " action(s) from " + tableName);

				table.delete(deletes);

			} finally {

				table.flushCommits();

				table.close();

			}

		} catch (IOException e) {

			throw new ActionManagerException(e);

		}

	}

	public Map<String, String> getRow(byte[] id) throws ActionManagerException {

		try {

			final HTable table = new HTable(config, tableName);

			try {

				Result result = table.get(new Get(id));

				return transformRow.apply(result);

			} finally {

				table.close();

			}

		} catch (IOException e) {

			throw new ActionManagerException(e);

		}

	}

	public List<Map<String, String>> retrieveRows(String prefix, String start, int limit) throws ActionManagerException {

		List<Map<String, String>> list = Lists.newArrayList();

		try {

			final HTable table = new HTable(config, tableName);

			try {

				Filter filter = prefix == null || prefix.isEmpty() ? new PageFilter(limit) : new FilterList(FilterList.Operator.MUST_PASS_ALL, new PageFilter(

						limit), new PrefixFilter(Bytes.toBytes(prefix)));

				Scan scan = new Scan();

				if (start != null) {

					scan.setStartRow(Bytes.toBytes(start));

				}

				scan.setFilter(filter);

				ResultScanner rs = table.getScanner(scan);

				try {

					Iterator<Map<String, String>> iter = Iterators.transform(rs.iterator(), transformRow);

					while (iter.hasNext()) {

						list.add(iter.next());

					}

				} finally {

					rs.close();

				}

			} finally {

				table.close();

			}

		} catch (IOException e) {

			throw new ActionManagerException(e);

		}

		return list;

	}

	public List<String> retrieveActionsByAgent(String agentId, int limit) throws ActionManagerException {

		List<String> list = Lists.newArrayList();

		String prefix = ACTION_TYPE.pkg + "|" + Hashing.md5(agentId) + "|";

		try {

			final HTable table = new HTable(config, tableName);

			try {

				FilterList filter = new FilterList(FilterList.Operator.MUST_PASS_ALL);

				filter.addFilter(new PageFilter(limit));

				filter.addFilter(new PrefixFilter(Bytes.toBytes(prefix)));

				Scan scan = new Scan();

				scan.setFilter(filter);

				ResultScanner rs = table.getScanner(scan);

				Iterator<Result> iter = rs.iterator();

				while (iter.hasNext()) {

					Result r = iter.next();

					Map<String, String> map = Maps.newHashMap();

					map.put("id", Bytes.toString(r.getRow()));

					map.put("content", Bytes.toString(r.getValue(ActionManagerConstants.ACTION_COLFAMILY, Bytes.toBytes(ACTION_TYPE.pkg.toString()))));

					list.add(new Gson().toJson(map));

				}

			} finally {

				table.close();

			}

		} catch (IOException e) {

			throw new ActionManagerException(e);

		}

		return list;

	}

	public String getTableName() {

		return tableName;

	}

	@Required

	public void setTableName(String tableName) {

		this.tableName = tableName;

	}

}

package eu.dnetlib.actionmanager.hbase;

import java.util.List;

import java.util.Map;

import org.apache.hadoop.hbase.client.Delete;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.util.Bytes;

import org.springframework.beans.factory.annotation.Required;

import com.google.common.collect.Lists;

import eu.dnetlib.actionmanager.actions.ActionFactory;

import eu.dnetlib.actionmanager.actions.XsltInfoPackageAction;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.is.ISClient;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.actionmanager.set.ActionManagerSet;

public class HBaseActionManagerCore {

	private ActionFactory actionFactory;

	private HBaseClient hbaseClient;

	private ISClient informationServiceClient;

	/*

	 * public void insertOrUpdateAtomicAction(AtomicAction action, String provenance, String trust) throws

	 * ActionManagerException { hbaseClient.write(action.asPutOperations(null, provenance, trust, null)); }

	 * 

	 * public void insertOrUpdateAtomicActions(List<AtomicAction> actions, String provenance, String trust) throws

	 * ActionManagerException { for (AtomicAction action : actions) { insertOrUpdateAtomicAction(action, provenance,

	 * trust); } }

	 */

	public int applyInfoPackageAction(

			String splitRuleId,

			String set,

			Agent agent,

			Operation operation,

			String infoPackage,

			Provenance provenance,

			String trust,

			String nsprefix) throws ActionManagerException {

		if (informationServiceClient.existsSet(set)) {

			XsltInfoPackageAction action = actionFactory.generateInfoPackageAction(splitRuleId, set, agent, operation, infoPackage);

			List<Put> puts = action.asPutOperations(null, provenance, trust, nsprefix);

			hbaseClient.write(puts);

			return puts.size();

		} else {

			throw new ActionManagerException("Missing set " + set);

		}

	}

	public int applyInfoPackageAction(			

			String splitRuleId,

			String set,

			Agent agent,

			Operation operation,

			String infoPackage,

			Provenance provenance,

			String trust,

			String nsprefix,

			HTable table) throws ActionManagerException {

		if (informationServiceClient.existsSet(set)) {

			XsltInfoPackageAction action = actionFactory.generateInfoPackageAction(splitRuleId, set, agent, operation, infoPackage);

			List<Put> puts = action.asPutOperations(null, provenance, trust, nsprefix);

			hbaseClient.write(puts, table);

			return puts.size();

		} else {

			throw new ActionManagerException("Missing set " + set);

		}

	}

	public void deleteInfoPackageAction(String id) throws ActionManagerException {

		Map<String, String> row = hbaseClient.getRow(Bytes.toBytes(id));

		List<Delete> deletes = Lists.newArrayList();

		for (String key : row.keySet()) {

			if (key.startsWith("rel:")) {

				deletes.add(new Delete(Bytes.toBytes(key.substring(4))));

			}

		}

		deletes.add(new Delete(Bytes.toBytes(id)));

		hbaseClient.delete(deletes);

	}

	public ActionFactory getActionFactory() {

		return actionFactory;

	}

	@Required

	public void setActionFactory(ActionFactory actionFactory) {

		this.actionFactory = actionFactory;

	}

	public HBaseClient getHbaseClient() {

		return hbaseClient;

	}

	@Required

	public void setHbaseClient(HBaseClient hbaseClient) {

		this.hbaseClient = hbaseClient;

	}

	public List<ActionManagerSet> listAvailableSets() {

		return informationServiceClient.listSets();

	}

	public ISClient getInformationServiceClient() {

		return informationServiceClient;

	}

	@Required

	public void setInformationServiceClient(ISClient informationServiceClient) {

		this.informationServiceClient = informationServiceClient;

	}

}

package eu.dnetlib.actionmanager.blackboard;

public enum ActionManagerActions {

	FEED,

	PROMOTE,

	DELETE

}

package eu.dnetlib.actionmanager.blackboard;

import java.util.concurrent.Callable;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.actionmanager.wf.ActionManagerWorkflowLauncher;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

public class DeleteActionManagerAction extends AbstractActionManagerAction implements BlackboardServerAction<ActionManagerActions> {

	//	<MESSAGE id="bb-1234" date="2012-11-06T15:10:56+01:00">

	//		<ACTION>DELETE</ACTION>

	//		<PARAMETER name="set" value=""/>

	//		<ACTION_STATUS>ASSIGNED</ACTION_STATUS>

	//	</MESSAGE>

	@Override

	public void execute(final BlackboardServerHandler handler, final BlackboardJob job) {

		String set = job.getParameters().get("set");

		if (set == null || set.isEmpty()) {

			set = ActionManagerWorkflowLauncher.ALL_SETS;

		}

		try {

			getActionManagerWorkflowLauncher().executeDelete(set, new Callable<String>() {

				@Override

				public String call() {

					handler.done(job);

					return "done";

				}

			}, new Callable<String>() {

				@Override

				public String call() {

					handler.failed(job, new ActionManagerException("MSRO Workwlow is failed"));

					return "failed";

				}

			});

		} catch (Exception e) {

			handler.failed(job, e);

		}

	}

}

package eu.dnetlib.actionmanager.blackboard;

import javax.annotation.Resource;

import eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore;

import eu.dnetlib.actionmanager.wf.ActionManagerWorkflowLauncher;

public class AbstractActionManagerAction {

	@Resource

	protected HBaseActionManagerCore core;

	@Resource

	private ActionManagerWorkflowLauncher actionManagerWorkflowLauncher;

	public HBaseActionManagerCore getCore() {

		return core;

	}

	public void setCore(HBaseActionManagerCore core) {

		this.core = core;

	}

	public ActionManagerWorkflowLauncher getActionManagerWorkflowLauncher() {

		return actionManagerWorkflowLauncher;

	}

	public void setActionManagerWorkflowLauncher(ActionManagerWorkflowLauncher actionManagerWorkflowLauncher) {

		this.actionManagerWorkflowLauncher = actionManagerWorkflowLauncher;

	}

}

package eu.dnetlib.actionmanager.blackboard;

import javax.annotation.Resource;

import org.apache.commons.codec.binary.Base64;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

// TODO: Auto-generated Javadoc

/**

 * The Class FeedActionManagerAction.

 */

public class FeedActionManagerAction extends AbstractActionManagerAction implements BlackboardServerAction<ActionManagerActions> {

	/** The Constant log. */

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

	/** The result set client factory. */

	@Resource

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * input parameters: epr: result-set end point reference containg the set of package info ruleId: one of the rules

	 * available to apply set: the destination set for the current info-packages provenance: the source of the

	 * info-packages fed trust: trust level nsprefix: namespace prefix agentID agentType.

	 * 

	 * @param handler

	 *            the handler

	 * @param job

	 *            the job

	 */

	@Override

	public void execute(BlackboardServerHandler handler, BlackboardJob job) {

		log.debug("Inside execute" + this.getClass().getSimpleName());

		String epr = job.getParameters().get("epr");

		String ruleId = job.getParameters().get("ruleId");

		String set = job.getParameters().get("set");

		Provenance provenance = Provenance.convert(job.getParameters().get("provenance"));

		String trust = job.getParameters().get("trust");

		String nsprefix = job.getParameters().get("nsprefix");

		String agentID = job.getParameters().get("agentId");

		String agentName = job.getParameters().get("agentName");

		String agentType = job.getParameters().get("agentType");

		int n = 0;

		try {

			n = feed(epr, ruleId, set, provenance, trust, nsprefix, agentID, agentName, agentType);

		} catch (ActionManagerException e) {

			handler.failed(job, e);

			return;

		}

		log.debug("Actions added: " + n);

		handler.done(job);

	}

	/**

	 * Feed.

	 * 

	 * @param epr

	 *            the epr

	 * @param ruleId

	 *            the rule id

	 * @param set

	 *            the set

	 * @param provenance

	 *            the provenance

	 * @param trust

	 *            the trust

	 * @param nsprefix

	 *            the nsprefix

	 * @param agentID

	 *            the agent id

	 * @param agentType

	 *            the agent type

	 * @return the int

	 * @throws ActionManagerException

	 *             the action manager exception

	 */

	private int feed(

			String epr,

			String ruleId,

			String set,

			Provenance provenance,

			String trust,

			String nsprefix,

			String agentID,

			String agentName,

			String agentType

	) throws ActionManagerException {

		String decodeEPR = new String(Base64.decodeBase64(epr));

		final Iterable<String> records = resultSetClientFactory.getClient(decodeEPR);

		int n = 0;

		Agent agent = new Agent();

		agent.setId(agentID);

		agent.setName(agentName);

		agent.setType(Agent.AGENT_TYPE.valueOf(agentType));

		for (String infoPackage : records) {

			if (infoPackage != null) {

				n += core.applyInfoPackageAction(ruleId, set, agent, Operation.INSERT, infoPackage, provenance, trust, nsprefix);

			}

		}

		return n;

	}

}

package eu.dnetlib.actionmanager.blackboard;

import java.util.concurrent.Callable;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.actionmanager.wf.ActionManagerWorkflowLauncher;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

public class PromoteActionManagerAction extends AbstractActionManagerAction implements BlackboardServerAction<ActionManagerActions> {

	@Override

	public void execute(final BlackboardServerHandler handler, final BlackboardJob job) {

		String set = job.getParameters().get("set");

		if (set == null || set.isEmpty()) {

			set = ActionManagerWorkflowLauncher.ALL_SETS;

		}

		try {

			getActionManagerWorkflowLauncher().executeCommit(set, new Callable<String>() {

				@Override

				public String call() {

					handler.done(job);

					return "done";

				}

			}, new Callable<String>() {

				@Override

				public String call() {

					handler.failed(job, new ActionManagerException("MSRO Workwlow is failed"));

					return "failed";

				}

			});

		} catch (Exception e) {

			handler.failed(job, e);

		}

	}

}

package eu.dnetlib.actionmanager.hack;

import java.io.IOException;

import java.io.StringReader;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.client.HTable;

import org.dom4j.Document;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Required;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Agent.AGENT_TYPE;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore;

import eu.dnetlib.enabling.resultset.WorkflowCountingResultSetFactory;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import eu.dnetlib.workflow.AsyncJobNode;

public class StoreInfoPackageJobNode extends AsyncJobNode {

	private static final Log log = LogFactory.getLog(StoreInfoPackageJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private ResultSetClientFactory resultSetClientFactory;

	private HBaseActionManagerCore core;

	private String splitRuleId = "claimedMigration";

	private String defaultTrust;

	private WorkflowCountingResultSetFactory countingRSFactory;

	@Override

	protected void executeAsync(final Engine engine, final NodeToken token) {

		final SAXReader reader = new SAXReader();

		try {

			final HTable table = getTable();

			try {

				W3CEndpointReference epr = (W3CEndpointReference) token.getEnv().getTransientAttribute("epr");

				for (String record : resultSetClientFactory.getClient(getCountingRSFactory().createCountingResultSet(epr, token))) {

					// System.out.println("\n\nRECORD:\n\n" + record);

					Document doc = reader.read(new StringReader(record));

					String id = doc.valueOf("//*[local-name()='objIdentifier']");

					String user = doc.valueOf("//*[local-name()='user']");

					String nsprefix = doc.valueOf("//*[local-name()='datasourceprefix']");

					RequiredSet set = null;

					Provenance provenance = null;

					Operation operation = Operation.INSERT;

					Agent agent = null;

					if ((user != null) && !user.trim().isEmpty()) {

						agent = new Agent(user.trim(), user.trim(), AGENT_TYPE.human);

						if (id.startsWith("crossref____::")) {

							set = RequiredSet.userclaim_search_crossref;

							provenance = Provenance.user_claim_pid;

						} else if (id.startsWith("driver______::")) {

							set = RequiredSet.userclaim_search_driver;

							provenance = Provenance.user_claim_search;

						} else {

							log.warn("I can't assign set and provenance for id " + id);

							log.warn(record);

						}

					} else {

						log.warn("********* Missing user for record: " + id);

					}

					core.applyInfoPackageAction(splitRuleId, set.toString(), agent, operation, record, provenance, defaultTrust, nsprefix, table);

				}

				engine.complete(token, Arc.DEFAULT_ARC);

			} catch (Exception e) {

				log.error("Error storing EPR");

				failed(engine, token, e);

			} finally {

				table.flushCommits();

				table.close();

			}

		} catch (Throwable e) {

			log.error("Error storing EPR");

			failed(engine, token, e);

		}

	}

	private HTable getTable() throws IOException {

		return new HTable(core.getHbaseClient().getConfig(), core.getHbaseClient().getTableName());

	}

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	@Required

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	public HBaseActionManagerCore getCore() {

		return core;

	}

	@Required

	public void setCore(final HBaseActionManagerCore core) {

		this.core = core;

	}

	public String getSplitRuleId() {

		return splitRuleId;

	}

	@Required

	public void setSplitRuleId(final String splitRuleId) {

		this.splitRuleId = splitRuleId;

	}

	public String getDefaultTrust() {

		return defaultTrust;

	}

	@Required

	public void setDefaultTrust(final String defaultTrust) {

		this.defaultTrust = defaultTrust;

	}

	public WorkflowCountingResultSetFactory getCountingRSFactory() {

		return countingRSFactory;

	}

	@Required

	public void setCountingRSFactory(final WorkflowCountingResultSetFactory countingRSFactory) {

		this.countingRSFactory = countingRSFactory;

	}

}

package eu.dnetlib.actionmanager.hack;

import java.io.StringWriter;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Required;

import org.springframework.core.io.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.database.rmi.DatabaseService;

import eu.dnetlib.enabling.tools.ServiceLocator;

import eu.dnetlib.workflow.AsyncJobNode;

public class PerformSQLJobNode extends AsyncJobNode {

	private ServiceLocator<DatabaseService> dbServiceLocator;

	private String dbName;

	private Resource query;

	private Resource countQuery;

	private Resource xslt;

	private static final Log log = LogFactory.getLog(PerformSQLJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Override

	public void executeAsync(final Engine engine, final NodeToken token) {

		final StringWriter q = new StringWriter();

		final StringWriter cq = new StringWriter();

		final StringWriter xs = new StringWriter();

		try {

			IOUtils.copy(query.getInputStream(), q);

			IOUtils.copy(countQuery.getInputStream(), cq);

			IOUtils.copy(xslt.getInputStream(), xs);

			log.info("Executing query [" + q + "] on db " + dbName);

			W3CEndpointReference epr = getDbServiceLocator().getService().alternativeXsltSearchSQL(dbName, q.toString(), cq.toString(), xs.toString());

			log.info("Optained epr: " + epr);

			token.getEnv().setTransientAttribute("epr", epr);

			engine.complete(token, Arc.DEFAULT_ARC);

		} catch (Throwable e) {

			failed(engine, token, e);

		}

	}

	public String getDbName() {

		return dbName;

	}

	@Required

	public void setDbName(String dbName) {

		this.dbName = dbName;

	}

	public ServiceLocator<DatabaseService> getDbServiceLocator() {

		return dbServiceLocator;

	}

	@Required

	public void setDbServiceLocator(ServiceLocator<DatabaseService> dbServiceLocator) {

		this.dbServiceLocator = dbServiceLocator;

	}

	public Resource getQuery() {

		return query;

	}

	@Required

	public void setQuery(Resource query) {

		this.query = query;

	}

	public Resource getCountQuery() {

		return countQuery;

	}

	@Required

	public void setCountQuery(Resource countQuery) {

		this.countQuery = countQuery;

	}

	public Resource getXslt() {

		return xslt;

	}

	@Required

	public void setXslt(Resource xslt) {

		this.xslt = xslt;

	}

}

package eu.dnetlib.actionmanager.hack;

public enum RequiredSet {

	userclaim_search_driver,

	userclaim_search_crossref,

	sysimport_mining_repository_pubmed,

	sysimport_mining_repository_arxiv,

	sysimport_mining_repository_isiWOS,

	sysimport_mining_repository_copernicus,

	sysimport_mining_repository_unknown

}

package eu.dnetlib.actionmanager.hack;

import org.springframework.beans.factory.annotation.Required;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore;

import eu.dnetlib.actionmanager.set.ActionManagerSet;

import eu.dnetlib.workflow.AsyncJobNode;

public class RegisterMissingSets extends AsyncJobNode {

	private HBaseActionManagerCore core;

	@Override

	protected void executeAsync(Engine engine, NodeToken token) {

		try {

			for (RequiredSet rs : RequiredSet.values()) {

				String set = rs.toString();

				if (!core.getInformationServiceClient().existsSet(set.toString())) {

					core.getInformationServiceClient().registerSetProfile(new ActionManagerSet(set, set));

				}

			}

			engine.complete(token, Arc.DEFAULT_ARC);

		} catch (Throwable e) {

			failed(engine, token, e);

		}

	}

	public HBaseActionManagerCore getCore() {

		return core;

	}

	@Required

	public void setCore(HBaseActionManagerCore core) {

		this.core = core;

	}

}

package eu.dnetlib.actionmanager.inspector;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import org.springframework.stereotype.Controller;

import org.springframework.ui.ModelMap;

import org.springframework.web.bind.annotation.ModelAttribute;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestParam;

import com.google.common.base.Function;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import eu.dnetlib.actionmanager.ActionManagerConstants.ACTION_TYPE;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Agent.AGENT_TYPE;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore;

import eu.dnetlib.actionmanager.set.ActionManagerSet;

import eu.dnetlib.actionmanager.wf.ActionManagerWorkflowLauncher;

import eu.dnetlib.enabling.inspector.AbstractInspectorController;

@Controller

public class ActionManagerController extends AbstractInspectorController {

	private static final int DEFAULT_PAGE_SIZE = 10;

	@Resource

	private HBaseActionManagerCore core;

	@Resource

	private ActionManagerWorkflowLauncher actionManagerWorkflowLauncher;

	@ModelAttribute(value = "types")

	public ACTION_TYPE[] listActionTypes() {

		return ACTION_TYPE.values();

	}

	@ModelAttribute(value = "provenances")

	public Provenance[] listProvenances() {

		return Provenance.values();

	}

	@ModelAttribute(value = "sets")

	public List<ActionManagerSet> listSets() {

		return core.listAvailableSets();

	}

	@RequestMapping("/inspector/actions/listActions.do")

	public String listActions(

			final ModelMap map,

			@RequestParam(value = "type", required = false) String stype,

			@RequestParam(value = "start", required = false) String start) throws Exception {

		ACTION_TYPE type = (stype == null) ? ACTION_TYPE.all : ACTION_TYPE.valueOf(stype);

		String prefix = (type == ACTION_TYPE.all) ? null : type + "|";

		map.addAttribute("type", type);

		List<Map<String, String>> actions = core.getHbaseClient().retrieveRows(prefix, start, DEFAULT_PAGE_SIZE + 1);

		int size = actions.size();

		if (size > DEFAULT_PAGE_SIZE) {

			map.addAttribute("next", actions.remove(size - 1).get("rowId"));

		}

		map.addAttribute("actions", Lists.transform(actions, new Function<Map<String, String>, Map<String, Object>>() {

			@Override

			public Map<String, Object> apply(Map<String, String> map) {

				return formatAction(map);

			}

		}));

		String template;

		switch (type) {

		case aac:

			template = "listAtomicActions";

			break;

		case pkg:

			template = "listInfoPackages";

			break;

		default:

			template = "listAll";

			break;

		}

		return "inspector/actions/" + template;

	}

	@SuppressWarnings("unchecked")

	private Map<String, Object> formatAction(Map<String, String> map) {

		Map<String, Object> res = Maps.newHashMap();

		for (String key : map.keySet()) {

			if (key.equals("rowId")) {

				res.put("rowId", map.get(key));

			} else if (key.startsWith("date:")) {

				res.put("date", map.get(key));

			} else if (key.startsWith("set:")) {

				res.put("set", map.get(key));

			} else if (key.startsWith("operation:")) {

				res.put("operation", map.get(key));

			} else if (key.endsWith(":" + ACTION_TYPE.pkg)) {

				res.put("infoPackage", map.get(key).replaceAll("<", "&lt;").replaceAll(">", "&gt;"));

			} else if (key.startsWith("rel:")) {

				String nk = map.get(key).trim();

				if (!res.containsKey(nk)) {

					res.put(nk, new ArrayList<String>());

				}

				((List<String>) res.get(nk)).add(key.substring(4));

			} else if (key.startsWith("target:")) {

				res.put(key.replaceAll(":", "_"), map.get(key));

			} else {

				res.put(key, map.get(key));

			}

		}

		return res;

	}

	@RequestMapping("/inspector/actions/registerInfoPackage.do")

	public void registerInfoPackage(

			final ModelMap map,

			@RequestParam(value = "ruleId", required = false) String ruleId,

			@RequestParam(value = "set", required = false) String set,

			@RequestParam(value = "infoPackage", required = false) String infoPackage,

			@RequestParam(value = "provenance", required = false) String provenance,

			@RequestParam(value = "trust", required = false) String trust,

			@RequestParam(value = "nsprefix", required = false) String nsprefix) throws Exception {

		if (ruleId != null && !ruleId.isEmpty() && set != null && !set.isEmpty() && infoPackage != null && !infoPackage.isEmpty() && trust != null

				&& !trust.isEmpty() && provenance != null && !provenance.isEmpty() && nsprefix != null && !nsprefix.isEmpty()

				&& core.getActionFactory().getXslts().containsKey(ruleId)) {

			Agent agent = new Agent();

			agent.setId("inspector");

			agent.setName("inspector");

			agent.setType(AGENT_TYPE.human);

			int n = core.applyInfoPackageAction(ruleId, set, agent, Operation.INSERT, infoPackage, Provenance.convert(provenance), trust, nsprefix);

			map.addAttribute("message", "1 infoPackageAction + " + (n - 1) + " atomicAction(s) registered");

		}

		map.addAttribute("rules", core.getActionFactory().getXslts().keySet());

		map.addAttribute("ruleId", ruleId);

		map.addAttribute("set", set);

		map.addAttribute("infoPackage", infoPackage);

		map.addAttribute("trust", trust);

		map.addAttribute("nsprefix", nsprefix);

	}

	@RequestMapping("/inspector/actions/deleteInfoPackage.do")

	public String deleteInfoPackage(final ModelMap map, @RequestParam(value = "id", required = false) String id) throws Exception {

		core.deleteInfoPackageAction(id);

		return "redirect:listActions.do?type=" + ACTION_TYPE.pkg;

	}

	@RequestMapping("/inspector/actions/bulkdelete.do")

	public void clear(final ModelMap map, @RequestParam(value = "set", required = false) String set) throws Exception {

		map.addAttribute("allSets", ActionManagerWorkflowLauncher.ALL_SETS);

		if (set != null && !set.isEmpty()) {

			actionManagerWorkflowLauncher.executeDelete(set, null, null);

			map.addAttribute("message", "Workflow started !!!");

		}

	}

	@RequestMapping("/inspector/actions/commit.do")

	public void commit(final ModelMap map, @RequestParam(value = "set", required = false) String set) throws Exception {

		map.addAttribute("allSets", ActionManagerWorkflowLauncher.ALL_SETS);

		if (set != null && !set.isEmpty()) {

			actionManagerWorkflowLauncher.executeCommit(set, null, null);

			map.addAttribute("message", "Workflow started !!!");

		}

	}

	@RequestMapping("/inspector/actions/createSet.do")

	public void createSet(

			final ModelMap map,

			@RequestParam(value = "id", required = false) String id,

			@RequestParam(value = "name", required = false) String name) throws Exception {

		if (id != null && name != null && !id.isEmpty() && !name.isEmpty()) {

			core.getInformationServiceClient().registerSetProfile(new ActionManagerSet(id, name));

		}

		map.addAttribute("sets", core.listAvailableSets());

	}

}

package eu.dnetlib.actionmanager;

import java.util.List;

import javax.jws.WebParam;

import org.springframework.beans.factory.annotation.Required;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.actionmanager.common.Operation;

import eu.dnetlib.actionmanager.common.Provenance;

import eu.dnetlib.actionmanager.hbase.HBaseActionManagerCore;

import eu.dnetlib.actionmanager.rmi.ActionManagerException;

import eu.dnetlib.actionmanager.rmi.ActionManagerService;

import eu.dnetlib.actionmanager.set.ActionManagerSet;

import eu.dnetlib.enabling.tools.AbstractBaseService;

import eu.dnetlib.enabling.tools.blackboard.NotificationHandler;

public class ActionManagerServiceImpl extends AbstractBaseService implements ActionManagerService {

	private HBaseActionManagerCore core;