Skip to content

Commit

Permalink
Merge pull request #1 from ResearchWorx/bagit-upgrade
Browse files Browse the repository at this point in the history 
Bagit upgrade
  • Loading branch information
@tsuyoi
tsuyoi committed Dec 30, 2018
2 parents 4eb21cb + db6f669 commit ee83400
Show file tree
Hide file tree
Showing 12 changed files with 4,085 additions and 1,632 deletions.
146 changes: 116 additions & 30 deletions pom.xml
View file
Expand Up @@ -18,11 +18,14 @@

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<jdk>1.7</jdk>
<cresco.plugin.library.version>0.4.25</cresco.plugin.library.version>
<jdk>1.8</jdk>
<cresco.plugin.library.version>0.5.3</cresco.plugin.library.version>
<maven.assembly.version>3.0.0</maven.assembly.version>
<maven.source.version>3.0.0</maven.source.version>
<maven.javadoc.version>2.10.3</maven.javadoc.version>
<maven.compiler.version>3.5.1</maven.compiler.version>
<truevfs.version>0.11.0</truevfs.version>
<truecommons.version>2.5.0</truecommons.version>
</properties>

<dependencies>
Expand All @@ -36,34 +39,112 @@
<artifactId>auto-service</artifactId>
<version>1.0-rc2</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.3</version>
</dependency>

<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>18.0</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk</artifactId>
<version>1.11.221</version>
</dependency>
<dependency>
<groupId>commons-configuration</groupId>
<artifactId>commons-configuration</artifactId>
<version>1.10</version>
</dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.6</version>
</dependency>

</dependencies>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.3</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>18.0</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-s3</artifactId>
<version>1.11.371</version>
</dependency>
<dependency>
<groupId>commons-configuration</groupId>
<artifactId>commons-configuration</artifactId>
<version>1.10</version>
</dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.6</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-compress</artifactId>
<version>1.17</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.5</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.6</version>
</dependency>
<dependency>
<groupId>org.apache.tika</groupId>
<artifactId>tika-core</artifactId>
<version>1.18</version>
</dependency>
<dependency>
<groupId>gov.loc</groupId>
<artifactId>bagit</artifactId>
<version>5.2.0</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-access</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-kernel-impl</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-driver-tar</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-driver-tar-bzip2</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-driver-tar-gzip</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-driver-tar-xz</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truevfs</groupId>
<artifactId>truevfs-driver-zip-raes</artifactId>
<version>${truevfs.version}</version>
</dependency>
<dependency>
<groupId>net.java.truecommons</groupId>
<artifactId>truecommons-key-hurlfb</artifactId>
<version>${truecommons.version}</version>
<scope>runtime</scope>
</dependency>
</dependencies>

