package com.ximple.eofms.jobs;
|
|
import java.math.BigDecimal;
|
import java.nio.ByteBuffer;
|
import java.nio.ByteOrder;
|
import java.sql.ResultSet;
|
import java.sql.SQLException;
|
import java.sql.Statement;
|
import java.sql.Types;
|
import java.util.Date;
|
import java.io.IOException;
|
|
import org.apache.commons.collections.OrderedMap;
|
import org.apache.commons.collections.OrderedMapIterator;
|
import org.apache.commons.collections.map.LinkedMap;
|
import org.apache.commons.logging.Log;
|
import org.apache.commons.logging.LogFactory;
|
import org.quartz.JobDetail;
|
import org.quartz.JobExecutionContext;
|
import org.quartz.JobExecutionException;
|
|
import com.vividsolutions.jts.geom.GeometryFactory;
|
|
import oracle.jdbc.OracleConnection;
|
import oracle.jdbc.OracleResultSet;
|
import oracle.sql.ARRAY;
|
import oracle.sql.BLOB;
|
|
import com.ximple.eofms.util.BinConverter;
|
import com.ximple.eofms.util.ByteArrayCompressor;
|
import com.ximple.io.dgn7.ComplexElement;
|
import com.ximple.io.dgn7.Dgn7fileException;
|
import com.ximple.io.dgn7.Element;
|
import com.ximple.io.dgn7.ElementType;
|
import com.ximple.io.dgn7.IElementHandler;
|
import com.ximple.util.PrintfFormat;
|
|
/**
|
*
|
*/
|
public class OracleConvertDgn2ShpJob extends AbstractOracleDatabaseJob
|
{
|
static Log logger = LogFactory.getLog(OracleConvertDgn2ShpJob.class);
|
|
private static final int FETCHSIZE = 30;
|
private static final int BATCHSIZE = 25;
|
private static final int COMMITSIZE = 20;
|
|
class Pair
|
{
|
Object first;
|
Object second;
|
|
public Pair(Object first, Object second)
|
{
|
this.first = first;
|
this.second = second;
|
}
|
}
|
|
GeometryFactory _geomFactory = new GeometryFactory();
|
|
public Log getLogger()
|
{
|
return logger;
|
}
|
|
protected AbstractOracleJobContext prepareJobContext(String filterPath)
|
{
|
return new OracleConvertJobContext(filterPath);
|
}
|
|
public void execute(JobExecutionContext context) throws JobExecutionException
|
{
|
// Every job has its own job detail
|
JobDetail jobDetail = context.getJobDetail();
|
|
// The name is defined in the job definition
|
String jobName = jobDetail.getName();
|
|
// Log the time the job started
|
logger.info(jobName + " fired at " + new Date());
|
extractJobConfiguration(jobDetail);
|
|
OracleConvertJobContext jobContext = (OracleConvertJobContext) prepareJobContext(_filterPath);
|
jobContext.setConnectionInfo(_oracleHost, _oraclePort, _oracleInstance);
|
jobContext.setLogin(_username, _password);
|
jobContext.setShapeData(_dataPath);
|
jobContext.setExecutionContext(context);
|
|
try
|
{
|
copyConnectivity(jobContext);
|
exetcuteConvert(jobContext, _orgSchema, _dataPath);
|
//exetcuteConvert(jobContext, "CMMS_SPATIALDB", _dataPath);
|
|
//close all open filewriter instance
|
jobContext.closeFeatureWrite();
|
} catch (SQLException e)
|
{
|
logger.warn(e.getMessage(), e);
|
throw new JobExecutionException("Database error.", e);
|
} catch (IOException ex) {
|
ex.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
|
}
|
|
}
|
|
//Connectivity複製一個版本，在查詢電流方向時用來比對OMS資料庫的電器連接性(Connectivity)
|
private void copyConnectivity(OracleConvertJobContext jobContext) throws SQLException
|
{
|
OracleConnection connection = jobContext.getOracleConnection() ;
|
Statement stmt = connection.createStatement();
|
stmt.execute(OracleConvertJobContext.TRUNCATE_CONNECTIVITY_WEBCHECK);
|
stmt.execute(OracleConvertJobContext.COPY_CONNECTIVITY_TO_WEBCHECK);
|
}
|
|
private void exetcuteConvert(OracleConvertJobContext jobContext,
|
String querySchema, String dataPath) throws SQLException
|
{
|
int order = 0;
|
OrderedMap map = getBlobStorageList(jobContext.getOracleConnection(), querySchema, "SD$SPACENODES"
|
, null);
|
|
logger.info("begin convert job:[" + map.size() + "]:testmode=" + _testMode);
|
|
int total = map.size(); //spacenodes count
|
int step = total / 100;
|
int current = 0;
|
|
//jobContext.startTransaction();
|
jobContext.getExecutionContext().put("ConvertDgn2ShpJobProgress", 0);
|
for (OrderedMapIterator it = map.orderedMapIterator(); it.hasNext();)
|
{
|
it.next();
|
|
Pair pair = (Pair) it.getValue();
|
String tableSrc = (String) pair.first;
|
|
logger.info("begin convert:[" + order + "]-" + tableSrc);
|
queryIgsetElement(jobContext, querySchema, tableSrc);
|
|
order++;
|
|
if (_testMode)
|
{
|
if ((_testCount < 0) || (order >= _testCount))
|
break;
|
}
|
|
if ((order % COMMITSIZE) == 0)
|
{
|
// OracleConnection connection = jobContext.getOracleConnection();
|
// connection.commitTransaction();
|
jobContext.commitTransaction();
|
//jobContext.startTransaction();
|
System.gc();
|
}
|
|
int now = order % step;
|
if (now != current)
|
{
|
current = now;
|
jobContext.getExecutionContext().put("ConvertDgn2ShpJobProgress", current);
|
|
}
|
}
|
jobContext.getExecutionContext().put("ConvertDgn2ShpJobProgress", 100);
|
|
jobContext.commitTransaction();
|
|
logger.info("end convert job:[" + order + "]");
|
System.gc();
|
}
|
|
protected OrderedMap getBlobStorageList(OracleConnection connection, String schemaSrc, String tableSrc,
|
OrderedMap orderedMap) throws SQLException
|
{
|
if (orderedMap == null)
|
orderedMap = new LinkedMap(99);
|
String fetchStmtFmt = "SELECT SNID, SPACETABLE FROM \"%s\".\"%s\"";
|
PrintfFormat spf = new PrintfFormat(fetchStmtFmt);
|
String fetchStmt = spf.sprintf(new Object[]{schemaSrc, tableSrc});
|
Statement stmt = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
|
stmt.setFetchSize(FETCHSIZE);
|
|
ResultSet rs = stmt.executeQuery(fetchStmt);
|
|
while (rs.next())
|
{
|
int size = rs.getMetaData().getColumnCount();
|
Object[] values = new Object[size];
|
|
for (int i = 0; i < size; i++)
|
{
|
values[i] = rs.getObject(i + 1);
|
}
|
|
Integer key = ((BigDecimal) values[0]).intValue();
|
String name = (String) values[1];
|
|
Pair pair = (Pair) orderedMap.get(key);
|
if (pair == null)
|
orderedMap.put(key, new Pair(name, null));
|
else
|
pair.first = name;
|
}
|
|
rs.close();
|
stmt.close();
|
|
return orderedMap;
|
}
|
|
protected OrderedMap getRawFormatStorageList(OracleConnection connection, String schemaSrc, String tableSrc,
|
OrderedMap orderedMap) throws SQLException
|
{
|
if (orderedMap == null)
|
orderedMap = new LinkedMap(99);
|
String fetchStmtFmt = "SELECT RNID, SPACETABLE FROM \"%s\".\"%s\"";
|
PrintfFormat spf = new PrintfFormat(fetchStmtFmt);
|
String fetchStmt = spf.sprintf(new Object[]{schemaSrc, tableSrc});
|
Statement stmt = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
|
stmt.setFetchSize(FETCHSIZE);
|
|
ResultSet rs = stmt.executeQuery(fetchStmt);
|
|
while (rs.next())
|
{
|
int size = rs.getMetaData().getColumnCount();
|
Object[] values = new Object[size];
|
|
for (int i = 0; i < size; i++)
|
{
|
values[i] = rs.getObject(i + 1);
|
}
|
|
Integer key = ((BigDecimal) values[0]).intValue();
|
String name = (String) values[1];
|
|
Pair pair = (Pair) orderedMap.get(key);
|
if (pair == null)
|
orderedMap.put(key, new Pair(null, name));
|
else
|
pair.second = name;
|
}
|
|
rs.close();
|
stmt.close();
|
|
return orderedMap;
|
}
|
|
protected void queryIgsetElement(OracleConvertJobContext jobContext,
|
String srcschema, String srctable) throws SQLException
|
{
|
OracleConnection connection = jobContext.getOracleConnection();
|
String fetchSrcStmtFmt = "SELECT IGDSELM FROM \"%s\".\"%s\" ORDER BY ROWID";
|
PrintfFormat spf = new PrintfFormat(fetchSrcStmtFmt);
|
String fetchSrcStmt = spf.sprintf(new Object[]{srcschema, srctable});
|
Statement stmtSrc = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
|
stmtSrc.setFetchSize(FETCHSIZE);
|
|
ResultSet rsSrc = stmtSrc.executeQuery(fetchSrcStmt);
|
|
while (rsSrc.next())
|
{
|
byte[] raw = null;
|
|
if (rsSrc.getMetaData().getColumnType(1) == Types.BLOB)
|
{
|
BLOB blob = (BLOB) rsSrc.getBlob(1);
|
|
raw = getBytesFromBLOB(blob);
|
blob.close();
|
} else
|
{
|
raw = rsSrc.getBytes(1);
|
}
|
|
try
|
{
|
Element element = fetchBinaryElement(raw);
|
jobContext.putFeatureCollection(element);
|
} catch (Dgn7fileException e)
|
{
|
logger.warn("Dgn7Exception", e);
|
}
|
}
|
|
rsSrc.close();
|
stmtSrc.close();
|
}
|
|
protected void queryRawElement(OracleConvertJobContext jobContext,
|
String srcschema, String srctable) throws SQLException
|
{
|
OracleConnection connection = jobContext.getOracleConnection();
|
String fetchDestStmtFmt = "SELECT ELEMENT FROM \"%s\".\"%s\" ORDER BY ROWID";
|
PrintfFormat spf = new PrintfFormat(fetchDestStmtFmt);
|
String fetchDestStmt = spf.sprintf(new Object[]{srcschema, srctable});
|
Statement stmtDest = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
|
stmtDest.setFetchSize(FETCHSIZE);
|
|
ResultSet rsDest = stmtDest.executeQuery(fetchDestStmt);
|
|
while (rsDest.next())
|
{
|
ARRAY rawsValue = ((OracleResultSet) rsDest).getARRAY(1);
|
long[] rawData = rawsValue.getLongArray();
|
byte[] comparessedValue;
|
|
/*
|
if (dataMode == TransferTask.DataMode.Normal)
|
{
|
comparessedValue = BinConverter.unmarshalByteArray(rawData, true);
|
} else
|
{
|
comparessedValue = BinConverter.unmarshalCompactByteArray(rawData);
|
}
|
*/
|
comparessedValue = BinConverter.unmarshalByteArray(rawData, true);
|
|
byte[] rawDest = ByteArrayCompressor.decompressByteArray(comparessedValue);
|
|
|
try
|
{
|
Element element = fetchBinaryElement(rawDest);
|
jobContext.putFeatureCollection(element);
|
} catch (Dgn7fileException e)
|
{
|
logger.warn("Dgn7Exception:" + e.getMessage(), e);
|
}
|
}
|
|
rsDest.close();
|
stmtDest.close();
|
}
|
|
// Binary to Element
|
private Element fetchBinaryElement(byte[] raws) throws Dgn7fileException
|
{
|
ByteBuffer buffer = ByteBuffer.wrap(raws);
|
buffer.order(ByteOrder.LITTLE_ENDIAN);
|
short signature = buffer.getShort();
|
|
// byte type = (byte) (buffer.get() & 0x7f);
|
byte type = (byte) ((signature >>> 8) & 0x007f);
|
|
// silly Bentley say contentLength is in 2-byte words
|
// and ByteByffer uses raws.
|
// track the record location
|
int elementLength = (buffer.getShort() * 2) + 4;
|
ElementType recordType = ElementType.forID(type);
|
IElementHandler handler;
|
|
handler = recordType.getElementHandler();
|
|
Element dgnElement = (Element) handler.read(buffer, signature, elementLength);
|
if (recordType.isComplexElement() && (elementLength < raws.length))
|
{
|
int offset = elementLength;
|
while (offset < (raws.length - 4))
|
{
|
buffer.position(offset);
|
signature = buffer.getShort();
|
type = (byte) ((signature >>> 8) & 0x007f);
|
elementLength = (buffer.getShort() * 2) + 4;
|
if (raws.length < (offset + elementLength))
|
{
|
System.out.println("Length not match:" + offset + ":" + buffer.position() + ":" + buffer.limit());
|
break;
|
}
|
recordType = ElementType.forID(type);
|
handler = recordType.getElementHandler();
|
if (handler != null)
|
{
|
Element subElement = (Element) handler.read(buffer, signature, elementLength);
|
((ComplexElement) dgnElement).add(subElement);
|
offset += elementLength;
|
} else
|
{
|
byte[] remain = new byte[buffer.remaining()];
|
System.arraycopy(raws, offset, remain, 0, buffer.remaining());
|
for (int i = 0; i < remain.length; i++)
|
{
|
if (remain[i] != 0)
|
{
|
logger.info("fetch element has some error. index=" + (offset + i) + ":value=" + remain[i]);
|
System.out.println("fetch element has some error. index=" + (offset + i) + ":value=" + remain[i]);
|
}
|
}
|
break;
|
}
|
}
|
}
|
|
return dgnElement;
|
}
|
}
|
