gCubeSystem
/
is-collector
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

git-svn-id: http://svn.research-infrastructures.eu/public/d4science/gcube/trunk/information-system/gCubeIS/Collector@14799 82a268e6-3cf1-43bd-a215-b396298e98cf

This commit is contained in:
Manuele Simi 2009-08-27 17:10:46 +00:00
parent ce01864bf9
commit 4a5f22ff7c
9 changed files with 277 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/org/gcube/informationsystem/collector/impl/porttypes/XMLStorageAccess.java
View File

@ -10,7 +10,6 @@ import org.gcube.informationsystem.collector.stubs.BackupNotAvailableFaultType;
import org.gcube.informationsystem.collector.stubs.ShutdownFailedFaultType;
import org.gcube.informationsystem.collector.stubs.XMLStorageNotAvailableFaultType;
import org.gcube.common.core.contexts.GCUBEServiceContext;
import org.gcube.common.core.contexts.GCUBEServiceContext.Status;
import org.gcube.common.core.porttypes.GCUBEPortType;
import org.gcube.common.core.types.VOID;
import org.gcube.informationsystem.collector.impl.contexts.ICServiceContext;
@ -102,7 +101,7 @@ public class XMLStorageAccess extends GCUBEPortType {
try {
State.dispose();
ICServiceContext.getContext().setStatus(Status.DOWN);
//ICServiceContext.getContext().setStatus(Status.DOWN);
} catch (Exception e) {
logger.error("Shutdown failed", e);
ShutdownFailedFaultType fault = new ShutdownFailedFaultType();
@ -124,15 +123,13 @@ public class XMLStorageAccess extends GCUBEPortType {
logger.info("Connect operation invoked");
try {
State.initialize();
ICServiceContext.getContext().setStatus(Status.READIED);
//ICServiceContext.getContext().setStatus(Status.READIED);
} catch (Exception e) {
logger.error("Initialisation failed", e);
XMLStorageNotAvailableFaultType fault = new XMLStorageNotAvailableFaultType();
fault.addFaultDetailString("Initialisation failed," + e.getMessage());
throw fault;
}
return new VOID();
}
}

5
src/org/gcube/informationsystem/collector/impl/porttypes/XQueryAccess.java
View File

@ -9,6 +9,7 @@ import org.gcube.common.core.utils.logging.GCUBELog;
import org.gcube.informationsystem.collector.impl.contexts.ICServiceContext;
import org.gcube.informationsystem.collector.impl.xmlstorage.exist.State;
import org.gcube.informationsystem.collector.impl.xmlstorage.exist.XQuery;
import org.gcube.informationsystem.collector.impl.xmlstorage.exist.XMLStorageManager.XMLStorageNotAvailableException;
import org.gcube.informationsystem.collector.stubs.XQueryExecuteRequest;
import org.gcube.informationsystem.collector.stubs.XQueryExecuteResponse;
import org.gcube.informationsystem.collector.stubs.XQueryFaultType;
@ -50,6 +51,10 @@ public class XQueryAccess extends GCUBEPortType {
response.setDataset(buildDataSet(result));
} catch (XMLStorageNotAvailableException e) {
XQueryFaultType fault = new XQueryFaultType();
fault.addFaultDetailString("XMLStorage is not currently available for XQuery execution");
throw fault;
} catch (Exception e) {
XQueryFaultType fault = new XQueryFaultType();
fault.addFaultDetailString("Exception when executing the requested XQuery");

71
src/org/gcube/informationsystem/collector/impl/xmlstorage/exist/QueryManager.java
View File

@ -1,5 +1,11 @@
package org.gcube.informationsystem.collector.impl.xmlstorage.exist;
import org.xmldb.api.base.Collection;
import org.xmldb.api.base.ResourceSet;
import org.xmldb.api.base.XMLDBException;
import org.gcube.common.core.utils.logging.GCUBELog;
/**
* Query Manager
*
@ -8,4 +14,69 @@ package org.gcube.informationsystem.collector.impl.xmlstorage.exist;
*/
public class QueryManager extends XMLStorageManager {
private final static GCUBELog logger = new GCUBELog(QueryManager.class);
private final int MAXQUERYATTEMPTS = 3;
/**
* Executes the given XQuery on the current collection or on the root collection if any was
* previously loaded
*
* @param query the XQuery to run
* @return a formatted resultset
* @throws XMLStorageNotAvailableException
*/
public ResourceSet executeXQuery(XQuery query) throws XMLStorageNotAvailableException {
if (this.getStatus() != STATUS.INITIALISED)
throw new XMLStorageNotAvailableException("XMLStorage is not available");
boolean retry = true;
int attempts = 0;
ResourceSet result = null;
Collection currentCollection = null;
currentCollection = this.loadAllCollections();
while ((retry) && (attempts < MAXQUERYATTEMPTS)) {
try {
// wait until the DB is unlocked
while (State.getDataManager().isLocked())
Thread.sleep(1000);
// execute query and get results in ResourceSet
if (currentCollection == null)
result = query.execute(this.rootCollection);
else
result = query.execute(currentCollection);
retry = false;
} catch (XMLDBException edb) {
logger.error("Failed to execute XQuery " + query.toString());
logger.error("Error details: " + edb.errorCode + " " + edb.getMessage(), edb);
// if the cause is a NullPointer, this can be due to a temporary
// lock on the database instance
if (edb.getCause() instanceof java.lang.NullPointerException) {
retry = true;
attempts++;
logger.warn("Trying a new attempt for query execution");
} else
retry = false;
} catch (Exception e) {
logger.error("", e);
// if the cause is a NullPointer, this can be due to a temporary
// lock on the database instance
if (e instanceof java.lang.NullPointerException) {
retry = true;
attempts++;
logger.warn("Trying a new attempt for query execution");
} else
retry = false;
}
}
this.resetCollection(currentCollection);
return result;
}
}

42
src/org/gcube/informationsystem/collector/impl/xmlstorage/exist/State.java
View File

@ -3,6 +3,7 @@ package org.gcube.informationsystem.collector.impl.xmlstorage.exist;
import org.gcube.common.core.utils.logging.GCUBELog;
import org.gcube.informationsystem.collector.impl.contexts.ICServiceContext;
import org.gcube.informationsystem.collector.impl.persistence.AggregatorPersistentResource;
import org.gcube.informationsystem.collector.impl.xmlstorage.exist.XMLStorageManager.STATUS;
import java.util.ArrayList;
import java.util.Enumeration;
@ -28,10 +29,10 @@ public class State {
private static QueryManager queryManager;
/** Thread that periodically sweeps the XMLStorage from expired resources */
public static Thread sweeperT;
public static Thread sweeperT = null;
/** Thread that periodically backups the XMLStorage */
public static Thread schedulerT;
public static Thread schedulerT = null;
/**
* List of recently deleted resources. It is used to avoid the storage of RPs of a deleted
@ -63,32 +64,41 @@ public class State {
logger.info("Initialising the sweeper...");
// start the sweeper to periodically cleanup the storage and some data structures
Sweeper sweeper = new Sweeper(Long.valueOf((String) ICServiceContext.getContext().getProperty("sweeperIntervalinMillis", true)), Long.valueOf((String) ICServiceContext.getContext()
.getProperty("resourceExpirationTimeInMillis", true)));
State.sweeperT = new Thread(sweeper);
State.sweeperT.start();
if (State.sweeperT == null) {
Sweeper sweeper = new Sweeper(Long.valueOf((String) ICServiceContext.getContext().getProperty("sweeperIntervalinMillis", true)), Long.valueOf((String) ICServiceContext.getContext()
.getProperty("resourceExpirationTimeInMillis", true)));
State.sweeperT = new Thread(sweeper);
State.sweeperT.start();
}
//start the scheduler for automatic backups (if any)
logger.info("Initialising the scheduled backups...");
org.gcube.informationsystem.collector.impl.xmlstorage.backup.Scheduler scheduler = DataManager.getScheduler();
if (scheduler != null) {
schedulerT = new Thread(scheduler);
schedulerT.start();
if (State.schedulerT == null)
State.schedulerT = new Thread(scheduler);
State.schedulerT.start();
}
logger.info("IC service initialization completed");
}
private static void initializeDataManager() throws Exception {
// open the connection used to store resources
State.dataManager = new DataManager();
dataManager.initialize();
if (State.dataManager == null)
State.dataManager = new DataManager();
if (State.dataManager.getStatus() != STATUS.INITIALISED)
dataManager.initialize();
else
logger.info("DataManager already initalized");
}
private static void initializeQueryManager() throws Exception {
// open the connection used to query stored resources
State.queryManager = new QueryManager();
queryManager.initialize();
if (State.queryManager == null)
State.queryManager = new QueryManager();
if (State.queryManager.getStatus() != STATUS.INITIALISED)
queryManager.initialize();
else
logger.info("QueryManager already initalized");
}
/**
@ -99,7 +109,9 @@ public class State {
public static void dispose() throws Exception {
logger.info("Disposing IC service's resources...");
State.sweeperT.interrupt();
State.sweeperT = null;
State.schedulerT.interrupt();
State.schedulerT = null;
State.dataManager.shutdown();
State.queryManager.shutdown();

149
src/org/gcube/informationsystem/collector/impl/xmlstorage/exist/XMLStorageManager.java
View File

@ -3,7 +3,6 @@ package org.gcube.informationsystem.collector.impl.xmlstorage.exist;
import org.xmldb.api.base.XMLDBException;
import org.xmldb.api.base.Collection;
import org.xmldb.api.base.Database;
import org.xmldb.api.base.ResourceSet;
import org.xmldb.api.DatabaseManager;
@ -20,7 +19,6 @@ import org.gcube.informationsystem.collector.impl.persistence.PersistentResource
import org.gcube.informationsystem.collector.impl.persistence.PersistentResource.MalformedResourceException;
import org.gcube.informationsystem.collector.impl.persistence.PersistentResource.RESOURCETYPE;
import org.gcube.informationsystem.collector.impl.xmlstorage.exist.XMLStorageManager;
import org.gcube.informationsystem.collector.impl.xmlstorage.exist.XQuery;
import java.io.BufferedReader;
import java.io.FileReader;
@ -49,7 +47,7 @@ public class XMLStorageManager {
// private Collection currentCollection;
private Collection rootCollection;
protected Collection rootCollection;
private Collection profilesRootCollection;
@ -65,7 +63,8 @@ public class XMLStorageManager {
enum STATUS {INITIALISED, CLOSED, SHUTDOWN};
protected static STATUS status = STATUS.CLOSED;
private STATUS status = STATUS.CLOSED;
/**
* Creates a new manager
@ -84,7 +83,7 @@ public class XMLStorageManager {
*/
public void initialize() throws XMLStorageNotAvailableException {
if (status == STATUS.INITIALISED) {
if (this.getStatus() == STATUS.INITIALISED) {
logger.warn("XMLStorage already initialized");
return;
}
@ -122,7 +121,7 @@ public class XMLStorageManager {
this.rootCollection.setProperty("encoding", "UTF-8");
this.profilesRootCollection.setProperty("pretty", "true");
this.profilesRootCollection.setProperty("encoding", "UTF-8");
status= STATUS.INITIALISED;
this.setStatus(STATUS.INITIALISED);
} catch (XMLDBException edb) {
logger.error("unable to initialize XML storage ", edb);
throw new XMLStorageNotAvailableException("unable to initialize XML storage");
@ -148,7 +147,7 @@ public class XMLStorageManager {
try {
DatabaseInstanceManager manager = (DatabaseInstanceManager) rootCollection.getService("DatabaseInstanceManager", "1.0");
manager.shutdown();
status = STATUS.SHUTDOWN;
this.setStatus(STATUS.SHUTDOWN);
} catch (XMLDBException edb) {
logger.error("Unable to shutdown XML storage");
logger.error("" + edb.getCause());
@ -194,7 +193,7 @@ public class XMLStorageManager {
* Discards the current collection
*
*/
private void resetCollection(Collection currentCollection) {
protected void resetCollection(Collection currentCollection) {
try {
currentCollection.close();
} catch (XMLDBException edb) {
@ -388,63 +387,66 @@ public class XMLStorageManager {
}
/**
* Executes the given query on the current collection or on the root collection if any was
* previously loaded
*
* @param query
* - the query to run
* @return
*/
public ResourceSet executeXQuery(XQuery query) {
boolean retry = true;
int attempts = 0, max_attempts = 3;
ResourceSet result = null;
Collection currentCollection = null;
currentCollection = this.loadAllCollections();
while ((retry) && (attempts < max_attempts)) {
try {
// wait until the DB is unlocked
while (State.getDataManager().isLocked())
Thread.sleep(1000);
// execute query and get results in ResourceSet
if (currentCollection == null)
result = query.execute(this.rootCollection);
else
result = query.execute(currentCollection);
retry = false;
} catch (XMLDBException edb) {
logger.error("Failed to execute XQuery " + query.toString());
logger.error("Error details: " + edb.errorCode + " " + edb.getMessage(), edb);
// if the cause is a NullPointer, this can be due to a temporary
// lock on the database instance
if (edb.getCause() instanceof java.lang.NullPointerException) {
retry = true;
attempts++;
logger.warn("Trying a new attempt for query execution");
} else
retry = false;
} catch (Exception e) {
logger.error("", e);
// if the cause is a NullPointer, this can be due to a temporary
// lock on the database instance
if (e instanceof java.lang.NullPointerException) {
retry = true;
attempts++;
logger.warn("Trying a new attempt for query execution");
} else
retry = false;
}
}
this.resetCollection(currentCollection);
return result;
}
// /**
// * Executes the given XQuery on the current collection or on the root collection if any was
// * previously loaded
// *
// * @param query the XQuery to run
// * @return a formatted resultset
// * @throws XMLStorageNotAvailableException
// */
// public ResourceSet executeXQuery(XQuery query) throws XMLStorageNotAvailableException {
//
// if (status != STATUS.INITIALISED)
// throw new XMLStorageNotAvailableException("XMLStorage is not available");
//
// boolean retry = true;
// int attempts = 0, max_attempts = 3;
// ResourceSet result = null;
// Collection currentCollection = null;
// currentCollection = this.loadAllCollections();
//
// while ((retry) && (attempts < max_attempts)) {
// try {
// // wait until the DB is unlocked
// while (State.getDataManager().isLocked())
// Thread.sleep(1000);
// // execute query and get results in ResourceSet
// if (currentCollection == null)
// result = query.execute(this.rootCollection);
// else
// result = query.execute(currentCollection);
// retry = false;
// } catch (XMLDBException edb) {
// logger.error("Failed to execute XQuery " + query.toString());
// logger.error("Error details: " + edb.errorCode + " " + edb.getMessage(), edb);
// // if the cause is a NullPointer, this can be due to a temporary
// // lock on the database instance
// if (edb.getCause() instanceof java.lang.NullPointerException) {
// retry = true;
// attempts++;
// logger.warn("Trying a new attempt for query execution");
// } else
// retry = false;
//
// } catch (Exception e) {
// logger.error("", e);
// // if the cause is a NullPointer, this can be due to a temporary
// // lock on the database instance
// if (e instanceof java.lang.NullPointerException) {
// retry = true;
// attempts++;
// logger.warn("Trying a new attempt for query execution");
// } else
// retry = false;
//
// }
// }
//
// this.resetCollection(currentCollection);
//
// return result;
// }
/**
* Deletes a WS-ResourceProperties resource identified by the given ID
@ -606,6 +608,23 @@ public class XMLStorageManager {
return listAllColletionIDs(this.loadPropertiesCollection());
}
/**
* @return the status
*/
public STATUS getStatus() {
logger.trace("Status is " + status);
return status;
}
/**
* @param status the status to set
*/
public void setStatus(STATUS status) {
logger.trace("New status is " + status);
this.status = status;
}
private String[] listAllColletionIDs(Collection collection) {
String[] ids = null;
String collectionName = "";

9
src/org/gcube/informationsystem/collector/stubs/testsuite/BackupTester.java
View File

@ -34,18 +34,17 @@ public class BackupTester {
*/
public static void main(String[] args) {
/*if (args.length != 3) {
if (args.length != 3) {
logger.fatal("Usage: BackupTester <host> <port> <Scope>");
return;
}*/
//final String portTypeURI = "http://" + args[0] + ":" + args[1] + "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
}
final String portTypeURI = "http://" + args[0] + ":" + args[1] + "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
String portTypeURI = "http://node10.d.d4science.research-infrastructures.eu:8080/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
XMLStorageAccessPortType port = null;
try {
port = new XMLStorageAccessServiceLocator().getXMLStorageAccessPortTypePort(new URL(portTypeURI));
port = GCUBERemotePortTypeContext.getProxy(port, GCUBEScope.getScope("/CNRPrivate"));
port = GCUBERemotePortTypeContext.getProxy(port, GCUBEScope.getScope(args[2]));
} catch (Exception e) {
logger.error("",e);
}

71
src/org/gcube/informationsystem/collector/stubs/testsuite/ConnectTester.java Normal file
View File

@ -0,0 +1,71 @@
package org.gcube.informationsystem.collector.stubs.testsuite;
import java.net.URL;
import java.rmi.RemoteException;
import org.gcube.common.core.contexts.GCUBERemotePortTypeContext;
import org.gcube.common.core.scope.GCUBEScope;
import org.gcube.common.core.types.VOID;
import org.gcube.common.core.utils.logging.GCUBEClientLog;
import org.gcube.informationsystem.collector.stubs.BackupFailedFaultType;
import org.gcube.informationsystem.collector.stubs.ShutdownFailedFaultType;
import org.gcube.informationsystem.collector.stubs.XMLStorageAccessPortType;
import org.gcube.informationsystem.collector.stubs.XMLStorageNotAvailableFaultType;
import org.gcube.informationsystem.collector.stubs.service.XMLStorageAccessServiceLocator;
/**
* Tester for <em>Connect</em> operation of the
* <em>gcube/informationsystem/collector/XMLStorageAccess</em> portType
*
* @author Manuele Simi (ISTI-CNR)
*
*/
public class ConnectTester {
private static GCUBEClientLog logger = new GCUBEClientLog(ConnectTester.class);
/**
* @param args
* <ol>
* <li>IC host
* <li>IC port
* <li>Caller Scope
* </ol>
*/
public static void main(String[] args) {
if (args.length != 3) {
logger.fatal("Usage: ConnectTester <host> <port> <Scope>");
return;
}
final String portTypeURI = "http://" + args[0] + ":" + args[1] + "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
//final String portTypeURI = "http://node10.d.d4science.research-infrastructures.eu:8080/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
XMLStorageAccessPortType port = null;
try {
port = new XMLStorageAccessServiceLocator().getXMLStorageAccessPortTypePort(new URL(portTypeURI));
port = GCUBERemotePortTypeContext.getProxy(port, GCUBEScope.getScope(args[2]));
} catch (Exception e) {
logger.error("",e);
}
logger.info("Submitting shutdown request to " + portTypeURI+ "...");
try {
port.connect(new VOID());
} catch (XMLStorageNotAvailableFaultType e) {
logger.error("",e);
} catch (ShutdownFailedFaultType e) {
logger.error("",e);
} catch (BackupFailedFaultType e) {
logger.error("",e);
} catch (RemoteException e) {
logger.error("",e);
}
}
}

8
src/org/gcube/informationsystem/collector/stubs/testsuite/RestoreTester.java
View File

@ -33,13 +33,13 @@ public class RestoreTester {
*/
public static void main(String[] args) {
/*if (args.length != 3) {
if (args.length != 3) {
logger.fatal("Usage: RestoreTester <host> <port> <Scope>");
return;
}*/
//final String portTypeURI = "http://" + args[0] + ":" + args[1] + "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
}
final String portTypeURI = "http://" + args[0] + ":" + args[1] + "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
final String portTypeURI = "http://node10.d.d4science.research-infrastructures.eu:8080/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
XMLStorageAccessPortType port = null;
try {

18
src/org/gcube/informationsystem/collector/stubs/testsuite/ShutdownTester.java
View File

@ -21,7 +21,7 @@ import org.gcube.informationsystem.collector.stubs.service.XMLStorageAccessServi
*/
public class ShutdownTester {
private static GCUBEClientLog logger = new GCUBEClientLog(RestoreTester.class);
private static GCUBEClientLog logger = new GCUBEClientLog(ShutdownTester.class);
/**
* @param args
@ -33,18 +33,18 @@ public class ShutdownTester {
*/
public static void main(String[] args) {
/*
* if (args.length != 3) { logger.fatal("Usage: RestoreTester <host> <port> <Scope>");
* return; }
*/
// final String portTypeURI = "http://" + args[0] + ":" + args[1] +
// "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
final String portTypeURI = "http://node10.d.d4science.research-infrastructures.eu:8080/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
if (args.length != 3) {
logger.fatal("Usage: ShutdownTester <host> <port> <Scope>");
return;
}
final String portTypeURI = "http://" + args[0] + ":" + args[1] + "/wsrf/services/gcube/informationsystem/collector/XMLStorageAccess";
XMLStorageAccessPortType port = null;
try {
port = new XMLStorageAccessServiceLocator().getXMLStorageAccessPortTypePort(new URL(portTypeURI));
port = GCUBERemotePortTypeContext.getProxy(port, GCUBEScope.getScope("/CNRPrivate"));
port = GCUBERemotePortTypeContext.getProxy(port, GCUBEScope.getScope(args[2]), 240000);
} catch (Exception e) {
logger.error("",e);
}