<build>
<plugins>
Expand All @@ -74,6 +155,9 @@
<configuration>
<source>${jdk}</source>
<target>${jdk}</target>
<compilerArgs>
<arg>-Xlint:unchecked</arg>
</compilerArgs>
</configuration>
</plugin>
<plugin>
Expand All @@ -95,7 +179,9 @@
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>${maven.assembly.version}</version>
<configuration>
<archive>
<manifestEntries>
Expand Down
38 changes: 20 additions & 18 deletions src/main/java/com/researchworx/cresco/plugins/gobjectIngestion/Executor.java
View file
Expand Up @@ -91,8 +91,7 @@ private void pathStage1 (MsgEvent pme, int pathStage) {
} else {
logger.error("Unknown MsgEvent.Type : " + pathStageName);
}
}
catch(Exception ex) {
} catch(Exception ex) {
logger.error("main " + pathStageName + " : " + ex.getMessage() + " " + pme.getParams().toString());
ex.printStackTrace();
}
Expand All @@ -102,12 +101,7 @@ private void pathStage2 (MsgEvent pme, int pathStage) {
String pathStageName = "pathStage" + String.valueOf(pathStage);
try {
MsgEvent.Type eventType = MsgEvent.Type.valueOf(pme.getParam("gmsg_type"));

String transfer_status_file = pme.getParam("transfer_status_file");
String bucket_name = pme.getParam("bucket_name");
String transfer_watch_file = pme.getParam("transfer_watch_file");
String message = pme.getParam("msg");
String endpoint = pme.getParam("endpoint");
String seqId = pme.getParam("seq_id");
String reqId = pme.getParam("req_id");
if (reqId == null)
Expand All @@ -121,7 +115,7 @@ private void pathStage2 (MsgEvent pme, int pathStage) {
case 1:
break;
case 2:
Plugin.objectToFSp.processSequence(seqId, reqId, true);
Plugin.objectToFSp.preprocessBaggedSequence(seqId, reqId, false);
break;
case 3:
break;
Expand All @@ -131,16 +125,15 @@ private void pathStage2 (MsgEvent pme, int pathStage) {
Plugin.objectToFSp.endProcessSequence(seqId, reqId);
break;
default:
logger.error("Undefined pStep " + pathStageName + " !");
logger.error("Undefined sStep " + pathStageName + " !");
break;
}
} else if(eventType.equals(MsgEvent.Type.ERROR)) {

} else {
logger.error("Unknown MsgEvent.Type : " + pathStageName);
}
}
catch(Exception ex) {
} catch(Exception ex) {
logger.error("main " + pathStageName + " : " + ex.getMessage() + " " + pme.getParams().toString());
ex.printStackTrace();
}
Expand Down Expand Up @@ -171,8 +164,7 @@ private void pathStage3 (MsgEvent pme, int pathStage) {
} else {
logger.error("Unknown MsgEvent.Type : " + pathStageName);
}
}
catch(Exception ex) {
} catch(Exception ex) {
logger.error("main " + pathStageName + " : " + ex.getMessage() + " " + pme.getParams().toString());
ex.printStackTrace();
}
Expand All @@ -182,24 +174,32 @@ private void pathStage4 (MsgEvent pme, int pathStage) {
String pathStageName = "pathStage" + String.valueOf(pathStage);
try {
MsgEvent.Type eventType = MsgEvent.Type.valueOf(pme.getParam("gmsg_type"));
logger.info(pathStageName + " " + eventType.name() + " message");
String message = pme.getParam("msg");
String seqId = pme.getParam("seq_id");
String reqId = pme.getParam("req_id");
String sampleId = pme.getParam("sample_id");
if (reqId == null)
reqId = "unknown";
logger.info(pathStageName + " " + message + " " + eventType.name() + " message");
if(eventType.equals(MsgEvent.Type.INFO)) {
int sStep = Integer.parseInt(pme.getParam("sstep"));
switch (sStep) {
case 1:
break;
case 2:
Plugin.objectToFSp.processSample(pme.getParam("seq_id"), pme.getParam("sample_id"), pme.getParam("req_id"), true);
Plugin.objectToFSp.processBaggedSample(seqId, sampleId, reqId, false);
break;
case 3:
Plugin.objectToFSp.processBaggedSample(seqId, sampleId, reqId, false);
break;
case 4:
Plugin.objectToFSp.processBaggedSample(seqId, sampleId, reqId, false);
break;
case 7:
Plugin.objectToFSp.processSample(pme.getParam("seq_id"), pme.getParam("sample_id"), pme.getParam("req_id"), true);
Plugin.objectToFSp.processBaggedSample(seqId, sampleId, reqId, false);
break;
case 999:
Plugin.objectToFSp.endProcessSample(pme.getParam("seq_id"), pme.getParam("sample_id"), pme.getParam("req_id"));
Plugin.objectToFSp.endProcessSample(seqId, sampleId, reqId);
break;
default:
logger.error("Undefined sStep " + pathStageName + " !");
Expand All @@ -221,14 +221,16 @@ private void pathStage5 (MsgEvent pme, int pathStage) {
String pathStageName = "pathStage" + String.valueOf(pathStage);
try {
MsgEvent.Type eventType = MsgEvent.Type.valueOf(pme.getParam("gmsg_type"));
String seqId = pme.getParam("seq_id");
String reqId = pme.getParam("req_id");
logger.info(pathStageName + " " + eventType.name() + " message");
if(eventType.equals(MsgEvent.Type.INFO)) {
int sStep = Integer.parseInt(pme.getParam("sstep"));
switch (sStep) {
case 1:
break;
case 2:
Plugin.objectToFSp.downloadResults(pme.getParam("seq_id"), pme.getParam("req_id"));
Plugin.objectToFSp.downloadBaggedResults(seqId, reqId);
break;
case 3:
break;
Expand Down
45 changes: 38 additions & 7 deletions src/main/java/com/researchworx/cresco/plugins/gobjectIngestion/Plugin.java
View file
Expand Up @@ -6,6 +6,8 @@
import com.researchworx.cresco.plugins.gobjectIngestion.folderprocessor.FSObject;
import com.researchworx.cresco.plugins.gobjectIngestion.folderprocessor.ObjectFS;
import com.researchworx.cresco.plugins.gobjectIngestion.folderprocessor.WatchDirectory;
import com.researchworx.cresco.plugins.gobjectIngestion.objectstorage.Encapsulation;
import com.researchworx.cresco.plugins.gobjectIngestion.objectstorage.LargeBagVerifier;

import java.nio.file.Path;
import java.nio.file.Paths;
Expand All @@ -19,18 +21,45 @@ public class Plugin extends CPlugin {
public static boolean PathProcessorActive = false;

public int pathStage;
public String genomicControllerRegion;
public String genomicControllerAgent;
public String genomicControllerPlugin;
public static ObjectFS objectToFSp;
public static FSObject fStoObjectp;

private String genomicControllerRegion;
private String genomicControllerAgent;
private String genomicControllerPlugin;

static ObjectFS objectToFSp;
static FSObject fStoObjectp;

public static boolean processorIsActive() {
return PathProcessorActive;
}

public static void setActive() {
PathProcessorActive = true;
}

public static void setInactive() {
PathProcessorActive = false;
}

public void setExecutor() {
setExec(new Executor(this));
}

public void start() {
/*try {
URL instanceIDURL = new URL("http://169.254.169.254/latest/meta-data/instance-id");
HttpURLConnection conn = (HttpURLConnection) instanceIDURL.openConnection();
conn.setRequestMethod("GET");
BufferedReader in = new BufferedReader(new InputStreamReader(conn.getInputStream()));
String instanceID = in.readLine();
logger.info("Instance ID: {}", (instanceID != null) ? instanceID : "NULL");
} catch (ProtocolException e) {
logger.error("Protocol exception when getting instance ID: {}", e.getMessage());
} catch (IOException e) {
logger.error("I/O exception when getting instance ID: {}", e.getMessage());
}*/


setExec(new Executor(this));
//logger.setLogLevel(CLogger.Level.Debug);
logger.trace("Building new ConcurrentLinkedQueue");
Expand All @@ -48,7 +77,8 @@ public void start() {
genomicControllerRegion = getConfig().getStringParam("genomic_controller_region",getRegion());
genomicControllerAgent = getConfig().getStringParam("genomic_controller_agent",getAgent());
genomicControllerPlugin = getConfig().getStringParam("genomic_controller_plugin");

Encapsulation.setLogger(this);
LargeBagVerifier.setLogger(this);

logger.debug("[pathStage] == {}", pathStage);
logger.info("Building Stage [{}]", pathStage);
Expand Down Expand Up @@ -124,7 +154,7 @@ public void start() {
logger.trace("Instantiating new [WatchDirectory] from [watchDirectoryName] path");
WatchDirectory wd;
try {
wd = new WatchDirectory(dir, true, this);
wd = new WatchDirectory(dir, false, this);
Thread wdt = new Thread(wd);
wdt.start();
}
Expand Down Expand Up @@ -235,6 +265,7 @@ public MsgEvent genGMessage(MsgEvent.Type met, String msgBody) {

@Override
public void cleanUp() {
setInactive();
MsgEvent me = genGMessage(MsgEvent.Type.INFO, "Shutdown");
me.setParam("pathstage", String.valueOf(pathStage));
me.setParam("pstep","0");
Expand Down

0 comments on commit ee83400

Please sign in to comment.