FourElementsPublic/scheduler4e
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Initial populations

Browse Source
This commit is contained in:
bernarto
2021-11-16 11:19:21 +07:00
parent cdac14e7fd
commit bee21d279a
2168 changed files with 849329 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scheduler_libraries/lib-scheduler-engines/lib/blpapi3.jar
BIN
View File
Binary file not shown.
scheduler_libraries/lib-scheduler-engines/lib/jcs-1.3.3.1-RC.jar
BIN
View File
Binary file not shown.
scheduler_libraries/lib-scheduler-engines/lib/js-14.jar
BIN
View File
Binary file not shown.
scheduler_libraries/lib-scheduler-engines/pom.xml
+156
View File
@@ -0,0 +1,156 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-scheduler</artifactId>
<version>1.0</version>
</parent>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-scheduler-engines</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>lib-scheduler-engines</name>
<description>Scheduler Libraries related to Engines</description>
<url>http://www.fourelementscapital.com</url>
<dependencies>
<dependency>
<groupId>com.googlecode.cqengine</groupId>
<artifactId>cqengine</artifactId>
<version>1.2.2</version>
</dependency>
<dependency>
<groupId>com.jcraft</groupId>
<artifactId>jsch</artifactId>
<version>0.1.50</version>
</dependency>
<dependency>
<groupId>commons-beanutils</groupId>
<artifactId>commons-beanutils</artifactId>
<version>1.8.0</version>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
<version>1.2</version>
</dependency>
<dependency>
<groupId>commons-digester</groupId>
<artifactId>commons-digester</artifactId>
<version>2.0</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>1.4</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.0.9.Final</version>
</dependency>
<dependency>
<groupId>jfree</groupId>
<artifactId>jfreechart-swt</artifactId>
<version>1.0.17</version>
</dependency>
<dependency>
<groupId>net.jxta</groupId>
<artifactId>jxta-jxse</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>org.java-websocket</groupId>
<artifactId>Java-WebSocket</artifactId>
<version>1.3.0</version>
</dependency>
<dependency>
<groupId>org.jfree</groupId>
<artifactId>jcommon</artifactId>
<version>1.0.23</version>
</dependency>
<dependency>
<groupId>org.jfree</groupId>
<artifactId>jfreechart</artifactId>
<version>1.0.19</version>
</dependency>
<dependency>
<groupId>org.json</groupId>
<artifactId>json</artifactId>
<version>20090211</version>
</dependency>
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-catalina</artifactId>
<version>7.0.42</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-coyote</artifactId>
<version>7.0.39</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.bloomberglp.blpapi</groupId>
<artifactId>blpapi</artifactId>
<version>3.2.2.0</version>
<systemPath>${basedir}/lib/blpapi3.jar</systemPath>
<scope>system</scope>
</dependency>
<dependency>
<groupId>org.apache.jcs</groupId>
<artifactId>jcs</artifactId>
<version>1.3.3.1</version>
<systemPath>${basedir}/lib/jcs-1.3.3.1-RC.jar</systemPath>
<scope>system</scope>
</dependency>
<dependency>
<groupId>org.mozilla.javascript.tools.shell</groupId>
<artifactId>js-14</artifactId>
<version>1.2.3</version>
<systemPath>${basedir}/lib/js-14.jar</systemPath>
<scope>system</scope>
</dependency>
<dependency>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-fileutils</artifactId>
<version>2.0</version>
</dependency>
<dependency>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-db</artifactId>
<version>2.0</version>
</dependency>
<dependency>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-r</artifactId>
<version>2.0</version>
</dependency>
<dependency>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-scheduler-exception</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-scheduler-common</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>com.fourelementscapital</groupId>
<artifactId>lib-scheduler-p2p</artifactId>
<version>1.0</version>
</dependency>
</dependencies>
</project>
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/AbstractRScript.java
+81
View File
@@ -0,0 +1,81 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.util.Vector;
import com.fourelementscapital.db.vo.FlexiField;
public abstract class AbstractRScript extends ScheduledTask{
public AbstractRScript(String name, String uid) {
super(name, uid);
try{
addFormFields(getMyFields());
}catch(Exception e){
e.printStackTrace();
}
}
protected Vector<ScheduledTaskField> getMyFields(){
Vector fields=new Vector();
//list
ScheduledTaskField f3=new ScheduledTaskField();
f3.setShortname("rscript_param");
f3.setFieldlabel("");
f3.setFieldtype(FlexiField.TYPE_RSCRIPTEDITOR_PARAM);
f3.setPlacementform("codeinject");
//f3.setFineprint("header code fine print");
fields.add(f3);
ScheduledTaskField f4=new ScheduledTaskField();
f4.setShortname("rscript");
f4.setFieldlabel("R Script");
f4.setFieldtype(FlexiField.TYPE_RSCRIPTEDITOR);
fields.add(f4);
//fields.add(f5);
fields.addAll(getAdditionalRScriptField());
return fields;
}
public Vector<ScheduledTaskField> getAdditionalRScriptField(){
Vector fields=new Vector();
ScheduledTaskField f5=new ScheduledTaskField();
f5.setShortname("trigger_commodity");
f5.setFieldlabel("Trigger Commodity");
f5.setFieldtype(FlexiField.TYPE_TEXTBOX);
//f5.setFineprint("Commodity research ticker that defines the trigger time for Zero-day lag indicators e.g: CL,C");
f5.setFineprint("Building block which fixing time defines the trigger time for Zero-day lag indicators e.g: CL,C,JNso");
f5.setPlacementform("buildingblock");
fields.add(f5);
ScheduledTaskField f6=new ScheduledTaskField();
f6.setShortname("trigger_frequency");
f6.setFieldlabel("Trigger Frequency");
f6.setFieldtype(FlexiField.TYPE_TEXTBOX);
f6.setFineprint("Specific Intervals: 1/3 (starting at 1 and every 3 minutes)");
f6.setPlacementform("buildingblock");
fields.add(f6);
return fields;
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/AbstractRServe.java
+195
View File
@@ -0,0 +1,195 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import org.apache.commons.io.IOUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.JobExecutionException;
import org.rosuda.REngine.Rserve.RConnection;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.exception.ExceptionRServeUnixFailure;
import com.fourelementscapital.scheduler.exception.ExceptionUnixPeerUnknown;
import com.fourelementscapital.scheduler.exception.SchedulerException;
import com.fourelementscapital.scheduler.p2p.P2PService;
import com.fourelementscapital.scheduler.rserve.RServeSession;
public abstract class AbstractRServe extends AbstractRScript{
public AbstractRServe(String name, String uid) {
super(name, uid);
}
private Logger log = LogManager.getLogger(AbstractRServe.class.getName());
protected StackFrame stackframe=null;
public abstract RServeSession getRServeSession(int nid,String name) throws Exception;
public abstract void evalScript(RServeSession rs, String rscript) throws Exception;
public abstract void closeRconnection(RServeSession rs) throws Exception;
public void execute(StackFrame sframe) throws JobExecutionException,Exception {
this.stackframe=sframe;
Map<String, Object> data=sframe.getData();
String rscript=(String)data.get("rscript");
Number nid=(Number)data.get("id");
String name=sframe.getData().get("name").toString();
if(rscript==null) { throw new JobExecutionException("Task Failed because no R script found or empty"); }
//log.debug("R Engine initiated.");
if(rscript!=null && !rscript.equals("")){
try{
//System.out.println(" RServeLowPriorityTask.execute(): executed 11");
String line;
log.debug("R init source file initiated.");
StringBuffer sb=new StringBuffer();
log.debug("------------------<<<<r_script_init:"+Config.getString("r_script_init"));
//collecting Rinit.r content into stringbuffer
if(Config.getString("r_script_init")!=null && new File(Config.getString("r_script_init")).exists()){
log.debug("including r_script_init");
//adding Rinit.r file on top of the script...
BufferedReader reader1 = new BufferedReader(new FileReader(Config.getString("r_script_init")));
String line1;
while ((line1 = reader1.readLine()) != null) {
if(line1!=null && !line1.equals("")){
sb.append(line1+"\n");
}
}
reader1.close();
log.debug("R init source file initiated.");
}
//collecting main script...
BufferedReader reader = new BufferedReader(new StringReader(rscript));
while ((line = reader.readLine()) != null) {
if(line!=null && !line.equals("")){
//out.write(line+"\n");
sb.append(line+"\n");
}
}
//out.close();
reader.close();
RConnection c=null;
RServeSession rs=null;
File temp=null;
try{
rs=getRServeSession(nid.intValue(), name);
//c=getRconnection(nid.intValue(), name);
if(rs!=null && rs.getRconnection()!=null ){
new SchedulerExePlanLogs(nid.intValue(),sframe.getTrigger_time()).log("RServe session started..",SchedulerExePlanLogs.PEER_OK_RESPOND_RSERVE_SESSIONSTARTED);
c=rs.getRconnection();
c.assign(".trigger_time", sframe.getTrigger_time()+"");
c.assign(".scheduler_id", nid+"");
c.assign(".machine", P2PService.getComputerName());
c.assign(".enginetype","RServeUnix");
c.assign(".connection_ids","");
//sink staring here....
temp = File.createTempFile(nid.intValue()+"_"+sframe.getTrigger_time(),".log");
temp.getParentFile().setWritable(true, false);
String ff=temp.getPath();
if(ff.contains("\\")){
ff=ff.replaceAll("\\\\", "\\\\\\\\");
}
temp.delete();
log.debug("---- deleting ff:"+ff);
String fileass="file(\""+ff+"\", open = \"wt\");";
log.debug("file assign------fileass:"+fileass);
c.eval(".console_msg_zz<-"+fileass);
c.eval("sink(.console_msg_zz)");
c.eval("sink(.console_msg_zz, type = \"message\")") ;
log.debug("----sink----");
if(false){
//ignored as it is not useful anymore
c.eval("if(length(.r_scripts)>12){ .r_scripts=.r_scripts[(length(.r_scripts)-10):length(.r_scripts)] }"); //keep only last 10 when it goes more than 12 to avoid keeping too many things in memory.
SimpleDateFormat sdf=new SimpleDateFormat("yyyyMMdd_HHmmss");
String tl=nid+"|"+sframe.getTrigger_time()+"|"+sdf.format(new Date());
c.eval(".r_scripts=append(.r_scripts,\""+tl+"\")");
}
new SchedulerExePlanLogs(nid.intValue(),sframe.getTrigger_time()).log("RServe Script evaluation starting..",SchedulerExePlanLogs.PEER_OK_RSERVE_SCRIPT_EVAL_STARTING);
evalScript(rs,sb.toString());
}else{
throw new ExceptionRServeUnixFailure("Couldn't start RServ Connection!, Connection is null");
}
}catch(SchedulerException se){
throw se;
}catch(Exception e){
sframe.setTasklog(e.getMessage());
throw new ExceptionUnixPeerUnknown("RServeUnixTask: Error:"+e.getMessage());
}finally{
if(temp!=null && temp.exists()){
FileInputStream inputStream = new FileInputStream(temp);
try{
sframe.setConsole_message(IOUtils.toString(inputStream));
}catch(Exception e){
log.error("error while accessing the file, Error:"+e.getMessage());
}finally{
inputStream.close();
}
temp.delete();
}
if(c!=null){
//sink ending here....
c.eval("sink(type = \"message\")");
c.eval("sink()");
c.eval("close(.console_msg_zz)");
closeRconnection(rs);
}
//Runtime.getRuntime().gc();
}
try{ Thread.sleep(200); }catch(Exception e){} //200 milliseconds to remove the resource completely from memory
log.debug(" end of execute() script :"+name);
}catch(Exception e){
//ClientErrorMgmt.reportError(e,null);
throw e;
}
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/BloombergDownload.java
+171
View File
@@ -0,0 +1,171 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.jfree.util.Log;
import org.quartz.JobExecutionException;
import com.fourelementscapital.db.BBSyncDB;
import com.fourelementscapital.db.vo.FlexiField;
import com.fourelementscapital.scheduler.exception.ExceptionWarningNoFullData;
import com.fourelementscapital.scheduler.pluggin.SchedulerBloombergPlugin;
public class BloombergDownload extends ScheduledTask {
Logger log = LogManager.getLogger(BloombergDownload.class.getName());
public BloombergDownload(String name, String taskuid) {
//super("Bloomberg Download","bb_download");
super(name,taskuid);
try{
addFormFields(getMyFields());
}catch(Exception e){
e.printStackTrace();
}
}
public synchronized void execute(StackFrame sframe) throws JobExecutionException,ExceptionWarningNoFullData,Exception {
Map data=sframe.getData();
if(data.get("download_id")==null || (data.get("download_id")!=null && ((String)data.get("download_id")).equals("")) ) {
throw new JobExecutionException("Job can't be executed as you did not not select download query");
}else{
try{
String id=(String)data.get("download_id");
String id1=data.get("id").toString();
log.debug("sc_id:"+id1+":before opening database ...");
String rlsdt_adjustment=(String)data.get("rlsdt_adjustment");
BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
log.debug("sc_id:"+id1+":syncdb init");
syncdb.connectDB();
if(true){
//BoneCP bcp=syncdb.getConnectionPool();
//log.debug("sc_id:"+id1+"----------------------------------------");
//if(bcp!=null){
//log.debug("sc_id:"+id1+":syncdb CP Free connection :"+bcp.getStatistics().getTotalFree());
//log.debug("sc_id:"+id1+":syncdb CP Leased connection :"+bcp.getStatistics().getTotalLeased());
//log.debug("sc_id:"+id1+":syncdb CP Total Created connection :"+bcp.getStatistics().getTotalCreatedConnections());
//log.debug("sc_id:"+id1+":syncdb CP Cache Hits :"+bcp.getStatistics().getCacheHits());
//log.debug("sc_id:"+id1+":syncdb CP Cache miss:"+bcp.getStatistics().getCacheMiss());
//}
}
log.debug("sc_id:"+id1+":connected");
Map data1=syncdb.getDownloadQuery(Integer.parseInt(id));
log.debug("sc_id:"+id1+":after getDownloadQuery called");
if(rlsdt_adjustment!=null && !rlsdt_adjustment.equals("")){
try{
int dadj=Integer.parseInt(rlsdt_adjustment);
data1.put("enddt_adjustment",new Integer(dadj));
}catch(Exception e){
//ClientErrorMgmt.reportError(e, "Error while parsing release date adjument to Integer");
throw e;
}
}
if(data1!=null){
log.debug("sc_id:"+id+":after before bloombergDownloadJob init");
BloombergDownloadJob bjdo= new BloombergDownloadJob();
bjdo.setInvokedByScheduler();
try{
bjdo.downloadData(data1,sframe);
log.debug("sc_id:"+id1+":after before downloadData called");
}catch(ExceptionWarningNoFullData wnd){
throw wnd;
}catch(Exception e1){
throw new JobExecutionException(e1);
}finally{
sframe.setTasklog(bjdo.getSchedulerLog());
if(bjdo.getSchedulerLog()!=null && !bjdo.getSchedulerLog().equals("")){
sframe.setStatus(ScheduledTask.EXCECUTION_WARNING);
}
}
}else{
throw new JobExecutionException("Job can't be executed, relevant download query not found (you may have deleted)");
}
syncdb.closeDB();
}catch(ExceptionWarningNoFullData wnd1){
throw wnd1;
}catch(Exception e){
//ClientErrorMgmt.reportError(e, "id:"+id);
throw new JobExecutionException(e);
}
}
}
public List<ScheduledTaskField> listFormFields(){
super.removeAllFields();
try{
super.addFormFields(getMyFields());
}catch(Exception e){
Log.error("Error in refreshing fields");
}
return super.listFormFields();
}
private Vector<ScheduledTaskField> getMyFields() throws Exception {
Vector fields=new Vector();
ScheduledTaskField f3=new ScheduledTaskField();
f3.setShortname("rlsdt_adjustment");
f3.setFieldlabel("Release Date Adjustment");
f3.setFieldtype(FlexiField.TYPE_TEXTBOX);
fields.add(f3);
ScheduledTaskField f1=new ScheduledTaskField();
f1.setShortname(SchedulerBloombergPlugin.PLUGGIN_IN);
f1.setFieldlabel("Bloomberg Query");
f1.setFieldtype(FlexiField.TYPE_BLOOMBERG_PLUGGIN);
f1.setPluggindata(new SchedulerBloombergPlugin().getPlugginData());
fields.add(f1);
ScheduledTaskField f2=new ScheduledTaskField();
f2.setShortname(new SchedulerBloombergPlugin().getPlugginData().getFieldreference());
f2.setFieldlabel("download_id");
f2.setFieldtype(FlexiField.TYPE_HIDDENFIED);
fields.add(f2);
return fields;
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/BloombergDownloadJob.java
+487
View File
@@ -0,0 +1,487 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.io.File;
import java.io.FileWriter;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.TreeMap;
import java.util.Vector;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import com.fourelementscapital.db.BBSyncDB;
import com.fourelementscapital.db.InfrastructureDB;
import com.fourelementscapital.fileutils.SplitString;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.exception.ExceptionWarningNoFullData;
public class BloombergDownloadJob implements Job {
private Logger log = LogManager.getLogger(BloombergDownloadJob.class.getName());
public static String BB_DOWNLOAD_GROUP="BB_DOWNLOAD_GROUP";
private Vector<String> uniqueContracts=new Vector<String>();
private boolean invokedByUser=true;
private String schedulerLog=null;
public void execute(JobExecutionContext context) throws JobExecutionException {
this.invokedByUser=false;
Map data=(Map)context.getJobDetail().getJobDataMap().get("data");
//log.debug("triggered data:"+data);
int id=(Integer)data.get("id");
String name=(String)data.get("name");
Date started=new Date();
try{
//downloadData(data);
}catch(Exception e){
//logging
//externalLog(id,name,started,new Date(), null,null,0,false);
//ClientErrorMgmt.reportError(e, "error while synchronizing download id:"+data.get("id")+" name:"+data.get("name"));
throw e;
}
}
public void setInvokedByScheduler(){
this.invokedByUser=false;
}
/**
* @deprecated
* @param bb_syncid
* @throws Exception
*/
public void rescheduleJobs(long bb_syncid) throws Exception {
}
private String checkCommoditySpecified(String contract, Map scommodities,HashMap<String,String> mc) {
String commodity=null;
//parses data rediction tickers (For Example: CLZ9->CLZ99 ) download CLZ9 ticker and keep it in both CLZ9 and CLZ99
Pattern pattern = Pattern.compile("^(.*?)->(.*?)$");
Matcher matcher = pattern.matcher(contract);
if (matcher.find()) {
contract = matcher.group(1);
String copy=matcher.group(2);
if(copy!=null){
//StringTokenizer st=new StringTokenizer(copy,",");
//while(st.hasMoreTokens()) list.add(st.nextToken());
String cont=contract;
if(cont.indexOf("{")>=0 && cont.indexOf("}")>0){
String comm=cont.substring(cont.indexOf("{")+1,cont.indexOf("}"));
cont=cont.replaceAll("\\{"+comm+"\\}", "");
}
mc.put(cont, copy);
}
}
if(contract.indexOf("{")>=0 && contract.indexOf("}")>0){
commodity=contract.substring(contract.indexOf("{")+1,contract.indexOf("}"));
contract=contract.replaceAll("\\{"+commodity+"\\}", "");
scommodities.put(contract, commodity);
}
return contract;
}
private ArrayList<ArrayList<String>> splitTickers(ArrayList<String> tickers, int each) {
ArrayList<ArrayList<String>> all=new ArrayList<ArrayList<String>>();
if(tickers.size()>each) {
all.add(new ArrayList<String>());
for(String tick:tickers){
if((all.get(all.size()-1).size()>each)){
all.add(new ArrayList<String>());
}
all.get(all.size()-1).add(tick);
}
}else{
all.add(tickers);
}
return all;
/*
for(ArrayList<String> t1:all){
System.out.print("\n----------------");
for(String tick:t1){
System.out.print(tick +",");
}
}
*/
}
public synchronized void downloadData(Map m,StackFrame sframe) throws Exception {
int scheduler_id=((Number)sframe.getData().get("id")).intValue();
long trigger_time=sframe.getTrigger_time();
java.sql.Timestamp startedat=new java.sql.Timestamp(new Date().getTime());
TreeMap fields=new TreeMap();
processDate(m);
String date_from=(String)m.get("date_from");
String date_to=(String)m.get("date_to");
String marketsector=(String)m.get("marketsector");
String commString=(String)m.get("tickers");
ArrayList<String> tickers=new ArrayList<String>();
List tickers1=SplitString.split(commString);
HashMap scommodities=new HashMap ();
HashMap<String,String> multiplecopy=new HashMap<String,String>();
for(Iterator<String>i = tickers1.iterator();i.hasNext();){
String contractraw=i.next();
tickers.add(checkCommoditySpecified(contractraw,scommodities,multiplecopy));
}
BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
syncdb.connectDB();
int id=(Integer)m.get("id");
ArrayList fmap=syncdb.fieldMapping4BBSync(id);
for(int i=0;i<fmap.size();i++){
Map rdata=(Map)fmap.get(i);
fields.put(rdata.get("bb_field"),rdata.get("db_field"));
}
syncdb.closeDB();
log.debug("date_from:"+date_from);
log.debug("date_to:"+date_to);
log.debug("market sector:"+marketsector);
log.debug("fields:"+fields.keySet()); //Bloomberg field names i.e "PX_CLOSE"
log.debug("db fields:"+fields.values()); //friendly field names i.e "Close Price"
log.debug("tickers:"+tickers);
boolean is_market_sec=false;
if(m.get("is_mkt_securitydb")!=null && ((String)m.get("is_mkt_securitydb")).toLowerCase().equals("true")){
is_market_sec=true;
}
ArrayList bb_fields=new ArrayList(fields.keySet());
if(date_from!=null && date_to!=null && marketsector!=null && bb_fields.size()>0 && tickers.size()>0){
ArrayList<ArrayList<String>> all=splitTickers(tickers,100);
for(ArrayList<String> batch:all){
//TreeMap data=new ConnectBloomberg(tickers,date_from,date_to,bb_fields,marketsector,new MigrationMgmt(),true).getMultipleData();
//processBBData(null,data,is_market_sec,fields,tickers,scommodities,multiplecopy); //original
//processBBData(marketsector,data,is_market_sec,fields,tickers,scommodities,multiplecopy); //cloned copy of market sector for example CL_Commdty_CLOSE_Price
new SchedulerExePlanLogs(scheduler_id,trigger_time).log("# of tickers in this request:"+batch.size(),SchedulerExePlanLogs.PEER_OK_BLOOMBERG_DOWNLOAD);
//TreeMap data=new ConnectBloomberg(batch,date_from,date_to,bb_fields,marketsector,new MigrationMgmt(),true).getMultipleData();
//processBBData(null,data,is_market_sec,fields,batch,scommodities,multiplecopy); //original
//processBBData(marketsector,data,is_market_sec,fields,batch,scommodities,multiplecopy); //cloned copy of market sector for example CL_Commdty_CLOSE_Price
}
}
//UPDATE CONTRACT LOGS.
log.debug("unique contracts:"+uniqueContracts);
syncdb.closeDB();
BBSyncDB syncdb1=BBSyncDB.getBBSyncDB();
syncdb1.connectDB();
if(uniqueContracts.size()>0 && !is_market_sec){
java.sql.Timestamp lastsync=new java.sql.Timestamp(new Date().getTime());
syncdb1.updateContractLogs(uniqueContracts,marketsector,lastsync,scommodities,fields.values());
}else if(uniqueContracts.size()>0 && is_market_sec){
java.sql.Timestamp lastsync=new java.sql.Timestamp(new Date().getTime());
syncdb1.updateSecuritiesLogs(uniqueContracts,marketsector, lastsync,fields.values());
}
//update bloomberg counter;
SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd");
sdf.setTimeZone(TimeZone.getTimeZone("Asia/Tokyo"));
String japan_time=sdf.format(new Date());
int count=uniqueContracts.size()* fields.keySet().size();
InfrastructureDB infrastructureDB = InfrastructureDB.getInfrastructureDB();
infrastructureDB.connectDB();
if(count>0){
infrastructureDB.updateBloomberDailyCounter(japan_time, count);
}
infrastructureDB.closeDB();
//log.debug("job "+name+" excecuted");
log.debug("after updated unique tickers on ref database");
java.sql.Timestamp endedat=new java.sql.Timestamp(new Date().getTime());
syncdb1.updateTriggeredDate(id, startedat, endedat);
Date st=(Date)m.get("date_from_date");
Date end=(Date)m.get("date_to_date");
String name=(String)m.get("name");
log.debug("writting external log");
externalLog(id,name,startedat,endedat,st,end,uniqueContracts.size(),true,syncdb1);
syncdb1.closeDB();
schedulerLog="";
log.debug("writting scheduler logs");
schedulerLog=schedulerLog+"Date Between "+new SimpleDateFormat("dd-MMM-yy").format(st)+" and "+new SimpleDateFormat("dd-MMM-yy").format(end);
schedulerLog=schedulerLog+"\nNo of Tickers Requested:"+tickers.size();
schedulerLog=schedulerLog+"\nNo of Tickers Responded:"+uniqueContracts.size();
String missingtick=null;
for(Iterator it=tickers.iterator();it.hasNext();){
String tick=(String)it.next();
if(tick!=null && !tick.equals("") && uniqueContracts.indexOf(tick)<0){
missingtick=(missingtick==null)?tick:missingtick+"\n"+tick;
}
}
if(missingtick!=null){
schedulerLog=schedulerLog+"\nNo data found for the following tickers:\n"+missingtick;
throw new ExceptionWarningNoFullData("No data found for some tickers ");
}else{
schedulerLog=null;
}
log.debug("--------------end of downloadData() -----------------------");
}
public String getSchedulerLog(){
return this.schedulerLog;
}
private void processDate(Map m){
String date_option=(String)m.get("date_option");
int date_recentnumber=(Integer)m.get("date_recentnumber");
Date date_from=null;
Date date_to=null;
try{
if(m.get("date_from") instanceof Date){
date_from=(Date)m.get("date_from");
}else{
log.error("date_from is not date instance");
}
if(m.get("date_to") instanceof Date){
date_to=(Date)m.get("date_to");
}else{
log.error("date_to is not date instance");
}
}catch(Exception e){
//ClientErrorMgmt.reportError(e, "Error while parsing date, download ID:"+m.get("id"));
throw e;
}
if(date_option.toLowerCase().equals("daterange")){
}else if(date_option.toLowerCase().equals("datefrom")){
date_to=new Date();
}else{
date_to=new Date();
date_from=null;
Calendar today=Calendar.getInstance();
today.setTime(date_to);
boolean processed=false;
if(date_option.toLowerCase().equals("ndays")){
today.add(Calendar.DAY_OF_YEAR, -(date_recentnumber-1));
processed=true;
}else if(date_option.toLowerCase().equals("nweeks")){
today.add(Calendar.WEEK_OF_YEAR, -date_recentnumber );
processed=true;
}else if(date_option.toLowerCase().equals("nmonths")){
today.add(Calendar.MONTH, -date_recentnumber );
processed=true;
}else if(date_option.toLowerCase().equals("nyears")){
today.add(Calendar.YEAR, -date_recentnumber );
processed=true;
}
if(processed){
date_from=today.getTime();
}
}
if(date_from!=null){
m.put("date_from",new SimpleDateFormat("yyyyMMdd").format((Date)date_from));
m.put("date_from_date",date_from);
}
if(date_to!=null){
if(m.get("enddt_adjustment")!=null){
Integer dadj=(Integer)m.get("enddt_adjustment");
Calendar dt=Calendar.getInstance();
dt.setTime(date_to);
dt.add(Calendar.DAY_OF_YEAR, dadj);
m.put("date_to",new SimpleDateFormat("yyyyMMdd").format(dt.getTime()));
m.put("date_to_date",dt.getTime());
}else{
m.put("date_to",new SimpleDateFormat("yyyyMMdd").format((Date)date_to));
m.put("date_to_date",date_to);
}
}
}
private void externalLog(int id,String name,Date started, Date ended,Date period_from,Date period_to,long num_contracts, boolean flag ,BBSyncDB syncdb){
boolean SUCCESS=true;
boolean FAILED=false;
//String filename=new File(job.getConfolder()).getParent()+File.separator+"scheduler.log";
//File file=new File(filename);
String root=Config.getString("log_error_folder");
Date d = new Date();
String folder =root+"bloomberg_sync";
if (!new File(folder).isDirectory()) {
new File(folder).mkdirs();
}
String fname=root+"bloomberg_sync"+File.separator+""+new SimpleDateFormat("dd_MM_yyyy").format(d)+".csv";
File file=new File(fname);
if (!file.exists()) {
try {
//file.createNewFile();
FileWriter fw1 = new FileWriter(file,true);
fw1.append("Date, Time, Duration(mins), Period, # of Tickers, Manual/Scheduler, Status, ID, Download Name \r\n");
fw1.flush();
fw1.close();
}catch(Exception e){
log.error("error while creating file: Error:"+e.getMessage());
}
}
try {
boolean append = true;
FileWriter fw = new FileWriter(file,append);
//String timeformat="dd-MMM hh:mm:ss aaa ";
SimpleDateFormat timeformat=new SimpleDateFormat("h:mm a");
SimpleDateFormat durationformat =new SimpleDateFormat("mm:ss");
durationformat.setTimeZone(TimeZone.getTimeZone("GMT"));
SimpleDateFormat dateformat=new SimpleDateFormat("dd MMM-yy");
String logmessage="";
logmessage+=dateformat.format(started)+",";
logmessage+=timeformat.format(started)+",";
long diff=ended.getTime()-started.getTime();
logmessage+=durationformat.format(new Date(diff))+",";
long mills_per_day = 1000 * 60 * 60 * 24;
long day_diff = ( period_to.getTime() - period_from.getTime() ) / mills_per_day;
logmessage+=(Math.round(day_diff)+1)+" days from "+dateformat.format(period_from)+",";
logmessage+=num_contracts+",";
logmessage+=(this.invokedByUser?"Manual":"Scheduler")+",";
logmessage+=( flag==SUCCESS?"Success":"Failed")+",";
logmessage+=id+",";
logmessage+=name+",";
/*
logmessage+="ID:"+id;
logmessage+="\tDownload Name:"+name;
logmessage+="\tStarted At:"+new SimpleDateFormat(format).format(started);
if(flag==SUCCESS){
logmessage+="\tEnded At:"+new SimpleDateFormat(format).format(ended);
logmessage+="\tStatus:Success";
}else{
logmessage+="\tStatus:Failed";
}
if(period_from!=null && period_to!=null){
logmessage+="\t Date Between("+sf.format(period_from)+" AND "+sf.format(period_to)+")";
}
*/
fw.append(logmessage+"\r\n");
fw.flush();
fw.close();
//add to database.
//BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
//syncdb.connectDB();
syncdb.addSyncLogs(id, new Timestamp(started.getTime()), new Timestamp(ended.getTime()), null, ((flag)?"Success":"Failed"),(this.invokedByUser?"M":"S"));
//syncdb.closeDB();
//syncdb.closeDB();
}
catch (Exception ex) {
log.error("Error:"+ex.getMessage());
//ClientErrorMgmt.reportError(ex,"Error in external log file writing");
ex.printStackTrace();
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/ExecuteRUnix.java
+261
View File
@@ -0,0 +1,261 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.JobExecutionException;
import org.rosuda.REngine.REXP;
import org.rosuda.REngine.RList;
import org.rosuda.REngine.Rserve.RConnection;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.exception.ExceptionRServeUnixFailure;
import com.fourelementscapital.scheduler.p2p.P2PService;
import com.fourelementscapital.scheduler.rserve.RServeConnectionPool;
import com.fourelementscapital.scheduler.rserve.RServeSession;
//import com.fourelementscapital.client.ClientErrorMgmt;
public abstract class ExecuteRUnix extends ScheduledTask {
private Logger log = LogManager.getLogger(ExecuteRUnix.class.getName());
private StackFrame stackframe=null;
public ExecuteRUnix(String name,String taskuid ) {
super(name,taskuid);
}
public void execute(StackFrame sframe) throws JobExecutionException,Exception {
this.stackframe=sframe;
Map<String, Object> data=sframe.getData();
if(sframe.getRscript()==null) throw new JobExecutionException("Script failed as StackFrame doesn't containt RScript object");
String rscript=sframe.getRscript().getScript();
if(rscript==null) { throw new JobExecutionException("Task Failed because no R script found or empty"); }
log.debug("R Engine initiated.");
if(rscript!=null && !rscript.equals("")){
try{
//System.out.println(" RServeLowPriorityTask.execute(): executed 11");
String line;
log.debug("R init source file initiated.");
//BufferedReader reader = new BufferedReader(new StringReader(rscript));
String tempDir = System.getProperty("java.io.tmpdir");
log.debug("set temporary directory:"+tempDir);
File file=new File(tempDir+File.separator+"rscript"+new SimpleDateFormat("mmddyyy_hh_mmss_").format(new Date())+new Random().nextInt()+".R");
BufferedWriter out = new BufferedWriter(new FileWriter(file));
//adding Rinit.r file on top of the script...
BufferedReader reader1 = new BufferedReader(new FileReader(Config.getString("r_script_init")));
StringBuffer sb=new StringBuffer();
String line1;
while ((line1 = reader1.readLine()) != null) {
if(line1!=null && !line1.equals("")){
//out.write(line1+"\n");
sb.append(line1+"\n");
}
}
reader1.close();
log.debug("R init source file initiated.");
String filepath=file.getPath().replaceAll("\\\\", "\\\\\\\\");
RConnection c=null;
RServeSession rs=null;
try{
log.debug("before opening session----<");
log.debug("rs:"+rs);
rs=getRServeSession(sframe.getRscript().getUid(),sb.toString());
if(rs!=null && rs.getRconnection()!=null ){
c=rs.getRconnection();
log.debug("c:"+c.isConnected());
REXP sid1=c.eval(".r_session_id");
if(sid1!=null){
String sess_id=sid1.asString();
log.debug(".r_session_id:"+sess_id);
}
c.assign(".machine", P2PService.getComputerName());
log.debug("rscript:"+rscript);
REXP x1= c.parseAndEval(rscript);
String result=parse(x1);
sframe.getRscript().setResultXML(result);
log.debug("Script from temp file Evaluated");
log.debug("x1:"+x1);
if (x1!=null && x1.inherits("try-error")) {
throw new Exception("Error in R Script: "+x1.asString());
}
if(x1==null) {
throw new Exception("R script:"+data.get("name")+" failed");
}else{
sframe.setTasklog(null);
}
}
file.delete();
}catch(Exception e){
sframe.setTasklog(e.getMessage());
sframe.getRscript().setError(e.getMessage());
throw new Exception ("Error:"+e.getMessage());
//throw e;
}finally{
if(c!=null){
//ConnectionPool.done(c.detach());
closeRconnection(rs);
//c.close();
//c.shutdown();
}
Runtime.getRuntime().gc();
}
//try{ Thread.sleep(5); }catch(Exception e){} //200 milliseconds to remove the resource completely from memory
}catch(Exception e){
//e.printStackTrace();
//ClientErrorMgmt.reportError(e,null);
throw e;
}
}
}
/**
*
* @param rxps
* @return
* @throws Exception
*/
private String parse(REXP rxp) throws Exception {
String rtn="";
if(rxp.isNumeric()){
rtn+="<double>"+rxp.asDouble()+"</double>";
}
if(rxp.isInteger()){
rtn+="<integer>"+rxp.asInteger()+"</integer>";
}
if(rxp.isList()){
RList ar=rxp.asList();
rtn+="<array>";
for(Iterator i=ar.iterator();i.hasNext();){
Object obj=i.next();
if(obj instanceof REXP){
if(rxp.isNumeric()) rtn+="<double>"+((REXP)obj).asDouble()+"</double>";
if(rxp.isString()) rtn+="<string>"+((REXP)obj).asString()+"</string>";
if(rxp.isString()) rtn+="<integer>"+((REXP)obj).asInteger()+"</integer>";
}
}
rtn+="</array>";
}
if(rxp.isString()){
rtn+="<string>"+rxp.asString()+"</string>";
}
//if(rxp.getType()==REXP.XT_STR || rxp.getType()==REXP.XT_VECTOR || rxp.getType()==REXP.XT_DOUBLE || rxp.getType()==REXP.XT_ARRAY_DOUBLE){}
//else log.debug("rxp.getType():"+rxp.getType());
return rtn;
}
public RServeSession getRServeSession(String script_uid, String startup_rinit) throws Exception {
RServeSession rs=null;
RConnection c=null;
try{
rs=RServeConnectionPool.getRSession(startup_rinit);
}catch(Exception e){
throw new ExceptionRServeUnixFailure("Not able to get RSession from ConnectionPool!, Error:"+e.getMessage());
}
try{
log.debug("+++++++attaches Rconnection from RServeSession....+++");
c=(rs.getRconnection()!=null) ? rs.getRconnection(): rs.getRsession().attach();
rs.setRconnection(c);
}catch(Exception e){
throw new ExceptionRServeUnixFailure("Not able to attach connection to RSession!, Error:"+e.getMessage());
}
if(c!=null){
log.debug("c:"+c.isConnected());
rs.setRunning(true);
rs.setExecute_r_uid(script_uid);
rs.setStarted_time(new Date().getTime());
rs.setThread(Thread.currentThread());
}
return rs;
}
private void closeRconnection(RServeSession rs) throws Exception {
try{
rs.finished(rs.getRconnection());
RServeConnectionPool.done(rs);
}catch(Exception e){
if(rs.getKillmessage()!=null){
throw new ExceptionRServeUnixFailure(rs.getKillmessage());
}else{
throw new ExceptionRServeUnixFailure("Not able to detach RSession!, Error:"+e.getMessage());
}
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/ExecuteRWindows.java
+170
View File
@@ -0,0 +1,170 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.JobExecutionException;
import org.rosuda.JRI.REXP;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.exception.ExceptionPeerUnknown;
import com.fourelementscapital.scheduler.exception.ExceptionRscriptError;
import com.fourelementscapital.scheduler.exception.SchedulerException;
import com.fourelementscapital.scheduler.p2p.P2PService;
public abstract class ExecuteRWindows extends ScheduledTask {
public ExecuteRWindows(String name,String taskuid ) {
//super("Medium Priority (RServe)","rscript4rserve");
super(name,taskuid);
}
private Logger log = LogManager.getLogger(ExecuteRWindows.class.getName());
protected static MyRengine re=null;
public synchronized void execute(StackFrame sframe) throws JobExecutionException,Exception {
log.debug("execute() called");
if(Config.getString("p2p.ignorejri")!=null && Config.getString("p2p.ignorejri").equalsIgnoreCase("true")){
throw new Exception("JRI is not enalbed");
}
String rscript=sframe.getRscript().getScript();
if(rscript==null) { throw new JobExecutionException("Task Failed because no R script found or empty"); }
MyRengineCallbacks myrenging=new MyRengineCallbacks(); //new MyRengineCallbacks();
if(re==null){
log.debug("Rengine is null, creating one.......");
re=new MyRengine(null, false, myrenging );
re.setStackFrame(sframe);
//for the first time it loads Rinit;
BufferedReader reader1 = new BufferedReader(new FileReader(Config.getString("r_script_init")));
String line;
while ((line = reader1.readLine()) != null) {
if(line!=null && !line.equals("")){
REXP x=re.eval(line);
}
}
reader1.close();
}else{
re.setStackFrame(sframe);
//re.addMainLoopCallbacks(myrenging);
}
re.getRMailLoopCallback().setConsoleOutput(null);
if(rscript!=null && !rscript.equals("")){
try{
re.assign(".machine", P2PService.getComputerName());
sframe.setStarted_time(new Date().getTime());
log.debug("before executing script....");
BufferedReader reader = new BufferedReader(new StringReader(rscript));
StringBuffer sb=new StringBuffer();
String tempDir = System.getProperty("java.io.tmpdir");
File file=new File(tempDir+File.separator+"rscript"+new SimpleDateFormat("mmddyyy_hh_mmss_").format(new Date())+new Random().nextInt()+".R");
String line;
BufferedWriter out = new BufferedWriter(new FileWriter(file));
while ((line = reader.readLine()) != null) {
if(line!=null && !line.equals("")){
out.write(line+"\n");
}
}
out.close();
reader.close();
String filepath=file.getPath().replaceAll("\\\\", "\\\\\\\\");
log.debug("file:"+file.getPath()+" re:"+re);
REXP x1= re.eval("source(\""+filepath+"\")");
sframe.getRscript().setResult(x1);
log.debug("x1:"+x1);
file.delete();
if(x1==null) {
//sframe.setTasklog("R Script failed. See log file for more info");
REXP x3= re.eval("as.character(.Traceback)");
if(x3.rtype==REXP.STRSXP){
String a[]=x3.asStringArray();
String err="\n----Error stack----";
if(a.length>=2){
for(int i=a.length-2;i>=0;i--){
if(!a[i].startsWith("eval.with.vis")){
String ind="\n-";
ind+=">"+a[i];
err+=ind;
}
}
}
myrenging.externalLog(re, err);
String all_err="";
if(sframe.getTasklog()!=null){
all_err=err+"\n"+sframe.getTasklog();
}else{
all_err=err;
}
throw new ExceptionRscriptError("R Script Error:"+all_err);
}else{
throw new ExceptionRscriptError("R script:"+rscript+" failed, "+sframe.getTasklog());
}
}else{
//sframe.setTasklog(x1.toString());
//executed successfully so don't store log messages
sframe.setTasklog(null);
re.setStackFrame(null);
}
}catch(SchedulerException se){
throw se;
}catch(Exception e){
throw new ExceptionPeerUnknown("REngineScriptTask: Error:"+e.getMessage());
}
}
//setMessages(myrenging.logMessages());
log.debug("~~~~~~~~~~~R Script finishing execution...");
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/MyRengine.java
+50
View File
@@ -0,0 +1,50 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.rosuda.JRI.RMainLoopCallbacks;
import org.rosuda.JRI.Rengine;
public class MyRengine extends Rengine{
private StackFrame sframe=null;
Logger log = LogManager.getLogger(MyRengine.class.getName());
private RMainLoopCallbacks mlc=null;
public MyRengine(String[] args, boolean runMainLoop, RMainLoopCallbacks initialCallbacks){
super(args,runMainLoop,initialCallbacks);
this.mlc=initialCallbacks;
}
public void setStackFrame(StackFrame sframe){
log.debug("setting stack frame:"+sframe);
this.sframe=sframe;
}
public StackFrame getStackFrame(){
return this.sframe;
}
public MyRengineCallbacks getRMailLoopCallback(){
return (MyRengineCallbacks)this.mlc;
}
public void jriWriteConsole(String text, String text1){
log.debug("jriWriteConsole"+text+" text2:"+text1);
}
public void jriShowMessage(String message){
log.debug("jriShowMessage"+message);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/MyRengineCallbacks.java
+212
View File
@@ -0,0 +1,212 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.awt.FileDialog;
import java.awt.Frame;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.rosuda.JRI.RMainLoopCallbacks;
import org.rosuda.JRI.Rengine;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.rscript.RScript;
public class MyRengineCallbacks implements RMainLoopCallbacks {
private Logger log = LogManager.getLogger(MyRengineCallbacks.class.getName());
//private Vector errorMessages=new Vector();
private File outputConsole=null;
private String logmessages=null;
public String logMessages() {
return this.logmessages;
}
public void setConsoleOutput(File file) {
this.outputConsole=file;
}
public void rWriteConsole(Rengine re, String text, int oType) {
if(this.outputConsole!=null ){
try{
FileWriter fstream = new FileWriter(this.outputConsole,true);
BufferedWriter out = new BufferedWriter(fstream);
out.write(text);
out.close();
}catch(Exception e) {log.error("error:"+e.getMessage());}
// log.debug("$$$$$$$$$$ log file folder not exist, folder:"+con_out);
//}
}
if(oType==1){
log.debug("rWriteConsole():");
externalLog((MyRengine)re,text);
}
}
public void rBusy(Rengine re, int which) {
// System.out.println("MyRengine.rBusy("+which+")");
}
public String rReadConsole(Rengine re, String prompt, int addToHistory) {
// System.out.print("prompt:"+prompt);
try {
BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
String s=br.readLine();
log.debug("rReadConsole():s");
return (s==null||s.length()==0)?s:s+"\n";
} catch (Exception e) {
System.out.println("jriReadConsole exception: "+e.getMessage());
}
return null;
}
public void rShowMessage(Rengine re, String message) {
log.debug("rShowMessage():message:");
//System.out.println("MyRengine.rShowMessage \""+message+"\"");
//log.debug("Renginge instanceof:"+(re instanceof MyRengine)+" obj:"+((MyRengine)re).getStackFrame().getData());
externalLog((MyRengine)re,message);
}
//public Vector getErrorMessages(){
// return this.errorMessages;
//}
public String rChooseFile(Rengine re, int newFile) {
FileDialog fd = new FileDialog(new Frame(), (newFile==0)?"Select a file":"Select a new file", (newFile==0)?FileDialog.LOAD:FileDialog.SAVE);
fd.show();
String res=null;
if (fd.getDirectory()!=null) res=fd.getDirectory();
if (fd.getFile()!=null) res=(res==null)?fd.getFile():(res+fd.getFile());
return res;
}
public void rFlushConsole (Rengine re) {
}
public void rLoadHistory (Rengine re, String filename) {
}
public void rSaveHistory (Rengine re, String filename) {
}
public void externalLog(MyRengine mre, String message ){
if(mre.getStackFrame()!=null && mre.getStackFrame().getRscript()!=null){
RScript rs=mre.getStackFrame().getRscript();
if(rs.getError()==null){
rs.setError(message);
}else{
rs.setError(rs.getError()+"\n"+message);
}
}else{
externalLog1(mre,message);
}
}
public void externalLog1(MyRengine mre, String message ){
this.logmessages=(this.logmessages==null)?message:this.logmessages+"\n"+message;
log.debug("mre.getStackFrame():"+mre.getStackFrame());
if(mre.getStackFrame()!=null){
if(mre.getStackFrame().getTasklog()==null) {
mre.getStackFrame().setTasklog(message);
}else{
mre.getStackFrame().setTasklog(mre.getStackFrame().getTasklog()+" "+message);
}
}
//log.debug("setting log message on sframe:"+message);
//log.debug("setting log message on sframe11:"+mre.getStackFrame().getTasklog());
String root=Config.getString("log_error_folder");
Date d = new Date();
String folder =root+"r_scripts";
if (!new File(folder).isDirectory()) {
new File(folder).mkdirs();
}
String fname=root+"r_scripts"+File.separator+""+new SimpleDateFormat("dd_MM_yyyy").format(d)+".txt";
File file=new File(fname);
if (!file.exists()) {
try {
//file.createNewFile();
FileWriter fw1 = new FileWriter(file,true);
//fw1.append("Date, Time, Error Message\r\n");
fw1.flush();
fw1.close();
}catch(Exception e){
//log.error("error while creating file: Error:"+e.getMessage());
}
}
try {
boolean append = true;
FileWriter fw = new FileWriter(file,append);
//String timeformat="dd-MMM hh:mm:ss aaa ";
SimpleDateFormat timeformat=new SimpleDateFormat("h:mm a");
SimpleDateFormat dateformat=new SimpleDateFormat("dd MMM-yy");
String logmessage="";
logmessage+="Date:"+dateformat.format(new Date())+" Time:"+timeformat.format(new Date())+":";
//logmessage+="Error/Warning:"+message+"\n";
logmessage+=""+message+"\n";
fw.append(logmessage+"\r\n");
fw.flush();
fw.close();
}
catch (Exception ex) {
//log.error("Error:"+ex.getMessage());
//ClientErrorMgmt.reportError(ex,"Error in external log file writing");
ex.printStackTrace();
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/REngineWindows.java
+321
View File
@@ -0,0 +1,321 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Random;
import org.apache.commons.io.IOUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.JobExecutionException;
import org.rosuda.JRI.REXP;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.exception.ExceptionPeerUnknown;
import com.fourelementscapital.scheduler.exception.ExceptionRScriptNullError;
import com.fourelementscapital.scheduler.exception.ExceptionRscriptError;
import com.fourelementscapital.scheduler.exception.SchedulerException;
import com.fourelementscapital.scheduler.p2p.P2PService;
public abstract class REngineWindows extends AbstractRScript {
private Logger log = LogManager.getLogger(REngineWindows.class.getName());
private StackFrame stackframe=null;
protected static MyRengine re=null;
private static String rVersion=null;
private static boolean jriCompatible=false;
private static String packageVersions=null;
public REngineWindows(String name,String taskuid ) {
super(name,taskuid);
}
public synchronized void execute(StackFrame sframe) throws JobExecutionException,Exception {
log.debug("RScriptScheduledTask.execute() called");
this.stackframe=sframe;
Map<String, Object> data=sframe.getData();
Number nid=(Number)data.get("id");
String rscript=(String)data.get("rscript");
log.debug("before starting R script log id:"+sframe.getLogid());
log.debug("~~~~~~~~~~R Script started excecuting...");
if(rscript==null) { throw new JobExecutionException("Task Failed because no R script found or empty"); }
MyRengineCallbacks myrenging=new MyRengineCallbacks();
if(re==null){
re=new MyRengine(null, false, myrenging );
re.setStackFrame(sframe);
}else{
re.setStackFrame(sframe);
//re.addMainLoopCallbacks(myrenging);
}
re.getRMailLoopCallback().setConsoleOutput(null);
log.debug("R Engine initiated.");
if(rscript!=null && !rscript.equals("")){
File temp=null;
//bug fix that sink close has been put in finally method/
boolean sinked=false;
try{
BufferedReader reader1 = new BufferedReader(new FileReader(Config.getString("r_script_init")));
String line;
re.assign(".trigger_time", sframe.getTrigger_time()+"");
re.assign(".scheduler_id", nid+"");
re.assign(".machine", P2PService.getComputerName());
re.assign(".enginetype","REngine");
re.assign(".connection_ids","");
while ((line = reader1.readLine()) != null) {
if(line!=null && !line.equals("")){
REXP x=re.eval(line);
}
}
reader1.close();
log.debug("R init source file initiated.");
BufferedReader reader = new BufferedReader(new StringReader(rscript));
StringBuffer sb=new StringBuffer();
String tempDir = System.getProperty("java.io.tmpdir");
File file=new File(tempDir+File.separator+"rscript"+new SimpleDateFormat("mmddyyy_hh_mmss_").format(new Date())+new Random().nextInt()+".R");
BufferedWriter out = new BufferedWriter(new FileWriter(file));
while ((line = reader.readLine()) != null)
{
if(line!=null && !line.equals("")){
//System.out.println("--------->line:"+line);
//REXP x1= re.eval(line);
//System.out.println("Content"+x1.toString());
out.write(line+"\n");
}
}
out.close();
reader.close();
log.debug("R written to temp file :"+file.getPath());
//System.out.println("RScriptScheduledTask.execute() file:"+file.getPath());
String filepath=file.getPath().replaceAll("\\\\", "\\\\\\\\");
//sink staring here....
temp = File.createTempFile(nid.intValue()+"_"+sframe.getTrigger_time(),".log");
String ff=temp.getPath();
if(ff.contains("\\")){
ff=ff.replaceAll("\\\\", "\\\\\\\\");
}
temp.delete();
log.debug("---- deleting ff:"+ff);
String fileass="file(\""+ff+"\", open = \"wt\");";
log.debug("file assign------fileass:"+fileass);
re.eval(".console_msg_zz<-"+fileass);
re.eval("sink(.console_msg_zz)");
re.eval("sink(.console_msg_zz, type = \"message\")") ;
log.debug("----sink----");
sinked=true;
/*
String con_out=Config.getString("r_script_console_logs");
if(con_out!=null && new File(con_out).isDirectory()){
File file1=new File(con_out+nid+"_"+sframe.getTrigger_time()+".log");
re.getRMailLoopCallback().setConsoleOutput(file1);
}
*/
new SchedulerExePlanLogs(nid.intValue(),sframe.getTrigger_time()).log("REngine Script evaluation starting..",SchedulerExePlanLogs.PEER_OK_RENGINE_SCRIPT_EVAL_STARTING);
REXP x1= re.eval("source(\""+filepath+"\")");
try{
REXP genv=null;
REXP x5=re.eval("as.character(.connection_ids)");
log.debug("~~~~~~~~~~~~~~x5:"+x5.rtype);
log.debug("~~~~~~~~~~~~~~x5:asString:"+x5.asStringArray());
String ar[]=x5.asStringArray();
if(ar!=null){
for(int i=0;i<ar.length;i++){
//log.debug("~~~~~~~~~~~ ar["+i+"]:"+ar[i]);
if(ar[i]!=null && !ar[i].equals("")) {
sframe.getDbConnectionIds().add(ar[i]);
}
}
}
}catch(Exception e){
log.error("Error:"+e.getMessage());
}
log.debug("Script from temp file Evaluated");
log.debug("x1:"+x1);
if(!file.delete()){
file.deleteOnExit();
}
if(x1==null) {
//sframe.setTasklog("R Script failed. See log file for more info");
REXP x3= re.eval("as.character(.Traceback)");
if(x3.rtype==REXP.STRSXP){
String a[]=x3.asStringArray();
String err="\n----Error stack----";
if(a.length>=2){
for(int i=a.length-2;i>=0;i--){
if(!a[i].startsWith("eval.with.vis")){
String ind="\n-";
for(int ia=0;ia<i;ia++){
ind+="-";
}
ind+=">"+a[i];
err+=ind;
}
}
}
myrenging.externalLog(re, err);
String all_err="";
if(sframe.getTasklog()!=null){
all_err=err+"\n"+sframe.getTasklog();
}else{
all_err=err;
}
throw new ExceptionRscriptError("R Script Error:"+all_err);
}else{
throw new ExceptionRScriptNullError("R script:"+data.get("name")+" failed");
}
//System.out.println("x3:"+x3);
//System.out.println("x3.rtype:"+x3.rtype);
}else{
//sframe.setTasklog(x1.toString());
//executed successfully so don't store log messages
sframe.setTasklog(null);
re.setStackFrame(null);
}
//sink ending here....
updatePackageVersion();
}catch(SchedulerException se){
throw se;
}catch(Exception e){
//e.printStackTrace();
//ClientErrorMgmt.reportError(e,null);
//sframe.setTasklog("RServe is not running on "+P2PService.getComputerName());
throw new ExceptionPeerUnknown("RserveScheduleedTask: Error:"+e.getMessage());
} finally{
if(sinked && re!=null){
re.eval("sink(type = \"message\")");
re.eval("sink()");
re.eval("close(.console_msg_zz)");
}
if(temp!=null && temp.exists()){
FileInputStream inputStream = new FileInputStream(temp);
try{
sframe.setConsole_message(IOUtils.toString(inputStream));
}catch(Exception e){
log.error("error while accessing the file, Error:"+e.getMessage());
}finally{
inputStream.close();
}
temp.delete();
}
}
}
if(re==null){
re.end();
re=null;
}
//setMessages(myrenging.logMessages());
log.debug("~~~~~~~~~~~R Script finishing execution...");
}
public static void updatePackageVersion(){
//updates R Package version.
if(Config.getString("p2p.ignorejri")!=null && Config.getString("p2p.ignorejri").equalsIgnoreCase("true")){
}else{
if(re==null){
re=new MyRengine(null, false, new MyRengineCallbacks() );
}
//REXP x=re.eval("necessaryRemote <- as.character(read.table(file='//10.153.64.10/Public/Libs/mandatoryPackages.txt')[,1]);");
//REXP x=re.eval("necessaryRemote2 <- as.character(read.table(file='//10.153.64.10/Public/Libs/mandatoryExternalPackages.txt')[,1]);");
REXP x=re.eval("necessaryRemote2 <- c(as.character(read.table(file='//10.153.64.10/Public/Libs/mandatoryExternalPackages.txt')[,1]),as.character(read.table(file='//10.153.64.10/Public/Libs/mandatory4ecapPackages.txt')[,1]))");
x=re.eval("IP = installed.packages(); ");
x=re.eval("paste(as.character(IP[, 'Package'])[as.character(IP[, 'Package']) %in% necessaryRemote2], as.character(IP[, 'Version'])[as.character(IP[, 'Package']) %in% necessaryRemote2], sep=\"=\");");
if(x!=null){
String st1[]=x.asStringArray();
String result="PEER="+P2PService.getComputerName();
for(int i=0;i<st1.length;i++){
result+="|"+st1[i];
}
packageVersions=result;
}
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/RServeUnix.java
+115
View File
@@ -0,0 +1,115 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.util.Date;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.rosuda.REngine.REXP;
import org.rosuda.REngine.Rserve.RConnection;
import com.fourelementscapital.scheduler.exception.ExceptionRScriptNullError;
import com.fourelementscapital.scheduler.exception.ExceptionRServeUnixFailure;
import com.fourelementscapital.scheduler.exception.ExceptionRscriptError;
import com.fourelementscapital.scheduler.rserve.RServeConnectionPool;
import com.fourelementscapital.scheduler.rserve.RServeSession;
public abstract class RServeUnix extends AbstractRServe {
public RServeUnix(String name, String uid) {
super(name, uid);
}
//private RConnection c=null;
//private RServeSession rs=null;
private Logger log = LogManager.getLogger(RServeUnix.class.getName());
public RServeSession getRServeSession(int nid, String name) throws Exception {
log.debug("before opening session----:script name:"+name);
RServeSession rs=null;
RConnection c=null;
try{
String startup_rinit=null;
rs=RServeConnectionPool.getRSession(startup_rinit);
}catch(Exception e){
throw new ExceptionRServeUnixFailure("Not able to get RSession from ConnectionPool!, Error:"+e.getMessage());
}
try{
log.debug("+++++++attaches Rconnection from RServeSession....+++");
c=(rs.getRconnection()!=null) ? rs.getRconnection(): rs.getRsession().attach();
rs.setRconnection(c);
}catch(Exception e){
throw new ExceptionRServeUnixFailure("Not able to attach connection to RSession!, Error:"+e.getMessage());
}
if(c!=null){
log.debug("c:"+c.isConnected());
rs.setScriptname(name);
rs.setRunning(true);
rs.setScheduler_id(nid);
rs.setTrigger_time(super.stackframe.getTrigger_time());
rs.setStarted_time(new Date().getTime());
rs.setThread(Thread.currentThread());
new SchedulerExePlanLogs(nid,super.stackframe.getTrigger_time()).log("RConnection process id:"+rs.getProcessid()+" Current Hits:"+rs.getNoexecutions(),SchedulerExePlanLogs.PEER_OK_UNIX_RSERVE_CAPTURE_PROCESSINFO);
}
return rs;
}
@Override
public void evalScript(RServeSession rs,String rscript)
throws Exception {
RConnection c=rs.getRconnection();
c.assign(".tmp.",rscript);
log.debug("----script:----");
log.debug(rscript);
log.debug("-----:----");
REXP r = c.parseAndEval("try(eval(parse(text=.tmp.)),silent=TRUE)");
if (r!=null && r.inherits("try-error")) {
log.debug("r.asString:"+r.asString());
String msg=r.asString();
throw new ExceptionRscriptError("Error in R Script: "+msg);
}
if(r==null) {
throw new ExceptionRScriptNullError("Rscript Executed but unknown error message returned by RServe");
}else{
this.stackframe.setTasklog(null);
}
}
@Override
public void closeRconnection(RServeSession rs) throws Exception {
try{
rs.finished(rs.getRconnection());
RServeConnectionPool.done(rs);
log.debug("-------attaches Rconnection from RServeSession....----");
}catch(Exception e){
if(rs.getKillmessage()!=null){
throw new ExceptionRServeUnixFailure(rs.getKillmessage());
}else{
throw new ExceptionRServeUnixFailure("Not able to detach RSession!, Error:"+e.getMessage());
}
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/RServeWindows.java
+150
View File
@@ -0,0 +1,150 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Random;
import java.util.TreeMap;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.rosuda.REngine.REXP;
import org.rosuda.REngine.Rserve.RConnection;
import com.fourelementscapital.scheduler.exception.ExceptionRScriptNullError;
import com.fourelementscapital.scheduler.exception.ExceptionRServeWindowsFailure;
import com.fourelementscapital.scheduler.exception.ExceptionRscriptError;
import com.fourelementscapital.scheduler.rserve.RServeSession;
public abstract class RServeWindows extends AbstractRServe{
//private RConnection c=null;
private Process p=null;
private Logger log = LogManager.getLogger(RServeWindows.class.getName());
public RServeWindows(String name, String uid) {
super(name, uid);
}
public RServeSession getRServeSession(int nid, String name) throws Exception {
RServeSession rs=null;
this.p=Runtime.getRuntime().exec("c:\\rJava\\rserve1.bat");
int count=0;
TreeMap error=new TreeMap();
RConnection c=null;
while(c==null && count<20) {
try{c=new RConnection();}catch(Exception e){
//errorMessage+=""
String msg=e.getMessage();
if(error.keySet().contains(msg)){
int cnt=(Integer)error.get(msg);
cnt++;error.put(msg, cnt);
}else{
error.put(msg, 1);
}
}
try{ Thread.sleep(500); }catch(Exception e){}
count++;
}
if(c==null){
String msg="";
for(Iterator i=error.keySet().iterator();i.hasNext();){
String ky=(String)i.next();
int cnt=(Integer)error.get(ky);
msg+=ky+ ((cnt>1)?"("+cnt+")":"");
}
throw new ExceptionRServeWindowsFailure("Couldn't start RServ Connection! Error MSG:"+msg);
}else{
rs=new RServeSession();
rs.setRconnection(c);
}
return rs;
}
public void evalScript(RServeSession rs,String rscript) throws Exception {
RConnection c=rs.getRconnection();
String tempDir = System.getProperty("java.io.tmpdir");
File file=new File(tempDir+File.separator+"rscript"+new SimpleDateFormat("mmddyyy_hh_mmss_").format(new Date())+new Random().nextInt()+".R");
BufferedWriter out = new BufferedWriter(new FileWriter(file));
BufferedReader reader = new BufferedReader(new StringReader(rscript));
String line;
while ((line = reader.readLine()) != null) {
if(line!=null && !line.equals("")){
//System.out.println("--------->line:"+line);
//REXP x1= re.eval(line);
//System.out.println("Content"+x1.toString());
out.write(line+"\n");
}
}
out.close();
reader.close();
log.debug("R written to temp file :"+file.getPath());
//System.out.println("RScriptScheduledTask.execute() file:"+file.getPath());
String filepath=file.getPath().replaceAll("\\\\", "\\\\\\\\");
REXP x1= c.parseAndEval("try(source(\""+filepath+"\"),silent=TRUE)");
log.debug("Script from temp file Evaluated");
log.debug("x1:"+x1);
log.debug("error messages:"+c.getLastError());
if (x1!=null && x1.inherits("try-error") ) {
//throw new Exception("Error in R Script: "+x1.asString());
throw new ExceptionRscriptError(x1.asString());
}
if(x1==null) {
//sframe.setTasklog("R Script failed. See log file for more info");
//throw new Exception("R script:"+data.get("name")+" failed");
throw new ExceptionRScriptNullError("Rscript Executed but unknown error message returned by RServe");
}else{
this.stackframe.setTasklog(null);
}
}
public void closeRconnection(RServeSession rs) throws Exception {
RConnection c=rs.getRconnection();
if(c!=null){
c.shutdown();
if(this.p!=null){
p.destroy();
}
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/ScheduledTask.java
+217
View File
@@ -0,0 +1,217 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.JobExecutionException;
import com.fourelementscapital.db.SchedulerDB;
import com.fourelementscapital.db.vo.FlexiField;
import com.fourelementscapital.scheduler.exception.SchedulerException;
public abstract class ScheduledTask {
public static String FIELD_DEPENDENCY_IDS="dependentids";
public static String FIELD_DEPENDENCY_CHECKTIME="dependentchecktime";
public static String FIELD_DEPENDENCY_SUCCESS="onsuccess";
public static String FIELD_DEPENDENCY_FAIL="onfail";
public static String FIELD_DEPENDENCY_TIMEOUT="ontimeout";
public static String FIELD_CONCURRENT_EXEC="concurrent_execution";
private String name=null;
private String uniqueid=null;
private Vector<ScheduledTaskField> flexifields=new Vector<ScheduledTaskField>();
public static String EXCECUTION_SUCCESS="success";
public static String EXCECUTION_FAIL="fail";
public static String DEPENDENCY_TIMEOUT="dep_timeout";
public static String EXCECUTION_OVERLAPPED="overlapped";
public static String EXCECUTION_WARNING="warning";
public static String TIMOUT_WARNING="timeout";
public static String TASK_QUEUED="queued";
public static String TASK_EVENT_CALL_EXP_ID_VARIABLE=".caller_id";
public static String TASK_EVENT_CALL_EXP_TRIGGERTIME_VARIABLE=".caller_trigger_time";
public static String TASK_EVENT_CALL_EXP_ERRORMSG_VARIABLE=".caller_error_message";
public static final int PEER_ASSOCIATION_HISTORY_ADDED=1;
public static final int PEER_ASSOCIATION_HISTORY_REMOVED=0;
public static final int PEER_ASSOCIATION_HISTORY_PEER_ACTIVE=1;
public static final int PEER_ASSOCIATION_HISTORY_PEER_NOACTIVE=0;
private Logger log = LogManager.getLogger(ScheduledTask.class.getName());
public ScheduledTask(String name, String uid) {
this.name=name;
this.uniqueid=uid;
addNameField();
}
public String getName() {
return this.name;
}
public String getUniqueid() {
return this.uniqueid;
}
public List<ScheduledTaskField> listFormFields(){
return this.flexifields;
}
private void addNameField(){
SchedulerDB sdb=SchedulerDB.getSchedulerDB();
try{
sdb.connectDB();
List<Map> folders=sdb.listOfFolders(getUniqueid());
ScheduledTaskField f1=new ScheduledTaskField();
f1.setShortname("folder_id");
f1.setFieldlabel("Folder");
f1.setFieldtype(FlexiField.TYPE_DROP_DOWNLIST);
String foption="";
for(Map record: folders){
String line=record.get("id")+"~"+record.get("folder_name");
foption+=foption.equals("")?line:"|"+line;
}
f1.setFieldoptions(foption);
this.flexifields.add(f1);
}catch(Exception e){
log.error("error:"+e.getMessage());
}finally{
try{
sdb.closeDB();
}catch(Exception e1){
log.error("couldn't close sdb");
}
}
ScheduledTaskField fname=new ScheduledTaskField();
fname.setShortname("name");
fname.setFieldtype(FlexiField.TYPE_TEXTBOX);
fname.setFieldlabel("Name");
ScheduledTaskField depids=new ScheduledTaskField();
depids.setShortname(FIELD_DEPENDENCY_IDS);
depids.setFieldtype(FlexiField.TYPE_TEXTBOX);
depids.setFieldlabel("Dependent Task IDs");
depids.setFineprint("Runs only when those were successfully executed. Multiple IDs separated by comma. For example: 243,140,245 (Leave empty if not applicable)");
depids.setPlacementform("dependency");
ScheduledTaskField deptime=new ScheduledTaskField();
deptime.setShortname(FIELD_DEPENDENCY_CHECKTIME);
deptime.setFieldtype(FlexiField.TYPE_TEXTBOX);
deptime.setFieldlabel("Dependents Recency");
deptime.setFineprint(" Recency of all dependent task in Minutes (applicable only if Dependent task IDs entered)");
deptime.setPlacementform("dependency");
this.flexifields.add(fname);
this.flexifields.add(depids);
this.flexifields.add(deptime);
ScheduledTaskField success=new ScheduledTaskField();
success.setShortname(FIELD_DEPENDENCY_SUCCESS);
success.setFieldtype(FlexiField.TYPE_FREETEXT);
success.setFieldlabel("On Success");
//success.setFineprint(" Execute tasks when the status is success");
success.setPlacementform("events");
this.flexifields.add(success);
ScheduledTaskField fail=new ScheduledTaskField();
fail.setShortname(FIELD_DEPENDENCY_FAIL);
fail.setFieldtype(FlexiField.TYPE_FREETEXT);
fail.setFieldlabel("On Error/Fail");
//fail.setFineprint(" Execute tasks when there is an error");
fail.setPlacementform("events");
this.flexifields.add(fail);
ScheduledTaskField timeout=new ScheduledTaskField();
timeout.setShortname(FIELD_DEPENDENCY_TIMEOUT);
timeout.setFieldtype(FlexiField.TYPE_FREETEXT);
timeout.setFieldlabel("On Timeout");
//timeout.setFineprint(" Execute tasks when there is timeout.");
timeout.setPlacementform("events");
this.flexifields.add(timeout);
ScheduledTaskField overlap=new ScheduledTaskField();
overlap.setShortname(FIELD_CONCURRENT_EXEC);
overlap.setFieldtype(FlexiField.TYPE_DROP_DOWNLIST);
overlap.setPlacementform("concur");
overlap.setFieldlabel("Number of Concurrent Executions");
overlap.setFineprint(" >1 will allow task overlapping");
overlap.setFieldoptions("2|3|4|5|6|7|8|9|10");
overlap.setDropdowninitial("1");
this.flexifields.add(overlap);
}
protected void removeAllFields(){
this.flexifields.removeAllElements();
addNameField();
}
public void addFormFields(Collection<ScheduledTaskField> fields ) throws Exception{
for(Iterator<ScheduledTaskField> ffi=fields.iterator();ffi.hasNext();){
ScheduledTaskField ff=ffi.next();
if(!flexifields.contains(ff)){
this.flexifields.add(ff);
}else{
throw new Exception("Shortname is unique, Duplicate shortname are not allowed, or name shortname is already reserved.");
}
}
}
public abstract void execute(StackFrame sframe) throws JobExecutionException,SchedulerException,Exception ;
/*
public void setAddhocListener(String triggername,ScheduledTaskJobAdhocListener listner) {
this.adhoclisteners.put(triggername,listner);
}
public ScheduledTaskJobAdhocListener getAddhocListener(String triggername) {
return this.adhoclisteners.get(triggername);
}
public ScheduledTaskJobAdhocListener removeAddhocListener(String triggername) {
return this.adhoclisteners.remove(triggername);
}
*/
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/ScheduledTaskField.java
+75
View File
@@ -0,0 +1,75 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import com.fourelementscapital.db.vo.FlexiField;
import com.fourelementscapital.scheduler.pluggin.PlugginData;
public class ScheduledTaskField extends FlexiField {
private String shortname;
private PlugginData pluggindata;
private String fineprint;
private String placementform="general"; //default is general
public String getPlacementform() {
return placementform;
}
public void setPlacementform(String placementform) {
this.placementform = placementform;
}
public String getFineprint() {
return fineprint;
}
public void setFineprint(String fineprint) {
this.fineprint = fineprint;
}
public PlugginData getPluggindata() {
return pluggindata;
}
public void setPluggindata(PlugginData pluggindata) {
this.pluggindata = pluggindata;
}
public String getShortname() {
return shortname;
}
public void setShortname(String shortname) {
this.shortname = shortname;
}
public boolean equals(Object other) {
if(other instanceof ScheduledTaskField){
if(((ScheduledTaskField)other).getShortname().equals(this.getShortname())){
return true;
}else{
return false;
}
}
return false;
}
public String toString(){
return getShortname()+"."+getFieldlabel();
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/SchedulerExePlanLogs.java
+202
View File
@@ -0,0 +1,202 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.MissingResourceException;
import java.util.ResourceBundle;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.fourelementscapital.db.SchedulerDB;
import com.fourelementscapital.scheduler.p2p.P2PService;
public class SchedulerExePlanLogs {
public static final int SERVER_OK_FIXEDPEER=2000;
public static final int SERVER_OK_REMOVED_FROM_PQUEUE=2001;
public static final int SERVER_OK_RECEIVED_STATUS_FROM_PEER=2002;
public static final int SERVER_OK_PEER_ACCEPTED_TASK=2003;
public static final int SERVER_ERROR_EXEC_TIMEDOUT=3000;
//3001,3002 & 3003 needs to mapped on flow chart
public static final int SERVER_ERROR_QUEUE_KILLED_BYUSER=3001;
public static final int SERVER_ERROR_PEER_CRASHED_REMOVED_QUEUE=3002;
public static final int SERVER_ERROR_PEER_NORESP_REMOVED_QUEUE=3003;
public static final int SERVER_ERROR_WHILE_ADDING_QUEUE=3004;
public static final int SERVER_ERROR_ALARM_SENT=3005;
//3006 needs to mapped on flow chart
public static final int SERVER_ERROR_REMOVING_QUEUE_BASEDON_LOG=3006;
public static final int SERVER_ERROR_DEPENDENCY_TIMEDOUT=3007;
//3008 & 3009 needs to mapped on flow chart
public static final int SERVER_ERROR_REMOVING_QUEUE_PEER_NO_RESPONSE=3008;
public static final int SERVER_ERROR_REMOVE_QUEUE_NOTRUNNING_INPEER=3009;
public static final int SERVER_WARNING_OVERLAPPED=4000;
public static final int SERVER_WARNING_BOUNCEDTASK_FROMPEER=4002;
public static final int PEER_OK_BLOOMBERG_DOWNLOAD=5000;
public static final int PEER_OK_UNIX_RSERVE_CAPTURE_PROCESSINFO=5001;
public static final int PEER_OK_RSERVE_SCRIPT_EVAL_STARTING=5002;
public static final int PEER_OK_RENGINE_SCRIPT_EVAL_STARTING=5003;
public static final int PEER_OK_ADDED_PEER_QUEUE=5004;
public static final int PEER_OK_EXECUTION_STARTING=5005;
public static final int PEER_OK_EXECUTION_COMPLETED_WITH_NOEXCEPTION=5006;
public static final int PEER_OK_PEER_RECEVED_TASK=5007;
public static final int PEER_OK_RESPOND_TASKCOMPLETED_WITHSTATUS=5008;
public static final int PEER_OK_RESPOND_RSERVE_SESSIONSTARTED=5009;
public static final int PEER_ERROR_WHILE_RECEIVINGTASK=6000;
public static final int PEER_ERROR_EXECUTION_FAILURE=6001;
public static final int PEER_ERROR_EXECUTION_COMPLETED_WITH_EXCEPTION=6002;
public static final int PEER_WARNING_NOROOM_TO_EXEC=7000;
public static final int IGNORE_CODE=0;
private long trigger_time;
private int scheduler_id;
private Logger log = LogManager.getLogger(SchedulerExePlanLogs.class.getName());
public SchedulerExePlanLogs(int scheduler_id, long trigger_time){
this.scheduler_id=scheduler_id;
this.trigger_time=trigger_time;
}
/**
* @deprecated
* @param message
*/
public void log(String message, int repcode) {
SchedulerDB sdb=SchedulerDB.getSchedulerDB();
try{
//calculates different and adds in server's time
sdb.connectDB();
sdb.addSchedulerExeLogs(this.scheduler_id, this.trigger_time, new Date(), message,repcode, P2PService.getComputerName());
}catch(Exception e){
log.error("error:"+e.getMessage());
}finally{
try{
sdb.closeDB();
}catch(Exception e1){
log.error("error:"+e1.getMessage());
}
}
}
public void log(String key,Map data,int repCode) {
try{
String mess=getMessage(key, data);
log(mess,repCode);
}catch(Exception e){
//throw e;
log.error("Error:"+e.getMessage());
}
//System.out.print("Message:"+mess);
}
/*
public void log(String key,Map data,SchedulerDB sdb) {
try{
String mess=getMessage(key, data);
log(mess,sdb);
}catch(Exception e){
log.error("Error:"+e.getMessage());
}
//System.out.print("Message:"+mess);
}
*/
/**
* @deprecated
* @param message
* @param sdb
*/
public void log(String message,SchedulerDB sdb,int repCode) {
try{
//calculates different and adds in server's time
sdb.addSchedulerExeLogs(this.scheduler_id, this.trigger_time, new Date(), message,repCode, P2PService.getComputerName());
}catch(Exception e){
log.error("error:"+e.getMessage());
}
}
private String getMessage(String key, Map data) throws Exception {
ResourceBundle rb=ResourceBundle.getBundle("com.fe.scheduler.logmessages");
String val=key;
try{
val=rb.getString(key);
}catch(MissingResourceException miex){
val=key+" !key:not_found!";
}
//if(data!=null && data.size()>0){
val=replaceNewValues(val,data!=null?data:new HashMap());
//}
return val;
}
private String replaceNewValues(final String log,
final Map values){
final StringBuffer sb = new StringBuffer();
final Pattern pattern =
Pattern.compile("\\[(.*?)\\]", Pattern.DOTALL);
final Matcher matcher = pattern.matcher(log);
while(matcher.find()){
final String key = matcher.group(1);
if( values.get(key)!=null){
final String replacement = values.get(key).toString();
if(replacement == null){
matcher.appendReplacement(sb, "");
}else{
matcher.appendReplacement(sb, replacement);
}
}else{
matcher.appendReplacement(sb, "");
}
}
matcher.appendTail(sb);
return sb.toString();
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/StackFrame.java
+236
View File
@@ -0,0 +1,236 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import java.util.Map;
import java.util.Vector;
import com.fourelementscapital.scheduler.p2p.MessageBean;
import com.fourelementscapital.scheduler.rscript.RScript;
public class StackFrame {
public ScheduledTask task = null;
public Map data = null;
private long trigger_time=0;
private long nexttrigger_time=0;
private long started_time=0;
private String status=null;
private RScript rscript=null;
private String replyTo;
private String invokedby="";
private Vector dbConnectionIds=new Vector();
private String executed_code=null;
private String console_message=null;
private String queue_uid=null;
public String getQueue_uid() {
return queue_uid;
}
public void setQueue_uid(String queue_uid) {
this.queue_uid = queue_uid;
}
public String getConsole_message() {
return console_message;
}
public void setConsole_message(String console_message) {
this.console_message = console_message;
}
public String getExecuted_code() {
return executed_code;
}
public void setExecuted_code(String executed_code) {
this.executed_code = executed_code;
}
public Vector getDbConnectionIds() {
return dbConnectionIds;
}
public String getInvokedby() {
return invokedby;
}
public void setInvokedby(String invokedby) {
this.invokedby = invokedby;
}
public String getReplyTo() {
return replyTo;
}
public void setReplyTo(String replyTo) {
this.replyTo = replyTo;
}
public RScript getRscript() {
return rscript;
}
public void setRscript(RScript rscript) {
this.rscript = rscript;
}
public String getStatus() {
return status;
}
public void setStatus(String status) {
this.status = status;
}
private String tasklog=null;
private boolean dependencyfailed=false;
public boolean isDependencyfailed() {
return dependencyfailed;
}
public void setDependencyfailed(boolean dependencyfailed) {
this.dependencyfailed = dependencyfailed;
}
private StackFrameCallBack callback=null;
private MessageBean mbean=null;
/**
* @deprecated
*
*/
public MessageBean getMbean() {
return mbean;
}
/**
* @deprecated
* @param mbean
*/
public void setMbean(MessageBean mbean) {
this.mbean = mbean;
}
public void addCallBack(StackFrameCallBack callback){
this.callback=callback;
}
public StackFrameCallBack getCallBack(){
return this.callback;
}
public String getTasklog() {
return tasklog;
}
public void setTasklog(String tasklog) {
this.tasklog = tasklog;
}
public long getNexttrigger_time() {
return nexttrigger_time;
}
public void setNexttrigger_time(long nexttrigger_time) {
this.nexttrigger_time = nexttrigger_time;
}
public long getTrigger_time() {
return trigger_time;
}
public void setTrigger_time(long trigger_time) {
this.trigger_time = trigger_time;
}
private int logid=0;
public int getLogid() {
return logid;
}
public void setLogid(int logid) {
this.logid = logid;
}
public StackFrame(ScheduledTask task, Map data) {
this.task = task;
this.data = data;
}
public ScheduledTask getTask(){
return this.task;
}
public Map getData(){
return this.data;
}
public long getStarted_time() {
return started_time;
}
public void setStarted_time(long started_time) {
this.started_time = started_time;
}
public boolean equals(Object other) {
if (!(other instanceof StackFrame) ) return false;
final StackFrame other1 = (StackFrame) other;
if(this.rscript!=null){
if(this.rscript!=null && other1.rscript!=null && this.rscript.getUid().equals(other1.rscript.getUid())){
return true;
}else{
return false;
}
}else{
//validation for all types
return equals1(other);
}
}
public boolean equals1(Object other) {
if(this.data==null) return false;
//if ( !(other instanceof StackFrame) ) return false;
final StackFrame other1 = (StackFrame) other;
if(other1.data==null) return false;
if(other1.data.get("id")!=null && this.data.get("id")!=null){
Number onid=(Number)other1.data.get("id");
Number nid=(Number)this.data.get("id");
if (onid.intValue()==nid.intValue()) return true;
}
if(other1.data.get("script_id")!=null && this.data.get("script_id")!=null){
Number onid=(Number)other1.data.get("script_id");
Number nid=(Number)this.data.get("script_id");
if (onid.intValue()==nid.intValue()) return true;
}
return false;
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/engines/StackFrameCallBack.java
+18
View File
@@ -0,0 +1,18 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import com.fourelementscapital.scheduler.exception.SchedulerException;
public interface StackFrameCallBack {
public void callBack(StackFrame sframe, String status,SchedulerException se);
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/BBDownloadScheduledTask.java
+29
View File
@@ -0,0 +1,29 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import com.fourelementscapital.scheduler.engines.BloombergDownload;
public class BBDownloadScheduledTask extends BloombergDownload {
public BBDownloadScheduledTask(String name, String taskuid) {
super(name,taskuid);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/DirectRServeExecuteRUnix.java
+20
View File
@@ -0,0 +1,20 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import com.fourelementscapital.scheduler.engines.ExecuteRUnix;
public class DirectRServeExecuteRUnix extends ExecuteRUnix {
public DirectRServeExecuteRUnix(String name,String taskuid ) {
super(name,taskuid);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/ExecuteRWindowsHighPriority.java
+20
View File
@@ -0,0 +1,20 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import com.fourelementscapital.scheduler.engines.ExecuteRWindows;
public class ExecuteRWindowsHighPriority extends ExecuteRWindows {
public ExecuteRWindowsHighPriority(String name, String taskuid) {
super(name, taskuid);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/REngineScriptTask.java
+28
View File
@@ -0,0 +1,28 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import com.fourelementscapital.scheduler.engines.ExecuteRWindows;
public class REngineScriptTask extends ExecuteRWindows{
public REngineScriptTask(String name, String taskuid) {
super(name, taskuid);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/RScriptScheduledTask.java
+213
View File
@@ -0,0 +1,213 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Random;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.quartz.JobExecutionException;
import org.rosuda.JRI.REXP;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.engines.MyRengine;
import com.fourelementscapital.scheduler.engines.MyRengineCallbacks;
import com.fourelementscapital.scheduler.engines.REngineWindows;
import com.fourelementscapital.scheduler.engines.StackFrame;
public class RScriptScheduledTask extends REngineWindows {
protected static MyRengine re=null;
private Logger log = LogManager.getLogger(RScriptScheduledTask.class.getName());
private StackFrame stackframe=null;
public RScriptScheduledTask(String name, String taskuid) {
//super("High Priority (R Engine)","rscript");
super(name,taskuid);
}
private static String rVersion=null;
private static boolean jriCompatible=false;
private static String packageVersions=null;
/**
* @deprecated
*/
public static String getRVersion(){
if(rVersion==null){
//getRVersionFirstTime();
}
return rVersion;
}
/**
* @deprecated
*/
public static String getRPackageVersion(){
if(packageVersions==null){
//updatePackageVersion();
}
return packageVersions;
}
/**
* @deprecated
*/
public static boolean getJRICompatible(){
if(rVersion==null){
//getRVersionFirstTime();
}
return jriCompatible;
}
/**
* @deprecated
*/
public static void updatePackageVersion(){
//updates R Package version.
}
/**
* @deprecated
*/
public void executeScript(StackFrame sframe) throws JobExecutionException,Exception {
if(Config.getString("p2p.ignorejri")!=null && Config.getString("p2p.ignorejri").equalsIgnoreCase("true")){
throw new Exception("JRI is not enalbed");
}
//log.debug("RScriptScheduledTask.execute() called");
this.stackframe=sframe;
Map<String, Object> data=sframe.getData();
String rscript=(String)data.get("script");
Number restart=(Number)data.get("restart");
Number script_id=(Number)data.get("script_id");
if(rscript==null) { throw new JobExecutionException("Task Failed because no R script found or empty"); }
MyRengineCallbacks myrenging=new MyRengineCallbacks();
if(re!=null && restart.intValue()==1){
re.end();
//log.debug("restarting R session....");
//re.interrupt();
log.debug("interupting R thread....");
re=null;
try{
Thread.sleep(2000);
}catch(Exception e){ }
}
if(re==null){
log.debug("Rengine is null, creating one.......");
re=new MyRengine(null, false, myrenging );
re.setStackFrame(sframe);
}else{
re.setStackFrame(sframe);
//re.addMainLoopCallbacks(myrenging);
}
if(rscript!=null && !rscript.equals("")){
try{
BufferedReader reader = new BufferedReader(new StringReader(rscript));
StringBuffer sb=new StringBuffer();
String tempDir = System.getProperty("java.io.tmpdir");
File file=new File(tempDir+File.separator+"rscript"+new SimpleDateFormat("mmddyyy_hh_mmss_").format(new Date())+new Random().nextInt()+".R");
String line;
BufferedWriter out = new BufferedWriter(new FileWriter(file));
while ((line = reader.readLine()) != null) {
if(line!=null && !line.equals("")){
out.write(line+"\n");
}
}
out.close();
reader.close();
String filepath=file.getPath().replaceAll("\\\\", "\\\\\\\\");
REXP x1= re.eval("source(\""+filepath+"\")");
log.debug("x1:"+x1);
file.delete();
if(x1==null) {
//sframe.setTasklog("R Script failed. See log file for more info");
throw new Exception("R script:"+data.get("name")+" failed");
}else{
sframe.setTasklog(null);
re.setStackFrame(null);
}
}catch(Exception e){
//e.printStackTrace();
//ClientErrorMgmt.reportError(e,null);
throw e;
}
}
//setMessages(myrenging.logMessages());
log.debug("~~~~~~~~~~~R Script finishing execution...");
}
public static String codeInjectConcatenate(String default_param, String rscript, String injected_code) throws Exception {
String finalcode="";
if(default_param!=null && !default_param.equals("")) {
finalcode+="## Default Header Code ##\n";
finalcode+=default_param+"\n";
finalcode+="## --------------------## \n";
}
if(injected_code!=null && !injected_code.equals("")) {
finalcode+="\n\n## Injected Code ##\n";
finalcode+=injected_code+"\n";
finalcode+="## --------------## \n";
}
if(rscript!=null && !rscript.equals("")) {
finalcode+=rscript+"\n";
}
return finalcode;
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/RServeScheduledTask.java
+31
View File
@@ -0,0 +1,31 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import com.fourelementscapital.scheduler.engines.RServeWindows;
public class RServeScheduledTask extends RServeWindows {
public RServeScheduledTask(String name,String taskuid ) {
super(name,taskuid);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/group/RServeUnixTask.java
+28
View File
@@ -0,0 +1,28 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.group;
import com.fourelementscapital.scheduler.engines.RServeUnix;
public class RServeUnixTask extends RServeUnix {
public static final String ENGINE_NAME="rscript4rserveunix";
public static final String ENGINE_EXECUTER_UNIX_NAME="direct_script_unix";
public RServeUnixTask(String name,String taskuid ) {
super(name,taskuid);
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/pluggin/PlugginData.java
+62
View File
@@ -0,0 +1,62 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.pluggin;
public class PlugginData {
private String jsnew;
private String jscreate;
private String jsfetch;
public String getJsfetch() {
return jsfetch;
}
public void setJsfetch(String jsfetch) {
this.jsfetch = jsfetch;
}
//private String jsmodify;
//private String jsdelete;
private String fieldreference;
public String getFieldreference() {
return fieldreference;
}
public void setFieldreference(String fieldreference) {
this.fieldreference = fieldreference;
}
public String getJsnew() {
return jsnew;
}
public void setJsnew(String jsnew) {
this.jsnew = jsnew;
}
public String getJscreate() {
return jscreate;
}
public void setJscreate(String jscreate) {
this.jscreate = jscreate;
}
/*
public String getJsmodify() {
return jsmodify;
}
public void setJsmodify(String jsmodify) {
this.jsmodify = jsmodify;
}
public String getJsdelete() {
return jsdelete;
}
public void setJsdelete(String jsdelete) {
this.jsdelete = jsdelete;
}
*/
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/pluggin/PlugginInterface.java
+29
View File
@@ -0,0 +1,29 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.pluggin;
import java.util.Map;
import javax.servlet.http.HttpServletRequest;
public interface PlugginInterface {
public int addAction(Map data, HttpServletRequest request) throws Exception ;
public int updateAction(int scheduler_id, Map data,HttpServletRequest request) throws Exception ;
public Map fetchData(int record_id, HttpServletRequest request) throws Exception ;
public void deleteAction(int scheduler_id, HttpServletRequest request) throws Exception ;
//public void upadteAction(int recordid, Map data,HttpServletRequest request) throws Exception;
//public boolean deleteAction(int recordid,HttpServletRequest request) throws Exception;
public PlugginData getPlugginData();
public String getText(Map data);
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/pluggin/SchedulerBloombergPlugin.java
+429
View File
@@ -0,0 +1,429 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.pluggin;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.TreeMap;
import java.util.Vector;
import javax.servlet.http.HttpServletRequest;
import com.fourelementscapital.db.BBSyncDB;
import com.fourelementscapital.db.SchedulerDB;
import com.fourelementscapital.db.vo.BBSyncTrigger;
public class SchedulerBloombergPlugin implements PlugginInterface {
public static String PLUGGIN_IN="bloomberg_pluggin";
public int addAction(Map data,HttpServletRequest request) throws Exception {
try{
//BBSyncMgmt bbsyncmgmt=new BBSyncMgmt(request);
//System.out.println("data:"+data);
String name=(String)data.get("qdata.name");
String mkt_securities=(String)data.get("mkt_securities");
boolean mkt_securities1=false;
if(mkt_securities.equalsIgnoreCase("true")){
mkt_securities1=true;
}
String filtervalue=(String)data.get("filtervalue");
String date1s=(String)data.get("date1s");
String date2s=(String)data.get("date2s");
String number=(String)data.get("number");
String fieldids=(String)data.get("fieldids");
StringTokenizer token=new StringTokenizer(fieldids,",");
Vector fieldids1=new Vector();
while(token.hasMoreTokens()){
fieldids1.add(token.nextToken());
}
String contracts=(String)data.get("contracts");
String marketsector=(String)data.get("marketsector");
//System.out.println("name:"+name+" mkt_securities:"+ mkt_securities1+" filtervalue:"+filtervalue+" date1s:"+ date1s+" date2s:"+date2s+" number:"+ number+" fieldids:"+ fieldids+" contracts:"+ contracts+" marketsector:"+marketsector);
//int downloadid=bbsyncmgmt.saveSchedule2(0, name, mkt_securities1, filtervalue, date1s, date2s, number, fieldids1, contracts, null, null, null, marketsector, null);
int downloadid=saveSchedule2(0, name, mkt_securities1, filtervalue, date1s, date2s, number, fieldids1, contracts, null, null, null, marketsector, null);
return downloadid;
}catch(Exception e){
throw e;
}
}
private int saveSchedule2(
int id,
String name,
boolean mkt_secdb,
String dateoption,
String datefrom,
String dateto,
String datenumber,
Vector fields,
String contracts,
BBSyncTrigger daily,
BBSyncTrigger weekly,
BBSyncTrigger monthly,
String marketsector,
String timezone
) throws Exception {
try{
//log.debug("saveSchedule called");
//String fields="";
//if(blbfields.size()==dbfields.size()){
// for(int i=0;i<blbfields.size();i++){
// String f=blbfields.get(i)+"~"+dbfields.get(i);
// fields+=(i>0)?"|"+f:f;
// }
//}
BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
BBSyncTrigger trigobj=new BBSyncTrigger();
if(daily!=null && weekly==null && monthly==null){
trigobj=daily;
trigobj.setSynctype(BBSyncTrigger.HOURLY);
}
if(daily==null && weekly!=null && monthly==null){
trigobj=weekly;
trigobj.setSynctype(BBSyncTrigger.WEEKLY);
}
if(daily==null && weekly==null && monthly!=null){
trigobj=monthly;
trigobj.setSynctype(BBSyncTrigger.MONTHLY);
}
//log.debug("trigobj:"+trigobj);
try{
syncdb.connectDB();
//log.debug("after connected");
java.sql.Date d1=null;
java.sql.Date d2=null;
//log.debug("before parsing:"+datefrom);
SimpleDateFormat sf=new SimpleDateFormat("DD/mm/yyyy");
if(datefrom!=null){
//log.debug("parseing:"+sf.parse(datefrom));
d1=new java.sql.Date((sf.parse(datefrom)).getTime());
//new java.sql.Date(datefrom.getTime());
}
if(dateto!=null){
d2=new java.sql.Date((sf.parse(dateto)).getTime());
}
int datenumber1=0;
try{
datenumber1=Integer.parseInt(datenumber);
}catch(Exception e){}
int nid=syncdb.saveSchedule(
id,
name,
mkt_secdb+"",
dateoption,
d1,
d2,
datenumber1,
fields,
contracts,
trigobj,
marketsector,
timezone
);
syncdb.closeDB();
return nid;
}catch(Exception e){
//ClientErrorMgmt.reportError(e, "Couldn't save scheduling data");
//e.printStackTrace();
throw e;
}
}catch(Exception e){
//ClientErrorMgmt.reportError(e, null);
throw e;
}
}
public int updateAction(int scheduler_id,Map data,HttpServletRequest request) throws Exception {
try{
//BBSyncMgmt bbsyncmgmt=new BBSyncMgmt(request);
SchedulerDB sdb=SchedulerDB.getSchedulerDB();
sdb.connectDB();
Map sdata=sdb.getScheduler(scheduler_id);
sdb.closeDB();
String d_id=(String)sdata.get(getPlugginData().getFieldreference());
if(d_id!=null){
//System.out.println("data:"+data);
String name=(String)data.get("qdata.name");
String mkt_securities=(String)data.get("mkt_securities");
boolean mkt_securities1=false;
if(mkt_securities.equalsIgnoreCase("true")){
mkt_securities1=true;
}
String filtervalue=(String)data.get("filtervalue");
String date1s=(String)data.get("date1s");
String date2s=(String)data.get("date2s");
String number=(String)data.get("number");
String fieldids=(String)data.get("fieldids");
StringTokenizer token=new StringTokenizer(fieldids,",");
Vector fieldids1=new Vector();
while(token.hasMoreTokens()){
fieldids1.add(token.nextToken());
}
String contracts=(String)data.get("contracts");
String marketsector=(String)data.get("marketsector");
//System.out.println("name:"+name+" mkt_securities:"+ mkt_securities1+" filtervalue:"+filtervalue+" date1s:"+ date1s+" date2s:"+date2s+" number:"+ number+" fieldids:"+ fieldids+" contracts:"+ contracts+" marketsector:"+marketsector);
//int downloadid=bbsyncmgmt.saveSchedule2(Integer.parseInt(d_id), name, mkt_securities1, filtervalue, date1s, date2s, number, fieldids1, contracts, null, null, null, marketsector, null);
int downloadid=saveSchedule2(Integer.parseInt(d_id), name, mkt_securities1, filtervalue, date1s, date2s, number, fieldids1, contracts, null, null, null, marketsector, null);
return downloadid;
}else{
throw new Exception("Problem in updating Bloomberg data for this task");
}
}catch(Exception e){
throw e;
}
}
//public boolean deleteAction(int recordid,HttpServletRequest request) throws Exception {
//
// return true;
//}
public PlugginData getPlugginData() {
PlugginData pdata=new PlugginData();
pdata.setJsnew("bbpl_jsNew");
pdata.setJscreate("bbpl_jsCreate");
pdata.setJsfetch("bbpl_jsFetch");
//pdata.setJsmodify("bbpl_jsUpdate");
pdata.setFieldreference("download_id");
return pdata;
}
public Map fetchData(int recordid, HttpServletRequest request) throws Exception {
//BBSyncMgmt bbsyncmgmt=new BBSyncMgmt(request);
//return bbsyncmgmt.getDownloadQuery(recordid);
return getDownloadQuery(recordid);
}
private Map getDownloadQuery(long id) throws Exception {
try{
BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
syncdb.connectDB();
Map m=syncdb.getDownloadQuery(id);
//System.out.println("BBSyncMgmt.getDownloadQuery():m:"+m);
Object b=m.get("date_from");
if(b!=null){
m.put("date_from",new SimpleDateFormat("dd/MM/yyyy").format((Date)b));
}
Object b1=m.get("date_to");
if(b1!=null){
m.put("date_to",new SimpleDateFormat("dd/MM/yyyy").format((Date)b1));
}
if(m.get("trigger_time")!=null){
try{
Date d=new SimpleDateFormat("HH:mm").parse((String)m.get("trigger_time"));
Calendar c=Calendar.getInstance();
c.setTime(d);
m.put("trigger_hour",c.get(Calendar.HOUR_OF_DAY));
m.put("trigger_minute",c.get(Calendar.MINUTE));
}catch(Exception e){}
}
if(m.get("fields")!=null){
String fields1=(String)m.get("fields");
StringTokenizer st=new StringTokenizer(fields1,"|");
TreeMap fields=new TreeMap();
while(st.hasMoreTokens()){
String cfield=st.nextToken();
StringTokenizer st1=new StringTokenizer(cfield,"~");
if(st1.countTokens()==2){
String ky=st1.nextToken();
String value=st1.nextToken();
fields.put(ky,value);
}
}
m.put("fieldsmap",fields);
}
//put logs if there is
/*
Vector logs=syncdb.getSyncLogs((int)id);
for(Iterator<Map>i= logs.iterator();i.hasNext();) {
Map record=i.next();
try{
Timestamp s=(Timestamp)record.get("start_time");
Timestamp e=(Timestamp)record.get("end_time");
if(s!=null && e!=null){
SimpleDateFormat df=new SimpleDateFormat("dd MMM, yyyy");
SimpleDateFormat tf=new SimpleDateFormat("h:mm a");
record.put("start_time",tf.format(s));
record.put("end_time",tf.format(e));
record.put("start_date",df.format(s));
SimpleDateFormat dateFormat =new SimpleDateFormat("mm:ss");
dateFormat.setTimeZone(TimeZone.getTimeZone("GMT"));
long diff=e.getTime()-s.getTime();
record.put("duration", dateFormat.format(new Date(diff)));
}
}catch(Exception e){
}
}
m.put("logs",logs );
*/
syncdb.closeDB();
return m;
}catch(Exception e){
//ClientErrorMgmt.reportError(e, null);
throw e;
}
}
public void deleteAction(int scheduler_id, HttpServletRequest request) throws Exception {
SchedulerDB sdb=SchedulerDB.getSchedulerDB();
sdb.connectDB();
Map sdata=sdb.getScheduler(scheduler_id);
sdb.closeDB();
String d_id=(String)sdata.get(getPlugginData().getFieldreference());
if(d_id!=null){
//BBSyncMgmt bbsyncmgmt=new BBSyncMgmt(request);
//bbsyncmgmt.deleteQuery(Integer.parseInt(d_id));
deleteQuery(Integer.parseInt(d_id));
} else{
throw new Exception("Problem in deleting Bloomberg data for this task");
}
}
private Map deleteQuery(int id) throws Exception {
try{
BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
syncdb.connectDB();
syncdb.deleteQuery(id);
//new BloombergDownloadJob().scheduleDownloadQueries();
Map rtn=loadFirst();
rtn.remove("field_mapping");
syncdb.closeDB();
return rtn;
}catch(Exception e){
//e.printStackTrace();
//ClientErrorMgmt.reportError(e, "id:"+id);
throw e;
}
}
private Map loadFirst() throws Exception {
BBSyncDB syncdb=BBSyncDB.getBBSyncDB();
syncdb.connectDB();
TreeMap rtn=new TreeMap();
Vector v=syncdb.listAll();
Vector fields=syncdb.getFieldMapping();
for(int i=0;i<v.size();i++){
Map m=(Map)v.get(i);
Object b=m.get("last_executed");
if(b instanceof java.sql.Timestamp && b!=null ){
String recent=new SimpleDateFormat("d MMM,yy HH:mm").format((Date)b);
//m.put("name",((m.get("name")!=null)?m.get("name"):"")+ " <small>(Last Executed:"+recent+")</small>") ;
m.put("last_executed", recent) ;
}
}
syncdb.closeDB();
rtn.put("schedules",v);
rtn.put("field_mapping", fields);
return rtn;
}
public String getText(Map data) {
String rtn="Name:"+data.get("qdata.name")+"\n";
rtn+="Is_securities:"+data.get("mkt_securities")+"\n";
rtn+="Filter:"+data.get("filtervalue")+"\n";
rtn+="Date_option1:"+data.get("date1s")+"\n";
rtn+="Date_option2:"+data.get("date2s")+"\n";
rtn+="Date_option3:"+data.get("number")+"\n";
rtn+="Fieldids:"+data.get("fieldids")+"\n";
rtn+="Tickers:"+data.get("contracts")+"\n";
rtn+="Sector:"+data.get("marketsector")+"\n";
return rtn;
}
//public void upadteAction(int recordid, Map data,HttpServletRequest request)throws Exception {
//
//}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/pluggin/SchedulerPlugginRegister.java
+25
View File
@@ -0,0 +1,25 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.pluggin;
public class SchedulerPlugginRegister {
public static PlugginInterface getPluggin(String plugginid){
if(plugginid.equalsIgnoreCase(SchedulerBloombergPlugin.PLUGGIN_IN)){
return new SchedulerBloombergPlugin();
}else{
return null;
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/rscript/RScript.java
+220
View File
@@ -0,0 +1,220 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.rscript;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
import org.rosuda.JRI.REXP;
import com.fourelementscapital.scheduler.p2p.msg.MessageHandler;
public class RScript extends MessageHandler {
private String uid;
private String script;
private REXP result;
private String resultXML;
private String error;
private boolean executing;
private String taskuid;
private String peer;
private String executeAt;
private Long queued_time;
private String uniquename;
private String requesthost;
private long delay;
private Date startedtime;
private String resultJSON;
//private String queuefriendlytime;
public String getResultJSON() {
return resultJSON;
}
public void setResultJSON(String resultJSON) {
this.resultJSON = resultJSON;
}
public Date getStartedtime() {
return startedtime;
}
public void setStartedtime(Date startedtime) {
this.startedtime = startedtime;
}
public String getRequesthost() {
return requesthost;
}
public long getDelay() {
return delay;
}
public void setDelay(long delay) {
this.delay = delay;
}
public void setRequesthost(String requesthost) {
this.requesthost = requesthost;
}
public String getUniquename() {
return uniquename;
}
public void setUniquename(String uniquename) {
this.uniquename = uniquename;
}
public Long getQueued_time() {
return queued_time;
}
public void setQueued_time(Long queued_time) {
this.queued_time = queued_time;
}
public String getExecuteAt() {
return executeAt;
}
public void setExecuteAt(String executeAt) {
this.executeAt = executeAt;
}
public String getPeer() {
return peer;
}
public void setPeer(String peer) {
this.peer = peer;
}
public String getTaskuid() {
return taskuid;
}
public void setTaskuid(String taskuid) {
this.taskuid = taskuid;
}
public boolean isExecuting() {
return executing;
}
public void setExecuting(boolean executing) {
this.executing = executing;
}
public RScript() {
this.uid=UUID.randomUUID().toString();
}
public String getUid() {
return uid;
}
public void setUid(String uid) {
this.uid = uid;
}
public String getScript() {
return script;
}
public void setScript(String script) {
this.script = script;
}
public REXP getResult() {
return result;
}
public void setResult(REXP result) {
this.result = result;
}
public String getError() {
return error;
}
public void setError(String error) {
this.error = error;
}
public String getResultXML() {
return resultXML;
}
public void setResultXML(String resultXML) {
this.resultXML = resultXML;
}
public boolean equals(Object o) {
RScript other=(RScript)o;
if(other!=null && other.getUid().equals(this.getUid()) ){
return true;
} else{
return false;
}
}
public int compareTo(RScript o) {
return this.queued_time<o.queued_time?-1:
this.queued_time>o.queued_time?1:0;
}
public Map executeAtDestination() {
return null;
}
public String getScriptPreview(){
if(this.script.trim().length()>=100){
return this.script.trim().substring(0,99)+"....";
}else{
return this.script.trim();
}
}
public String getQueuefriendlytime(){
try{
SimpleDateFormat sdf=new SimpleDateFormat("dd-MMM HH:mm:ss");
Date d=new Date(this.queued_time);
d.setTime(this.queued_time);
return sdf.format(d);
}catch(Exception e){
return e.getMessage();
}
}
@Override
public void onSendingFailed() {
// TODO Auto-generated method stub
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/rscript/RScriptAsyncListenerImpl.java
+189
View File
@@ -0,0 +1,189 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.rscript;
import java.io.File;
import java.io.FileWriter;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.jcs.JCS;
import org.apache.jcs.engine.behavior.IElementAttributes;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import com.fourelementscapital.scheduler.config.Config;
import com.fourelementscapital.scheduler.error.ClientError;
public class RScriptAsyncListenerImpl implements RScriptListener{
private Logger log = LogManager.getLogger(RScriptAsyncListenerImpl.class.getName());
private String peer=null;
private String result=null;
//private HttpSession session=null;
private JCS cache;
private IElementAttributes att;
private Date started=null;
public static String ALIVE="alive";
public RScriptAsyncListenerImpl(JCS cache, IElementAttributes att) throws Exception {
//this.session=session;
this.cache=cache;
this.att=att;
}
public void onScriptSent(RScript rscript, String peer) throws Exception {
// TODO Auto-generated method stub
this.peer=peer;
log.debug("+++++ Script started....started:"+rscript.getScript()+" peer:"+peer);
if(this.cache!=null){
this.cache.put(rscript.getUid(),ALIVE,this.att);
}
this.started=new Date();
}
public String getPeer(){
return this.peer;
}
public String getResult(){
return this.result;
}
public void onScriptFinished(RScript rscript, String peer, String result, String status)throws Exception {
this.peer=peer;
this.result=result;
//if(this.session!=null){
// this.session.setAttribute(rscript.getUid(),this);
//}
log.debug("~~~~~~ rscript.getUid():"+rscript.getUid()+" result:"+result+" this.att:"+this.att);
if(this.cache!=null){
this.cache.put(rscript.getUid(),result,this.att);
}
String logmessage="";
//SimpleDateFormat timeformat=new SimpleDateFormat("HH:mm:ss SSS");
//SimpleDateFormat dateformat=new SimpleDateFormat("dd MMM-yy HH:mm:ss SSS");
//logmessage+="UID:"+rscript.getUid()+"\tStatus:"+status+"\tPeer:"+rscript.getPeer()+"\tStarted:"+dateformat.format(this.started)+"\tEnded:"+timeformat.format(new Date())+"\tScript:"+readFirst(rscript.getScript(),50);
String st_time=this.started!=null ? this.started.getTime()+"":"";
logmessage+=status+"\t"+rscript.getPeer()+"\t"+rscript.getQueued_time()+"\t"+st_time+"\t"+new Date().getTime()+"\t"+rscript.getRequesthost()+"\t"+rscript.getUniquename();
log.debug("~~~~~~ Script started....rscript.getError():"+rscript.getError());
if(status.equalsIgnoreCase("fail") && rscript.getError()!=null){
logmessage+="\tError:"+readFirst(rscript.getError(), 100);
}
log(logmessage);
}
public void onScriptTimedOut(RScript rscript) throws Exception {
this.result="<?xml version=\"1.0\"?><output status=\"TimedOut\">Script kicked out from the queue because scheduler couldn't find peer within the time limit</output>";
//if(this.session!=null){
// this.session.setAttribute(rscript.getUid(),this);
//}
if(this.cache!=null){
this.cache.put(rscript.getUid(),this.result,this.att);
}
SimpleDateFormat timeformat=new SimpleDateFormat("HH:mm SSS");
SimpleDateFormat dateformat=new SimpleDateFormat("dd MMM-yy HH:mm:ss SSS");
//String logmessage="UID:"+rscript.getUid()+"\tStatus:TimedOut\tAt:"+dateformat.format(new Date())+"\tScript:"+readFirst(rscript.getScript(),50);
String logmessage="TimedOut\t"+""+"\t"+rscript.getQueued_time()+"\t\t"+new Date().getTime()+"\t"+rscript.getRequesthost()+"\t"+rscript.getUniquename();
log(logmessage);
}
protected static String readFirst(String result, int no_char) {
String rtn=result.length()>no_char ?result.substring(0,no_char)+"....": result;
rtn=rtn.trim();
rtn=rtn.replaceAll("\n", " ");
rtn=rtn.replaceAll("\r", " ");
return rtn;
}
protected static void log(String logmessage){
Date d = new Date();
String root;
if(Config.getString("log_executeR_folder")!=null && !Config.getString("log_executeR_folder").equals("") ){
root=Config.getString("log_executeR_folder");
}else{
root=Config.getString("log_error_folder");
}
String folder =root+"direct_scripts";
if (!new File(folder).isDirectory()) {
new File(folder).mkdirs();
}
String fname=root+"direct_scripts"+File.separator+""+new SimpleDateFormat("dd_MM_yyyy").format(d)+".txt";
File file=new File(fname);
if (!file.exists()) {
try {
//file.createNewFile();
FileWriter fw1 = new FileWriter(file,true);
//fw1.append("Date, Time, Error Message\r\n");
fw1.flush();
fw1.close();
}catch(Exception e){
//log.error("error while creating file: Error:"+e.getMessage());
}
}
try {
boolean append = true;
FileWriter fw = new FileWriter(file,append);
//String timeformat="dd-MMM hh:mm:ss aaa ";
//logmessage+="Error/Warning:"+message+"\n";
//logmessage+=""+message+"\n";
fw.append(logmessage+"\r\n");
fw.flush();
fw.close();
}
catch (Exception ex) {
//log.error("Error:"+ex.getMessage());
ClientError.reportError(ex,"Error in external log file writing");
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/rscript/RScriptAsyncWebsocket.java
+136
View File
@@ -0,0 +1,136 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.rscript;
import java.nio.CharBuffer;
import java.util.Date;
import org.apache.catalina.websocket.MessageInbound;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
public class RScriptAsyncWebsocket implements RScriptListener{
Logger log = LogManager.getLogger(RScriptAsyncWebsocket.class.getName());
private MessageInbound messageinbound;
private String uniqueid;
private Date started=null;
public RScriptAsyncWebsocket(String uid,MessageInbound mib){
this.uniqueid=uid;
this.messageinbound=mib;
}
@Override
public void onScriptSent(RScript rscript, String peer) throws Exception {
this.started=new Date();
}
@Override
public void onScriptFinished(RScript rscript, String peer, String result,
String status) throws Exception {
log.debug("script :"+rscript.getUid()+" finished, thread:"+Thread.currentThread().getName() );
String logmessage="";
String st_time=this.started!=null ? this.started.getTime()+"":"";
logmessage+=status+"\t"+rscript.getPeer()+"\t"+rscript.getQueued_time()+"\t"+st_time+"\t"+new Date().getTime()+"\t"+rscript.getRequesthost()+"\t"+rscript.getUniquename();
//log.debug("~~~~~~ Script started....rscript.getError():"+rscript.getError());
if(status.equalsIgnoreCase("fail") && rscript.getError()!=null){
logmessage+="\tError:"+RScriptAsyncListenerImpl.readFirst(rscript.getError(), 100);
}
RScriptAsyncListenerImpl.log(logmessage);
long duration=new Date().getTime()-this.started.getTime();
String rtn=this.uniqueid+"~~"+result;
try{
this.messageinbound.getWsOutbound().writeTextMessage(CharBuffer.wrap(rtn));
this.messageinbound.getWsOutbound().flush();
}catch(Exception e){
log.error("Error, :onScriptFinished:"+e.getMessage());
}
/*
Future fu=IOExcecutors.threadpool.submit(
new Callable<String>() {
public String call(){
String logmessage="";
String st_time=this.started!=null ? this.started.getTime()+"":"";
logmessage+=status+"\t"+rscript.getPeer()+"\t"+rscript.getQueued_time()+"\t"+st_time+"\t"+new Date().getTime()+"\t"+rscript.getRequesthost()+"\t"+rscript.getUniquename();
//log.debug("~~~~~~ Script started....rscript.getError():"+rscript.getError());
if(status.equalsIgnoreCase("fail") && rscript.getError()!=null){
logmessage+="\tError:"+RScriptAsyncListenerImpl.readFirst(rscript.getError(), 100);
}
RScriptAsyncListenerImpl.log(logmessage);
long duration=new Date().getTime()-this.started.getTime();
String rtn=this.uid+"~~"+result;
try{
this.messageinbound.getWsOutbound().writeTextMessage(CharBuffer.wrap(rtn));
this.messageinbound.getWsOutbound().flush();
}catch(Exception e){
log.error("Error, :onScriptFinished:"+e.getMessage());
}
return "";
}
private RScript rscript;
private String peer;
private String result;
private String status;
private Date started;
private MessageInbound messageinbound;
private String uid;
public Callable<String> init(String u,MessageInbound mib, RScript rs, String p, String r, String st, Date std){
this.uid=u;
this.messageinbound=mib;
this.rscript=rs;
this.peer=p;
this.result=r;
this.status=st;
this.started=std;
return this;
}
}.init(this.uniqueid,this.messageinbound,rscript, peer,result,status,this.started)
);
*/
}
@Override
public void onScriptTimedOut(RScript rscript) throws Exception {
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/rscript/RScriptListener.java
+20
View File
@@ -0,0 +1,20 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.rscript;
import com.fourelementscapital.scheduler.rscript.RScript;
public interface RScriptListener {
public void onScriptSent(RScript rscript, String peer) throws Exception ;
public void onScriptFinished(RScript rscript, String peer, String result, String status) throws Exception;
public void onScriptTimedOut(RScript rscript) throws Exception;
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/rserve/RServeConnectionPool.java
+174
View File
@@ -0,0 +1,174 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.rserve;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.rosuda.REngine.Rserve.RConnection;
import com.fourelementscapital.scheduler.p2p.peer.PeerSpecificConfigurations;
public class RServeConnectionPool {
private static ConcurrentLinkedQueue<RServeSession> sessions=new ConcurrentLinkedQueue();
private static ConcurrentLinkedQueue<RServeSession> sessInUse=new ConcurrentLinkedQueue();
private static final Semaphore sessLock=new Semaphore(1,true);
private static long TIMEOUT_MS=2000;
/**
* gets RSession from connection pool, if no live connection available, it creates new and store it in pool
* @return
* @throws Exception
*/
private static void acquireLock(){
try{
Date start=new Date();
RServeConnectionPool.sessLock.tryAcquire(TIMEOUT_MS, TimeUnit.MILLISECONDS);
}catch(Exception e){
LogManager.getLogger(RServeConnectionPool.class.getName()).error("acquireLock() error:"+e.getMessage());
}
}
private static void releaseLock(){
try{
RServeConnectionPool.sessLock.release();
//log.debug("....releasing lock: thread:"+Thread.currentThread().getId());
}catch(Exception e){
LogManager.getLogger(RServeConnectionPool.class.getName()).error("releaseLock() error:"+e.getMessage());
}
}
public static synchronized RServeSession getRSession(String rinit) throws Exception {
//new RConnection()
acquireLock();
try{
RServeSession rtn=null;
if(RServeConnectionPool.sessions.size()>0){
//rtn=(RServeSession)RServeConnectionPool.sessions.lastElement();
rtn=(RServeSession)RServeConnectionPool.sessions.poll();
//RServeConnectionPool.sessions.remove(rtn);
RServeConnectionPool.sessInUse.add(rtn);
}else{
int count=0;
RConnection c=null;
while(c==null && count<10) {
try{c= new RConnection();}catch(Exception e){ ;}
try{ Thread.sleep(200); }catch(Exception e){}
count++;
}
if(c!=null){
rtn=new RServeSession();
c.assign(".r_session_id",rtn.getUid());
//loading rinit at the first time of opening connection.
if(rinit!=null && !rinit.trim().equals("")){
try{
c.parseAndEval(rinit);
}catch(Exception e){
throw new Exception("Not able to load Rnint at RServeConnectionPool.getRsession(), for rinit: "+rinit);
}
}
//c.eval(".r_scripts=NULL");
int processid=c.eval("Sys.getpid()").asInteger();
rtn.setProcessid(processid);
rtn.setRconnection(c);
//rtn.setRsession(c.detach());
RServeConnectionPool.sessInUse.add(rtn);
}
}
if(rtn!=null){
rtn.setNoexecutions(rtn.getNoexecutions()+1);
}
return rtn;
}finally{
releaseLock();
}
}
public static List<RServeSession> getAllSessions(){
ArrayList rtn=new ArrayList();
for(RServeSession rs: RServeConnectionPool.sessions){
rtn.add(rs);
}
for(RServeSession rs: RServeConnectionPool.sessInUse){
rtn.add(rs);
}
return rtn;
}
public static void remove(RServeSession rs){
if(RServeConnectionPool.sessInUse.contains(rs))RServeConnectionPool.sessInUse.remove(rs);
if(RServeConnectionPool.sessions.contains(rs))RServeConnectionPool.sessions.remove(rs);
}
public static synchronized void done(RServeSession rs) throws Exception {
acquireLock();
try{
int rtn=0;
String val=PeerSpecificConfigurations.getProperties().getProperty(PeerSpecificConfigurations.KEY_MAX_EXEC_SESSION);
if(val!=null && !val.trim().equals("")){
rtn=Integer.parseInt(val);
}
if(rtn>0 && rs.getNoexecutions()>=rtn){
Process p=Runtime.getRuntime().exec("kill -9 " + rs.getProcessid());
processOutput(p);
remove(rs);
}else{
RServeConnectionPool.sessInUse.remove(rs);
RServeConnectionPool.sessions.add(rs);
}
}finally{
releaseLock();
}
}
private static void processOutput(Process p) throws Exception {
InputStream inputStream = p.getInputStream();
InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
String line;
while ((line = bufferedReader.readLine()) != null)
{
//System.out.println("RServeSessionQuery. console_output:"+line);
}
}
}
scheduler_libraries/lib-scheduler-engines/src/main/java/com/fourelementscapital/scheduler/rserve/RServeSession.java
+165
View File
@@ -0,0 +1,165 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.rserve;
import org.rosuda.REngine.Rserve.RConnection;
import org.rosuda.REngine.Rserve.RSession;
import com.fourelementscapital.fileutils.RandomString;
public class RServeSession {
private RSession rsession=null;
private RConnection rconnection=null;
private String scriptname;
private String execute_r_uid=null;
private boolean running=false;
private int scheduler_id;
private long trigger_time;
private Thread thread;
private int processid=0;
private String killmessage=null;
private int noexecutions=0;
public int getNoexecutions() {
return noexecutions;
}
public void setNoexecutions(int noexecutions) {
this.noexecutions = noexecutions;
}
public String getKillmessage() {
return killmessage;
}
public void setKillmessage(String killmessage) {
this.killmessage = killmessage;
}
private String uid=null;
public RServeSession(){
this.uid=RandomString.getString(40);
}
public String getUid() {
return uid;
}
public String getScriptname() {
return scriptname;
}
public void setScriptname(String scriptname) {
this.scriptname = scriptname;
}
public void finished(RConnection rc) throws Exception {
try{
this.rsession=rc.detach();
this.running=false;
this.scheduler_id=0;
this.trigger_time=0;
this.started_time=0;
this.thread=null;
this.rconnection=null;
this.scriptname=null;
}catch(Exception e){
throw e;
}
}
public int getProcessid() {
return processid;
}
public void setProcessid(int processid) {
this.processid = processid;
}
public RConnection getRconnection() {
return rconnection;
}
public void setRconnection(RConnection rconnection) {
this.rconnection = rconnection;
}
public Thread getThread() {
return thread;
}
public void setThread(Thread thread) {
this.thread = thread;
}
public long getTrigger_time() {
return trigger_time;
}
public void setTrigger_time(long trigger_time) {
this.trigger_time = trigger_time;
}
public RSession getRsession() {
return rsession;
}
public void setRsession(RSession rsession) {
this.rsession = rsession;
}
public boolean isRunning() {
return running;
}
public void setRunning(boolean running) {
this.running = running;
}
public int getScheduler_id() {
return scheduler_id;
}
public void setScheduler_id(int scheduler_id) {
this.scheduler_id = scheduler_id;
}
public long getStarted_time() {
return started_time;
}
public void setStarted_time(long started_time) {
this.started_time = started_time;
}
public String getExecute_r_uid() {
return execute_r_uid;
}
public void setExecute_r_uid(String execute_r_uid) {
this.execute_r_uid = execute_r_uid;
}
private long started_time;
public boolean equals(Object other) {
RServeSession rs=(RServeSession)other;
if(rs.uid.equals(this.uid)){
return true;
}else{
return false;
}
}
}
scheduler_libraries/lib-scheduler-engines/src/test/java/com/fourelementscapital/scheduler/engines/AppTest.java
+47
View File
@@ -0,0 +1,47 @@
/******************************************************************************
*
* Copyright: Intellectual Property of Four Elements Capital Pte Ltd, Singapore.
* All rights reserved.
*
******************************************************************************/
package com.fourelementscapital.scheduler.engines;
import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;
/**
* Unit test for App
*/
public class AppTest extends TestCase {
/**
* Create the test case
*
* @param testName name of the test case
*/
public AppTest( String testName )
{
super( testName );
}
/**
* @return the suite of tests being tested
*/
public static Test suite()
{
return new TestSuite( AppTest.class );
}
/**
* Empty Test
*/
public void testApp()
{
assertTrue( true );
}
}