package com.ximple.eofms.jobs; import java.io.IOException; 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 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); for (String orgSchema : _orgSchema) { exetcuteConvert(jobContext, orgSchema, _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; } }