/* * rhodes * * Copyright (C) 2008 Rhomobile, Inc. All rights reserved. * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ package com.rho.sync; import com.rho.RhoClassFactory; import com.rho.RhoEmptyLogger; import com.rho.RhoEmptyProfiler; import com.rho.RhoLogger; import com.rho.RhoProfiler; import com.rho.RhodesApp; import com.rho.Tokenizer; import com.rho.net.*; import com.rho.db.*; import java.util.Enumeration; import java.util.Vector; import java.util.Hashtable; import com.rho.FilePath; import com.rho.TimeInterval; import com.rho.RhoAppAdapter; import com.rho.net.NetRequest.MultipartItem; public class SyncSource { private static final RhoLogger LOG = RhoLogger.RHO_STRIP_LOG ? new RhoEmptyLogger() : new RhoLogger("Sync"); private static final RhoProfiler PROF = RhoProfiler.RHO_STRIP_PROFILER ? new RhoEmptyProfiler() : new RhoProfiler(); static class CAttrValue { String m_strAttrib = ""; String m_strValue = ""; String m_strBlobSuffix = ""; CAttrValue(String strAttrib, String strValue) { m_strAttrib = strAttrib; m_strValue = strValue; if ( m_strAttrib.endsWith("-rhoblob") ) { m_strBlobSuffix = "-rhoblob"; m_strAttrib = m_strAttrib.substring(0,m_strAttrib.length()-m_strBlobSuffix.length()); } } }; SyncEngine m_syncEngine; DBAdapter m_dbAdapter; Integer m_nID; String m_strName = ""; long m_token = 0; String m_strSyncType = ""; boolean m_bTokenFromDB; int m_nCurPageCount, m_nInserted, m_nDeleted, m_nTotalCount; boolean m_bGetAtLeastOnePage = false; int m_nErrCode = RhoAppAdapter.ERR_NONE; String m_strError = "", m_strServerError = ""; int m_nRefreshTime = 0; int m_nProgressStep = -1; boolean m_bSchemaSource; static class CAssociation { String m_strSrcName, m_strAttrib; CAssociation( String strSrcName, String strAttrib ){m_strSrcName = strSrcName; m_strAttrib = strAttrib; } }; Vector/**/ m_arAssociations = new Vector(); Vector/*Ptr*/ m_arMultipartItems = new Vector(); Vector/**/ m_arBlobAttrs = new Vector(); Hashtable/**/ m_hashIgnorePushObjects = new Hashtable(); Hashtable/**/ m_hashBelongsTo = new Hashtable(); Integer getID() { return m_nID; } String getName() { return m_strName; } String getSyncType(){ return m_strSyncType; } String getServerError(){ return m_strServerError; } int getErrorCode(){ return m_nErrCode; } int getServerObjectsCount(){ return m_nInserted+m_nDeleted; } long getToken(){ return m_token; } boolean isTokenFromDB(){ return m_bTokenFromDB; } void setToken(long token){ m_token = token; m_bTokenFromDB = false; } boolean isEmptyToken() { return m_token == 0; } int getProgressStep(){ return m_nProgressStep; } void setProgressStep(int nProgressStep){ m_nProgressStep = nProgressStep; } boolean getGetAtLeastOnePage(){ return m_bGetAtLeastOnePage; } int getRefreshTime(){ return m_nRefreshTime; } Vector/**/ getAssociations(){ return m_arAssociations; } int getInsertedCount() { return m_nInserted; } int getDeletedCount() { return m_nDeleted; } void setCurPageCount(int nCurPageCount){m_nCurPageCount = nCurPageCount;} void setTotalCount(int nTotalCount){m_nTotalCount = nTotalCount;} int getCurPageCount(){return m_nCurPageCount;} int getTotalCount(){return m_nTotalCount;} SyncEngine getSync(){ return m_syncEngine; } SyncNotify getNotify(){ return getSync().getNotify(); } NetRequest getNet(){ return getSync().getNet(); } ISyncProtocol getProtocol(){ return getSync().getProtocol(); } void setRefreshTime( int nRefreshTime ){ m_nRefreshTime = nRefreshTime;} DBAdapter getDB(){ return m_dbAdapter; } SyncSource(SyncEngine syncEngine, DBAdapter db)throws DBException { m_syncEngine = syncEngine; m_dbAdapter = db; m_nID = new Integer(0); m_bTokenFromDB = true; m_nCurPageCount = 0; m_nInserted = 0; m_nDeleted = 0; m_nTotalCount = 0; m_bGetAtLeastOnePage = false; m_nErrCode = RhoAppAdapter.ERR_NONE; m_bSchemaSource = db.isTableExist(m_strName); } SyncSource(int id, String name, String strSyncType, DBAdapter db, SyncEngine syncEngine )throws DBException { m_syncEngine = syncEngine; m_dbAdapter = db; m_nID = new Integer(id); m_strName = name; m_strSyncType = strSyncType; m_nCurPageCount = 0; m_nInserted = 0; m_nDeleted = 0; m_nTotalCount = 0; m_bGetAtLeastOnePage = false; m_nErrCode = RhoAppAdapter.ERR_NONE; IDBResult res = db.executeSQL("SELECT token,associations from sources WHERE source_id=?", m_nID); if ( !res.isEnd() ) { m_token = res.getLongByIdx(0); m_bTokenFromDB = true; }else { m_token = 0; m_bTokenFromDB = true; } m_bSchemaSource = db.isTableExist(m_strName); parseAssociations(res.getStringByIdx(1)); } void parseAssociations(String strAssociations) { if (strAssociations.length() == 0 ) return; Tokenizer oTokenizer = new Tokenizer( strAssociations, "," ); String strSrcName = ""; while (oTokenizer.hasMoreTokens()) { String tok = oTokenizer.nextToken(); if (tok.length() == 0) continue; if ( strSrcName.length() > 0 ) { m_arAssociations.addElement( new CAssociation(strSrcName, tok) ); strSrcName = ""; }else strSrcName = tok; } } void sync() throws Exception { getNotify().reportSyncStatus(RhoAppAdapter.getMessageText("syncronizing") + getName() + "...", m_nErrCode, m_strError ); TimeInterval startTime = TimeInterval.getCurrentTime(); //m_bIsSearch = false; try{ if ( isTokenFromDB() && getToken() > 1 ) syncServerChanges(); //sync only server changes, which was paused before else { if ( isEmptyToken() ) processToken(1); syncClientChanges(); syncServerChanges(); /* boolean bSyncedServer = syncClientChanges(); if ( !bSyncedServer ) syncServerChanges(); */ } }catch(Exception exc) { getSync().stopSync(); throw exc; }finally{ TimeInterval endTime = TimeInterval.getCurrentTime(); getDB().executeSQL("UPDATE sources set last_updated=?,last_inserted_size=?,last_deleted_size=?, "+ "last_sync_duration=?,last_sync_success=?, backend_refresh_time=? WHERE source_id=?", new Long(endTime.toULong()/1000), new Integer(getInsertedCount()), new Integer(getDeletedCount()), new Long((endTime.minus(startTime)).toULong()), new Integer(m_bGetAtLeastOnePage?1:0), new Integer(m_nRefreshTime), getID() ); } } void syncClientChanges()throws Exception { PROF.START("Pull"); boolean bSyncClient = false; { IDBResult res = getDB().executeSQL("SELECT object FROM changed_values WHERE source_id=? and sent<=1 LIMIT 1 OFFSET 0", getID()); bSyncClient = !res.isEnd(); } if ( bSyncClient ) doSyncClientChanges(); PROF.STOP("Pull"); } /* boolean syncClientChanges()throws Exception { boolean bSyncedServer = false; if ( isPendingClientChanges() ) { LOG.INFO( "Client has unconfirmed created items. Call server to update them." ); syncServerChanges(); bSyncedServer = true; } if ( bSyncedServer && isPendingClientChanges() ) { LOG.INFO( "Server does not sent created items. Stop sync." ); getSync().setState(SyncEngine.esStop); } else { PROF.START("Pull"); boolean bSyncClient = false; { IDBResult res = getDB().executeSQL("SELECT object FROM changed_values WHERE source_id=? LIMIT 1 OFFSET 0", getID()); bSyncClient = !res.isEnd(); } if ( bSyncClient ) { doSyncClientChanges(); bSyncedServer = false; } PROF.STOP("Pull"); } return bSyncedServer; } boolean isPendingClientChanges()throws DBException { IDBResult res = getDB().executeSQL("SELECT object FROM changed_values WHERE source_id=? and update_type='create' and sent>1 LIMIT 1 OFFSET 0", getID()); return !res.isEnd(); }*/ void addBelongsTo(String strAttrib, Integer nSrcID) { m_hashBelongsTo.put(strAttrib, nSrcID); } Integer getBelongsToSrcID(String strAttrib) { if ( m_hashBelongsTo.containsKey(strAttrib) ) return (Integer)m_hashBelongsTo.get(strAttrib); return new Integer(-1); } void checkIgnorePushObjects()throws Exception { // ignore changes in pending creates { IDBResult res = getDB().executeSQL("SELECT distinct(object) FROM changed_values where source_id=? and sent>=2", getID() ); for( ; !res.isEnd(); res.next() ) { String strObject = res.getStringByIdx(0); m_hashIgnorePushObjects.put(strObject, new Integer(1)); } } //check for belongs_to String strAttribQuests = ""; Vector/**/ arValues = new Vector(); arValues.addElement(getID()); Enumeration keys = m_hashBelongsTo.keys(); while (keys.hasMoreElements()) { if ( strAttribQuests.length() > 0 ) strAttribQuests += ","; strAttribQuests += "?"; arValues.addElement(keys.nextElement()); } if ( strAttribQuests.length() > 0 ) { IDBResult res = getDB().executeSQLEx( "SELECT object, attrib, value FROM changed_values where source_id=? and sent<=1 and attrib IN ( " + strAttribQuests + " )", arValues ); for( ; !res.isEnd(); res.next() ) { String strObject = res.getStringByIdx(0); String strAttrib = res.getStringByIdx(1); String strValue = res.getStringByIdx(2); IDBResult res2 = getDB().executeSQL( "SELECT object FROM changed_values where source_id=? and sent>=2 and object=? LIMIT 1 OFFSET 0", getBelongsToSrcID(strAttrib), strValue ); if (!res2.isEnd()) m_hashIgnorePushObjects.put(strObject, new Integer(1) ); } } } void doSyncClientChanges()throws Exception { String arUpdateTypes[] = {"create", "update", "delete"}; boolean arUpdateSent[] = {false, false, false}; m_arMultipartItems.removeAllElements(); m_arBlobAttrs.removeAllElements(); String strBody = "{\"source_name\":" + JSONEntry.quoteValue(getName()) + ",\"client_id\":" + JSONEntry.quoteValue(getSync().getClientID()); boolean bSend = false; int i = 0; getDB().Lock(); try{ checkIgnorePushObjects(); for( i = 0; i < 3 && getSync().isContinueSync(); i++ ) { String strBody1; strBody1 = makePushBody_Ver3(arUpdateTypes[i], true); if (strBody1.length() > 0) { strBody += "," + strBody1; String strBlobAttrs = ""; for ( int j = 0; j < (int)m_arBlobAttrs.size(); j++) { if ( strBlobAttrs.length() > 0 ) strBlobAttrs += ","; strBlobAttrs += JSONEntry.quoteValue((String)m_arBlobAttrs.elementAt(j)); } if ( strBlobAttrs.length() > 0 ) strBody += ",\"blob_fields\":[" + strBlobAttrs + "]"; arUpdateSent[i] = true; bSend = true; } } strBody += "}"; }finally { getDB().Unlock(); } if ( bSend ) { LOG.INFO( "Push client changes to server. Source: " + getName() + "Size :" + strBody.length() ); LOG.TRACE("Push body: " + strBody); try{ if ( m_arMultipartItems.size() > 0 ) { MultipartItem oItem = new MultipartItem(); oItem.m_strBody = strBody; //oItem.m_strContentType = getProtocol().getContentType(); oItem.m_strName = "cud"; m_arMultipartItems.addElement(oItem); NetResponse resp = getNet().pushMultipartData( getProtocol().getClientChangesUrl(), m_arMultipartItems, getSync(), null ); if ( !resp.isOK() ) { getSync().setState(SyncEngine.esStop); m_nErrCode = RhoAppAdapter.ERR_REMOTESERVER; m_strError = resp.getCharData(); } }else { NetResponse resp = getNet().pushData( getProtocol().getClientChangesUrl(), strBody, getSync()); if ( !resp.isOK() ) { getSync().setState(SyncEngine.esStop); m_nErrCode = RhoAppAdapter.ERR_REMOTESERVER; m_strError = resp.getCharData(); } } }catch(Exception exc) { m_nErrCode = RhoAppAdapter.getNetErrorCode(exc); throw exc; } } for( i = 0; i < 3 && getSync().isContinueSync(); i++ ) { if ( arUpdateSent[i] ) { //oo conflicts if ( i < 1 ) //create getDB().executeSQL("UPDATE changed_values SET sent=2 WHERE source_id=? and update_type=? and sent=1", getID(), arUpdateTypes[i] ); else // getDB().executeSQL("DELETE FROM changed_values WHERE source_id=? and update_type=? and sent=1", getID(), arUpdateTypes[i] ); } } m_arMultipartItems.removeAllElements(); m_arBlobAttrs.removeAllElements(); } //{"source_name":"SampleAdapter","client_id":1,"create":{"1":{"brand":"Apple","name":"iPhone","price":"199.99"}}} //{"source_name":"SampleAdapter","client_id":1,"update":{"1":{"brand":"Apple","name":"iPhone","price":"199.99"}}} //{"source_name":"SampleAdapter","client_id":1,"delete":{"1":{"brand":"Apple","name":"iPhone","price":"199.99"}}} //{"source_name":"SampleAdapter","client_id":1,"delete":{"3":{"brand":"HTC","name":"Fuze","price":"299.99"}},"create":{"1":{"brand":"Apple","name":"iPhone","price":"199.99"}},"update":{"2":{"brand":"Android","name":"G2","price":"99.99"}}} String makePushBody_Ver3( String strUpdateType, boolean isSync)throws DBException { String strBody = ""; getDB().Lock(); if ( isSync ) getDB().updateAllAttribChanges(); IDBResult res = getDB().executeSQL("SELECT attrib, object, value, attrib_type "+ "FROM changed_values where source_id=? and update_type =? and sent<=1 ORDER BY object", getID(), strUpdateType ); if ( res.isEnd() ) { res.close(); getDB().Unlock(); return strBody; } String strCurObject = ""; boolean bFirst = true; for( ; !res.isEnd(); res.next() ) { String strAttrib = res.getStringByIdx(0); String strObject = res.getStringByIdx(1); String value = res.getStringByIdx(2); String attribType = res.getStringByIdx(3); if ( m_hashIgnorePushObjects.containsKey(strObject) ) continue; if ( attribType.compareTo("blob.file") == 0 ) { MultipartItem oItem = new MultipartItem(); oItem.m_strFilePath = RhodesApp.getInstance().resolveDBFilesPath(value); oItem.m_strContentType = "application/octet-stream"; oItem.m_strName = strAttrib + "-" + strObject; m_arBlobAttrs.addElement(strAttrib); m_arMultipartItems.addElement(oItem); } if ( strBody.length() == 0 ) { if ( !isSync ) strBody += "{"; else strBody += "\"" + strUpdateType + "\":{"; } if ( strObject.compareTo(strCurObject) != 0 ) { if ( strCurObject.length() > 0 ) { if ( !bFirst ) strBody += "}"; strBody += ","; } bFirst = true; strBody += JSONEntry.quoteValue(strObject); strCurObject = strObject; } if (!bFirst) strBody += ","; if ( strAttrib.length() > 0 ) { if ( bFirst ) strBody += ":{"; strBody += JSONEntry.quoteValue(strAttrib) + ":" + JSONEntry.quoteValue(value); bFirst = false; } } if ( strBody.length() > 0 ) { if ( !bFirst ) strBody += "}"; strBody += "}"; } if ( isSync ) getDB().executeSQL("UPDATE changed_values SET sent=1 WHERE source_id=? and update_type=? and sent=0", getID(), strUpdateType ); getDB().Unlock(); return strBody; } void applyChangedValues()throws Exception { String strBody = makePushBody_Ver3("create", false); if ( strBody != null && strBody.length() > 0 ) { JSONEntry oEntry = new JSONEntry(strBody); processSyncCommand("insert", oEntry, false ); } strBody = makePushBody_Ver3("delete", false); if ( strBody != null && strBody.length() > 0 ) { JSONEntry oEntry = new JSONEntry(strBody); processSyncCommand("delete", oEntry, false ); } strBody = makePushBody_Ver3("update", false); if ( strBody != null && strBody.length() > 0 ) { JSONEntry oEntry = new JSONEntry(strBody); processSyncCommand("insert", oEntry, false ); } } void syncServerChanges()throws Exception { LOG.INFO("Sync server changes source ID :" + getID() ); while( getSync().isContinueSync() ) { setCurPageCount(0); String strUrl = getProtocol().getServerQueryUrl(""); String strQuery = getProtocol().getServerQueryBody(getName(), getSync().getClientID(), getSync().getSyncPageSize()); if ( !m_bTokenFromDB && getToken() > 1 ) strQuery += "&token=" + getToken(); LOG.INFO( "Pull changes from server. Url: " + (strUrl+strQuery) ); NetResponse resp = null; try{ PROF.START("Net"); resp = getNet().pullData(strUrl+strQuery, getSync()); PROF.STOP("Net"); if ( !resp.isOK() ) { getSync().stopSync(); m_nErrCode = RhoAppAdapter.getErrorFromResponse(resp); m_strError = resp.getCharData(); continue; } }catch(Exception exc) { m_nErrCode = RhoAppAdapter.getNetErrorCode(exc); throw exc; } String szData = null; String strTestResp = getSync().getSourceOptions().getProperty(getID(), "rho_server_response"); if ( strTestResp != null && strTestResp.length() > 0 ) szData = strTestResp; else szData = resp.getCharData(); PROF.START("Parse"); JSONArrayIterator oJsonArr = new JSONArrayIterator(szData); PROF.STOP("Parse"); processServerResponse_ver3(oJsonArr); if (getSync().getSourceOptions().getBoolProperty(getID(), "pass_through")) processToken(0); if ( getToken() == 0 ) break; } if ( getSync().isSchemaChanged() ) getSync().stopSync(); } //{"create-error":{"0_broken_object_id":{"name":"wrongname","an_attribute":"error create"},"0_broken_object_id-error":{"message":"error create"}}} boolean processServerErrors(JSONEntry oCmds)throws Exception { String arErrTypes[] = {"source-error", "search-error", "create-error", "update-error", "delete-error", null}; boolean bRes = false; for( int i = 0; ; i++ ) { if ( arErrTypes[i] == null ) break; if ( !oCmds.hasName(arErrTypes[i]) ) continue; bRes = true; m_nErrCode = RhoAppAdapter.ERR_CUSTOMSYNCSERVER; JSONEntry errSrc = oCmds.getEntry(arErrTypes[i]); JSONStructIterator errIter = new JSONStructIterator(errSrc); for( ; !errIter.isEnd(); errIter.next() ) { String strKey = errIter.getCurKey(); if ( i == 0 || i == 1 )//"source-error", "search-error" { if ( errIter.getCurValue().hasName("message") ) { if ( m_strServerError.length() > 0 ) m_strServerError += "&"; m_strServerError += "server_errors[" + URI.urlEncode(strKey) + "][message]=" + URI.urlEncode(errIter.getCurValue().getString("message")); } } else { //"create-error", "update-error", "delete-error" String strObject = strKey; if ( strObject.endsWith("-error") ) { strObject = strObject.substring(0, strKey.length()-6); if ( m_strServerError.length() > 0 ) m_strServerError += "&"; m_strServerError += "server_errors[" + arErrTypes[i] + "][" + URI.urlEncode(strObject) + "][message]=" + URI.urlEncode(errIter.getCurValue().getString("message")); }else { JSONStructIterator attrIter = new JSONStructIterator(errIter.getCurValue()); for( ; !attrIter.isEnd(); attrIter.next() ) { String strAttrName = attrIter.getCurKey(); String strAttrValue = attrIter.getCurString(); if ( m_strServerError.length() > 0 ) m_strServerError += "&"; m_strServerError += "server_errors[" + arErrTypes[i] + "][" + URI.urlEncode(strObject) + "][attributes][" + URI.urlEncode(strAttrName) + "]=" + URI.urlEncode(strAttrValue); } } } } } return bRes; } void processServerResponse_ver3(JSONArrayIterator oJsonArr)throws Exception { PROF.START("Data1"); int nVersion = 0; if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("version") ) { nVersion = oJsonArr.getCurItem().getInt("version"); oJsonArr.next(); } if ( nVersion != getProtocol().getVersion() ) { LOG.ERROR("Sync server send data with incompatible version. Client version: " + getProtocol().getVersion() + "; Server response version: " + nVersion + ". Source name: " + getName() ); getSync().stopSync(); m_nErrCode = RhoAppAdapter.ERR_UNEXPECTEDSERVERRESPONSE; return; } if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("token")) { processToken(oJsonArr.getCurItem().getUInt64("token")); oJsonArr.next(); } if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("source") ) { //skip it. it uses in search only oJsonArr.next(); } if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("count") ) { setCurPageCount(oJsonArr.getCurItem().getInt("count")); oJsonArr.next(); } if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("refresh_time") ) { setRefreshTime(oJsonArr.getCurItem().getInt("refresh_time")); oJsonArr.next(); } if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("progress_count") ) { //TODO: progress_count //setTotalCount(oJsonArr.getCurItem().getInt("progress_count")); oJsonArr.next(); } if ( !oJsonArr.isEnd() && oJsonArr.getCurItem().hasName("total_count") ) { setTotalCount(oJsonArr.getCurItem().getInt("total_count")); oJsonArr.next(); } //if ( getServerObjectsCount() == 0 ) // getNotify().fireSyncNotification(this, false, RhoAppAdapter.ERR_NONE, ""); if ( getToken() == 0 ) { //oo conflicts getDB().executeSQL("DELETE FROM changed_values where source_id=? and sent>=3", getID() ); // } LOG.INFO("Got " + getCurPageCount() + "(Processed: " + getServerObjectsCount() + ") records of " + getTotalCount() + " from server. Source: " + getName() + ". Version: " + nVersion ); PROF.STOP("Data1"); if ( !oJsonArr.isEnd() && getSync().isContinueSync() ) { JSONEntry oCmds = oJsonArr.getCurItem(); PROF.START("Data"); if ( oCmds.hasName("schema-changed") ) { getSync().setSchemaChanged(true); }else if ( !processServerErrors(oCmds) ) { getDB().startTransaction(); if (getSync().getSourceOptions().getBoolProperty(getID(), "pass_through")) { if ( m_bSchemaSource ) getDB().executeSQL( "DELETE FROM " + getName() ); else getDB().executeSQL( "DELETE FROM object_values WHERE source_id=?", getID() ); } if ( oCmds.hasName("metadata") && getSync().isContinueSync() ) { String strMetadata = oCmds.getString("metadata"); getDB().executeSQL("UPDATE sources SET metadata=? WHERE source_id=?", strMetadata, getID() ); } if ( oCmds.hasName("links") && getSync().isContinueSync() ) processSyncCommand("links", oCmds.getEntry("links"), true ); if ( oCmds.hasName("delete") && getSync().isContinueSync() ) processSyncCommand("delete", oCmds.getEntry("delete"), true ); if ( oCmds.hasName("insert") && getSync().isContinueSync() ) processSyncCommand("insert", oCmds.getEntry("insert"), true ); PROF.STOP("Data"); PROF.START("DB"); getDB().endTransaction(); PROF.STOP("DB"); getNotify().fireObjectsNotification(); } } PROF.START("Data1"); if ( getCurPageCount() > 0 ) getNotify().fireSyncNotification(this, false, RhoAppAdapter.ERR_NONE, ""); PROF.STOP("Data1"); } void processSyncCommand(String strCmd, JSONEntry oCmdEntry, boolean bCheckUIRequest)throws Exception { JSONStructIterator objIter = new JSONStructIterator(oCmdEntry); for( ; !objIter.isEnd() && getSync().isContinueSync(); objIter.next() ) { String strObject = objIter.getCurKey(); JSONStructIterator attrIter = new JSONStructIterator( objIter.getCurValue() ); try { if ( m_bSchemaSource ) processServerCmd_Ver3_Schema(strCmd,strObject,attrIter); else { for( ; !attrIter.isEnd() && getSync().isContinueSync(); attrIter.next() ) { String strAttrib = attrIter.getCurKey(); String strValue = attrIter.getCurString(); processServerCmd_Ver3(strCmd,strObject,strAttrib,strValue); } } }catch(DBException exc) { LOG.ERROR("Sync of server changes failed for " + getName() + ";object: " + strObject, exc); } if ( getSyncType().compareTo("none") == 0 ) continue; if ( bCheckUIRequest ) { int nSyncObjectCount = getNotify().incLastSyncObjectCount(getID()); if ( getProgressStep() > 0 && (nSyncObjectCount%getProgressStep() == 0) ) getNotify().fireSyncNotification(this, false, RhoAppAdapter.ERR_NONE, ""); if ( getDB().isUIWaitDB() ) { LOG.INFO("Commit transaction because of UI request."); getDB().endTransaction(); SyncThread.getInstance().sleep(1000); getDB().startTransaction(); } } } } void processAssociations(String strOldObject, String strNewObject)throws Exception { for ( int i = 0; i < m_arAssociations.size(); i++ ) { SyncSource pSrc = getSync().findSourceByName( ((CAssociation)m_arAssociations.elementAt(i)).m_strSrcName); if ( pSrc != null ) pSrc.updateAssociation(strOldObject, strNewObject, ((CAssociation)m_arAssociations.elementAt(i)).m_strAttrib); } } void updateAssociation(String strOldObject, String strNewObject, String strAttrib)throws Exception { if ( m_bSchemaSource ) { String strSqlUpdate = "UPDATE "; strSqlUpdate += getName() + " SET " + strAttrib + "=? where " + strAttrib + "=?"; getDB().executeSQL(strSqlUpdate, strNewObject, strOldObject ); } else getDB().executeSQL("UPDATE object_values SET value=? where attrib=? and source_id=? and value=?", strNewObject, strAttrib, getID(), strOldObject ); getDB().executeSQL("UPDATE changed_values SET value=? where attrib=? and source_id=? and value=?", strNewObject, strAttrib, getID(), strOldObject ); } void processServerCmd_Ver3_Schema(String strCmd, String strObject, JSONStructIterator attrIter)throws Exception { if ( strCmd.compareTo("insert") == 0 ) { Vector/**/ vecValues = new Vector(), vecAttrs = new Vector(); String strCols = "", strQuest = "", strSet = ""; for( ; !attrIter.isEnd() && getSync().isContinueSync(); attrIter.next() ) { CAttrValue oAttrValue = new CAttrValue(attrIter.getCurKey(),attrIter.getCurString()); if ( !processBlob(strCmd,strObject,oAttrValue) ) continue; if ( strCols.length() > 0 ) strCols += ","; if ( strQuest.length() > 0) strQuest += ","; if ( strSet.length() > 0) strSet += ","; strCols += oAttrValue.m_strAttrib; strQuest += "?"; strSet += oAttrValue.m_strAttrib + "=?"; vecAttrs.addElement(oAttrValue.m_strAttrib); vecValues.addElement(oAttrValue.m_strValue); } vecValues.addElement(strObject); if ( strCols.length() > 0 ) strCols += ","; if ( strQuest.length() > 0) strQuest += ","; strCols += "object"; strQuest += "?"; String strSqlInsert = "INSERT INTO "; strSqlInsert += getName() + " ("; strSqlInsert += strCols + ") VALUES(" + strQuest + ")"; if ( !getSync().isContinueSync() ) return; IDBResult resInsert = getDB().executeSQLReportNonUniqueEx(strSqlInsert, vecValues ); if ( resInsert.isNonUnique() ) { String strSqlUpdate = "UPDATE "; strSqlUpdate += getName() + " SET " + strSet + " WHERE object=?"; getDB().executeSQLEx(strSqlUpdate, vecValues); if ( getSyncType().compareTo("none") != 0 ) { // oo conflicts for( int i = 0; i < (int)vecAttrs.size(); i++ ) { getDB().executeSQL("UPDATE changed_values SET sent=4 where object=? and attrib=? and source_id=? and sent>1", strObject, vecAttrs.elementAt(i), getID() ); } // } } if ( getSyncType().compareTo("none") != 0 ) getNotify().onObjectChanged(getID(),strObject, SyncNotify.enUpdate); m_nInserted++; }else if (strCmd.compareTo("delete") == 0) { Vector/**/ vecAttrs = new Vector(); String strSet = ""; for( ; !attrIter.isEnd() && getSync().isContinueSync(); attrIter.next() ) { CAttrValue oAttrValue = new CAttrValue(attrIter.getCurKey(),attrIter.getCurString()); if ( strSet.length() > 0 ) strSet += ","; vecAttrs.addElement(oAttrValue.m_strAttrib); strSet += oAttrValue.m_strAttrib + "=NULL"; } String strSqlUpdate = "UPDATE "; strSqlUpdate += getName() + " SET " + strSet + " WHERE object=?"; if ( strSet.length() == 0 || !getSync().isContinueSync() ) return; getDB().executeSQL(strSqlUpdate, strObject); //Remove item if all nulls String strSelect = "SELECT * FROM " + getName() + " WHERE object=?"; IDBResult res = getDB().executeSQL( strSelect, strObject ); if ( !res.isEnd() ) { boolean bAllNulls = true; for( int i = 0; i < res.getColCount(); i ++) { if ( !res.isNullByIdx(i) && res.getColName(i).compareTo("object")!=0 ) { bAllNulls = false; break; } } if (bAllNulls) { String strDelete = "DELETE FROM " + getName() + " WHERE object=?"; getDB().executeSQL( strDelete, strObject); } } if ( getSyncType().compareTo("none") != 0 ) { getNotify().onObjectChanged(getID(), strObject, SyncNotify.enDelete); // oo conflicts for( int i = 0; i < (int)vecAttrs.size(); i++ ) { getDB().executeSQL("UPDATE changed_values SET sent=3 where object=? and attrib=? and source_id=?", strObject, vecAttrs.elementAt(i), getID() ); } // } m_nDeleted++; }else if ( strCmd.compareTo("links") == 0 ) { String strValue = attrIter.getCurString(); processAssociations(strObject, strValue); String strSqlUpdate = "UPDATE "; strSqlUpdate += getName() + " SET object=? WHERE object=?"; getDB().executeSQL(strSqlUpdate, strValue, strObject); getDB().executeSQL("UPDATE changed_values SET object=?,sent=3 where object=? and source_id=?", strValue, strObject, getID() ); getNotify().onObjectChanged(getID(), strObject, SyncNotify.enCreate); } } boolean processBlob( String strCmd, String strObject, CAttrValue oAttrValue )throws Exception { //TODO: when server return delete with rhoblob postfix - delete isBlobAttr if ( !(oAttrValue.m_strBlobSuffix.length() > 0 || getDB().getAttrMgr().isBlobAttr(getID(), oAttrValue.m_strAttrib)) ) return true; boolean bDownload = true; String strDbValue = ""; if ( !getDB().getAttrMgr().isOverwriteBlobFromServer(getID(), oAttrValue.m_strAttrib) ) { if ( m_bSchemaSource ) { String strSelect = "SELECT " + oAttrValue.m_strAttrib + " FROM " + getName() + " WHERE object=?"; IDBResult res = getDB().executeSQL( strSelect, strObject); if (!res.isEnd()) { strDbValue = res.getStringByIdx(0); bDownload = strDbValue == null || strDbValue.length() == 0; } }else { IDBResult res = getDB().executeSQL( "SELECT value FROM object_values WHERE object=? and attrib=? and source_id=?", strObject, oAttrValue.m_strAttrib, getID() ); if (!res.isEnd()) { strDbValue = res.getStringByIdx(0); bDownload = strDbValue == null || strDbValue.length() == 0; } } } if ( bDownload ) { boolean bRes = false; getDB().endTransaction(); try{ bRes = downloadBlob(oAttrValue); }finally { getDB().startTransaction(); } return bRes; } /* String fName = makeFileName( oAttrValue ); String fOldName = RhodesApp.getInstance().resolveDBFilesPath(strDbValue); RhoClassFactory.createFile().renameOverwrite(fOldName, fName); oAttrValue.m_strValue = FilePath.getRelativePath( fName, RhodesApp.getInstance().getRhoRootPath()); */ oAttrValue.m_strValue = strDbValue; return true; } void processServerCmd_Ver3(String strCmd, String strObject, String strAttriba, String strValuea)throws Exception { CAttrValue oAttrValue = new CAttrValue(strAttriba,strValuea); if ( strCmd.compareTo("insert") == 0 ) { if ( !processBlob(strCmd,strObject,oAttrValue) ) return; IDBResult resInsert = getDB().executeSQLReportNonUnique("INSERT INTO object_values "+ "(attrib, source_id, object, value) VALUES(?,?,?,?)", oAttrValue.m_strAttrib, getID(), strObject, oAttrValue.m_strValue ); if ( resInsert.isNonUnique() ) { getDB().executeSQL("UPDATE object_values " + "SET value=? WHERE object=? and attrib=? and source_id=?", oAttrValue.m_strValue, strObject, oAttrValue.m_strAttrib, getID() ); if ( getSyncType().compareTo("none") != 0 ) { // oo conflicts getDB().executeSQL("UPDATE changed_values SET sent=4 where object=? and attrib=? and source_id=? and sent>1", strObject, oAttrValue.m_strAttrib, getID() ); // } } if ( getSyncType().compareTo("none") != 0 ) getNotify().onObjectChanged(getID(),strObject, SyncNotify.enUpdate); m_nInserted++; }else if (strCmd.compareTo("delete") == 0) { getDB().executeSQL("DELETE FROM object_values where object=? and attrib=? and source_id=?", strObject, oAttrValue.m_strAttrib, getID() ); if ( getSyncType().compareTo("none") != 0 ) { getNotify().onObjectChanged(getID(), strObject, SyncNotify.enDelete); // oo conflicts getDB().executeSQL("UPDATE changed_values SET sent=3 where object=? and attrib=? and source_id=?", strObject, oAttrValue.m_strAttrib, getID() ); // } m_nDeleted++; }else if ( strCmd.compareTo("links") == 0 ) { processAssociations(strObject, oAttrValue.m_strValue); getDB().executeSQL("UPDATE object_values SET object=? where object=? and source_id=?", oAttrValue.m_strValue, strObject, getID() ); getDB().executeSQL("UPDATE changed_values SET object=?,sent=3 where object=? and source_id=?", oAttrValue.m_strValue, strObject, getID() ); getNotify().onObjectChanged(getID(), strObject, SyncNotify.enCreate); } } private String makeFileName(CAttrValue value)throws Exception { String strExt = ""; URI uri = new URI(value.m_strValue); String strQuest = uri.getQueryString(); if (strQuest != null && strQuest.length() > 0) { int nExt = strQuest.indexOf("extension="); if ( nExt >= 0 ) { int nExtEnd = strQuest.indexOf("&", nExt); if (nExtEnd < 0 ) nExtEnd = strQuest.length(); strExt = strQuest.substring(nExt+10, nExtEnd); } } if ( strExt.length() == 0 ) { String strFileName = uri.getLastNamePart(); int nExt = strFileName != null ? strFileName.lastIndexOf('.') : -1; if ( nExt >= 0 ) strExt = strFileName.substring(nExt); } if ( strExt.length() == 0 ) strExt = ".bin"; else if ( strExt.charAt(0) != '.' ) strExt = "." + strExt; String fName = RhodesApp.getInstance().getBlobsDirPath() + "/id_" + TimeInterval.getCurrentTime().toULong() + strExt; return fName; /* String strExt = ".bin"; URI uri = new URI(value.m_strValue); int nDot = uri.getPath().lastIndexOf('.'); if ( nDot >= 0 ) strExt = uri.getPath().substring(nDot); else{ int nExt = uri.getQueryString().indexOf("extension="); if ( nExt >= 0 ){ int nExtEnd = uri.getQueryString().indexOf("&", nExt); if (nExtEnd < 0 ) nExtEnd = uri.getQueryString().length(); strExt = uri.getQueryString().substring(nExt+10, nExtEnd); } } String fName = RhodesApp.getInstance().getBlobsDirPath() + "/id_" + TimeInterval.getCurrentTime().toULong() + strExt; return fName;*/ } boolean downloadBlob(CAttrValue value)throws Exception { String fName = makeFileName( value ); String url = value.m_strValue; int nQuest = url.lastIndexOf('?'); if ( nQuest > 0 ) url += "&"; else url += "?"; url += "client_id=" + getSync().getClientID(); try{ NetResponse resp = getNet().pullFile(url, fName, getSync(), null); if ( !resp.isOK() ) { getSync().stopSync(); m_nErrCode = RhoAppAdapter.getErrorFromResponse(resp); return false; } }catch(Exception exc) { m_nErrCode = RhoAppAdapter.getNetErrorCode(exc); throw exc; } value.m_strValue = FilePath.getRelativePath( fName, RhodesApp.getInstance().getRhoRootPath()); return true; } void processToken(long token)throws DBException { if ( token > 1 && getToken() == token ){ //Delete non-confirmed records setToken( token ); //For m_bTokenFromDB = false; //getDB().executeSQL("DELETE FROM object_values where source_id=? and token=?", getID(), token ); //TODO: add special table for id,token }else { setToken( token ); getDB().executeSQL("UPDATE sources SET token=? where source_id=?", new Long(token), getID() ); } } }