Revision: 3376
Author: ferguson.sebastian
Date: Fri Mar 12 13:32:07 2010
Log: Added basic conflict resolution & a framework with which to do more
advanced stuff. There is now synchronization on the server so that people
do not try to write at the same time. Issues with gui components still need
to be resolved.
http://code.google.com/p/power-architect/source/detail?r=3376
Added:
/trunk/src/ca/sqlpower/architect/enterprise/JSONMessage.java
/trunk/src/ca/sqlpower/architect/enterprise/JSONResponseHandler.java
/trunk/src/ca/sqlpower/architect/enterprise/NetworkConflictResolver.java
Modified:
/trunk/src/ca/sqlpower/architect/enterprise/ArchitectClientSideSession.java
/trunk/src/ca/sqlpower/architect/swingui/dbtree/DBTreeModel.java
=======================================
--- /dev/null
+++ /trunk/src/ca/sqlpower/architect/enterprise/JSONMessage.java Fri Mar 12
13:32:07 2010
@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2010, SQL Power Group Inc.
+ *
+ * This file is part of Power*Architect.
+ *
+ * Power*Architect 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.
+ *
+ * Power*Architect 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 <http://www.gnu.org/licenses/>.
+ */
+
+package ca.sqlpower.architect.enterprise;
+
+public class JSONMessage {
+ private final String message;
+ private final boolean successful;
+
+ public JSONMessage(String message, boolean successful) {
+ this.message = message;
+ this.successful = successful;
+ }
+
+ public String getBody() {
+ return message;
+ }
+
+ public boolean isSuccessful() {
+ return successful;
+ }
+}
=======================================
--- /dev/null
+++ /trunk/src/ca/sqlpower/architect/enterprise/JSONResponseHandler.java
Fri Mar 12 13:32:07 2010
@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2010, SQL Power Group Inc.
+ *
+ * This file is part of Power*Architect.
+ *
+ * Power*Architect 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.
+ *
+ * Power*Architect 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 <http://www.gnu.org/licenses/>.
+ */
+
+package ca.sqlpower.architect.enterprise;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.client.ResponseHandler;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+public class JSONResponseHandler implements ResponseHandler<JSONMessage> {
+
+ /*
+ * Unsuccessful responses should have information sent in a header,
+ * either as "unsuccessfulResponse" or "exceptionStackTrace"
+ */
+
+ public JSONMessage handleResponse(HttpResponse response) {
+ try {
+
+ BufferedReader reader = new BufferedReader(
+ new
InputStreamReader(response.getEntity().getContent()));
+ StringBuffer buffer = new StringBuffer();
+
+ String line;
+ while ((line = reader.readLine()) != null) {
+ buffer.append(line).append("\n");
+ }
+ JSONObject message = new JSONObject(buffer.toString());
+
+ // Does the response contain data? If so, return it.
Communication
+ // with the resource has been successful.
+ if (message.getString("responseKind").equals("data")) {
+ return new JSONMessage(message.getString("data"), true);
+ } else {
+ // Has the request been unsuccessful?
+ if
(message.getString("responseKind").equals("unsuccessful")) {
+ return new JSONMessage(message.getString("data"),
false);
+ } else {
+ // Does the response contain an exception? If so,
reconstruct, and then
+ // re-throw it. There has been an exception on the
server.
+ if
(message.getString("responseKind").equals("exceptionStackTrace")) {
+
+ JSONArray stackTraceStrings = new
JSONArray(message.getString("data"));
+ StringBuffer stackTraceMessage = new
StringBuffer();
+ for (int i = 0; i < stackTraceStrings.length();
i++) {
+
stackTraceMessage.append("\n").append(stackTraceStrings.get(i));
+ }
+
+ throw new Exception(stackTraceMessage.toString());
+
+ } else {
+ // This exception represents a(n epic)
client-server miscommunication
+ throw new Exception("Unable to parse response ");
+ }
+ }
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+}
=======================================
--- /dev/null
+++
/trunk/src/ca/sqlpower/architect/enterprise/NetworkConflictResolver.java
Fri Mar 12 13:32:07 2010
@@ -0,0 +1,420 @@
+/*
+ *
+ * This file is part of Power*Architect.
+ *
+ * Power*Architect 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.
+ *
+ * Power*Architect 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 <http://www.gnu.org/licenses/>.
+ */
+
+package ca.sqlpower.architect.enterprise;
+
+import java.net.URI;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.StringEntity;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import ca.sqlpower.dao.MessageSender;
+import ca.sqlpower.dao.PersistedObjectEntry;
+import ca.sqlpower.dao.PersistedPropertiesEntry;
+import ca.sqlpower.dao.PersistedSPOProperty;
+import ca.sqlpower.dao.PersistedSPObject;
+import ca.sqlpower.dao.RemovedObjectEntry;
+import ca.sqlpower.dao.SPPersistenceException;
+import ca.sqlpower.dao.SPPersisterListener;
+import ca.sqlpower.dao.SPSessionPersister;
+import ca.sqlpower.dao.SPPersister.DataType;
+import ca.sqlpower.dao.json.SPJSONMessageDecoder;
+import ca.sqlpower.dao.session.SessionPersisterSuperConverter;
+import ca.sqlpower.object.SPObject;
+import ca.sqlpower.util.SPSession;
+import ca.sqlpower.util.SQLPowerUtils;
+
+import com.enterprisedt.util.debug.Logger;
+import com.google.common.collect.LinkedListMultimap;
+import com.google.common.collect.Multimap;
+
+public class NetworkConflictResolver extends Thread implements
MessageSender<JSONObject> {
+
+ private static final Logger logger =
Logger.getLogger(NetworkConflictResolver.class);
+ private AtomicBoolean postingJSON = new AtomicBoolean(false);
+ private boolean updating = false;
+
+ private SPPersisterListener listener;
+ private SPSessionPersister persister;
+ private SessionPersisterSuperConverter converter;
+ private final SPSession session;
+
+ private int currentRevision = 0;
+
+ private long retryDelay = 1000;
+
+ private final SPJSONMessageDecoder jsonDecoder;
+
+ private final ProjectLocation projectLocation;
+ private final HttpClient outboundHttpClient;
+ private final HttpClient inboundHttpClient;
+
+ private String contextRelativePath;
+
+ private volatile boolean cancelled;
+
+ private JSONArray messageBuffer = new JSONArray();
+
+ private List<PersistedSPObject> inboundObjectsToAdd = new
LinkedList<PersistedSPObject>();
+ private Multimap<String, PersistedSPOProperty>
inboundPropertiesToChange = LinkedListMultimap.create();
+ private List<RemovedObjectEntry> inboundObjectsToRemove = new
LinkedList<RemovedObjectEntry>();
+
+ private List<PersistedSPObject> outboundObjectsToAdd = new
LinkedList<PersistedSPObject>();
+ private Multimap<String, PersistedSPOProperty>
outboundPropertiesToChange = LinkedListMultimap.create();
+ private List<RemovedObjectEntry> outboundObjectsToRemove = new
LinkedList<RemovedObjectEntry>();
+
+ private List<PersistedObjectEntry> outboundObjectsToAddRollbackList =
new LinkedList<PersistedObjectEntry>();
+ private List<PersistedPropertiesEntry>
outboundPropertiesToChangeRollbackList = new
LinkedList<PersistedPropertiesEntry>();
+ private List<RemovedObjectEntry> outboundObjectsToRemoveRollbackList =
new LinkedList<RemovedObjectEntry>();
+
+ public NetworkConflictResolver(
+ ProjectLocation projectLocation,
+ SPJSONMessageDecoder jsonDecoder,
+ HttpClient inboundHttpClient,
+ HttpClient outboundHttpClient,
+ SPSession session)
+ {
+ super("updater-" + projectLocation.getUUID());
+
+ this.jsonDecoder = jsonDecoder;
+ this.projectLocation = projectLocation;
+ this.inboundHttpClient = inboundHttpClient;
+ this.outboundHttpClient = outboundHttpClient;
+ this.session = session;
+
+ contextRelativePath = "/project/" + projectLocation.getUUID();
+ }
+
+ public void setListener(SPPersisterListener listener) {
+ this.listener = listener;
+ }
+
+ public void setConverter(SessionPersisterSuperConverter converter) {
+ this.converter = converter;
+ }
+
+ public void setPersister(SPSessionPersister persister) {
+ this.persister = persister;
+ }
+
+ public void flush() {
+ flush(false);
+ }
+
+ private void flush(boolean reflush) {
+ if (postingJSON.get() && !reflush) {
+ return;
+ } else {
+ postingJSON.set(true);
+ }
+ // Try to send json message ...
+ JSONMessage response = postJsonArray(messageBuffer.toString());
+ if (response.isSuccessful()) {
+ // Sent json message without conflict.
+ try {
+ currentRevision = (new
JSONObject(response.getBody())).getInt("currentRevision");
+ } catch (JSONException e) {
+ throw new RuntimeException("Could not update current
revision" + e.getMessage());
+ }
+ // Prepare for next send ...
+ clear(reflush);
+ } else {
+ // Did not successfully post json, we must update ourselves,
and then try again if we can.
+ if (!reflush) {
+ // These lists should reflect the state of the workspace
at the time of the conflict.
+ // The workspace could be updated several times before a
successful post is made.
+ fillOutboundPersistedLists();
+ }
+ // Try to rollback our changes
+ try {
+ session.getWorkspace().rollback("Hello this is a
rollback");
+ } catch (Exception e) {
+ throw new RuntimeException("Reflush failed on rollback",
e);
+ }
+ String json;
+ int newRev;
+ try {
+ JSONObject jsonObject = new JSONObject(response.getBody());
+ json = jsonObject.getString("data");
+ newRev = jsonObject.getInt("currentRevision");
+ } catch (Exception e) {
+ throw new RuntimeException("Reflush failed on getJson", e);
+ }
+ // Try to create inboundPersistedLists for comparison with the
outbound. These will be used
+ // for special case collision detection.
+ fillInboundPersistedLists(json);
+ // Try to apply update
+ decodeMessage(json, newRev);
+ // We need an additional step here for checking for special
case conflicts
+ if (detectConflict()) {
+ throw new RuntimeException("There is a conflict between
our state and the server's, our changes will be lost");
+ } else {
+ // Try to return the persisted objects to their state
pre-update.
+ try {
+
SPSessionPersister.redoForSession(session.getWorkspace(),
+ outboundObjectsToAdd,
outboundPropertiesToChange,
+ outboundObjectsToRemove, converter);
+ // We want to re-send our changes, but only if we were
able to restore them
+ flush(true);
+ } catch (Exception ex) {
+ throw new RuntimeException("Reflush failed on
rollforward", ex);
+ }
+ }
+ }
+ postingJSON.set(false);
+ }
+
+ public void clear() {
+ clear(false);
+ }
+
+ private void clear(boolean reflush) {
+ messageBuffer = new JSONArray();
+
+ if (reflush) {
+ inboundObjectsToAdd.clear();
+ inboundPropertiesToChange.clear();
+ inboundObjectsToRemove.clear();
+
+ outboundObjectsToAdd.clear();
+ outboundPropertiesToChange.clear();
+ outboundObjectsToRemove.clear();
+
+ outboundObjectsToAddRollbackList.clear();
+ outboundPropertiesToChangeRollbackList.clear();
+ outboundObjectsToRemoveRollbackList.clear();
+ }
+ }
+
+ public void send(JSONObject content) throws SPPersistenceException {
+ messageBuffer.put(content);
+ }
+
+ @Override
+ public void run() {
+ try {
+ while (!this.isInterrupted() && !cancelled) {
+ try {
+
+ while (updating) { // this should wait for persisting
to server as well.
+ synchronized (this) {
+ wait();
+ }
+ }
+
+ updating = true;
+
+ // Request an update from the server using the current
revision number.
+ JSONMessage message = getJsonArray(inboundHttpClient);
+ final JSONObject json = new
JSONObject(message.getBody());
+ session.runInForeground(new Runnable() {
+ public void run() {
+ try {
+ if (!postingJSON.get()) {
+ decodeMessage(json.getString("data"),
json.getInt("currentRevision"));
+ }
+ } catch (Exception e) {
+ // TODO: Discard corrupt workspace and
start again from scratch.
+ interrupt();
+ throw new RuntimeException("Update from
server failed! Unable to decode the message: ", e);
+ } finally {
+ synchronized (NetworkConflictResolver.this)
{
+ updating = false;
+ NetworkConflictResolver.this.notify();
+ }
+ }
+ }
+ });
+ } catch (Exception ex) {
+ logger.error("Failed to contact server. Will retry in "
+ retryDelay + " ms.", ex);
+ Thread.sleep(retryDelay);
+ }
+ }
+ } catch (InterruptedException ex) {
+ logger.info("Updater thread exiting normally due to
interruption.");
+ }
+
+ inboundHttpClient.getConnectionManager().shutdown();
+ }
+
+ /**
+ * Exists for code reuse.
+ *
+ * @param jsonArray
+ * @param newRevision
+ * @throws SPPersistenceException
+ */
+ private void decodeMessage(String jsonArray, int newRevision) {
+ try {
+ if (currentRevision < newRevision) {
+ // Now we can apply the update ...
+ jsonDecoder.decode(jsonArray);
+ currentRevision = newRevision;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to decode the message: " +
jsonArray, e);
+ }
+ }
+
+ public void interrupt() {
+ super.interrupt();
+ cancelled = true;
+ }
+
+ private void fillOutboundPersistedLists() {
+ for (PersistedSPObject obj : listener.getPersistedObjects()) {
+ outboundObjectsToAdd.add(obj);
+ }
+ for (String uuid : listener.getPersistedProperties().keySet()) {
+ for (PersistedSPOProperty prop :
listener.getPersistedProperties().asMap().get(uuid)) {
+ outboundPropertiesToChange.put(uuid, prop);
+ }
+ }
+ for (RemovedObjectEntry rem : listener.getObjectsToRemove()) {
+ outboundObjectsToRemove.add(rem);
+ }
+ for (PersistedObjectEntry poe : listener.getObjectsRollbackList())
{
+ outboundObjectsToAddRollbackList.add(poe);
+ }
+ for (PersistedPropertiesEntry ppe :
listener.getPropertiesRollbackList()) {
+ outboundPropertiesToChangeRollbackList.add(ppe);
+ }
+ for (RemovedObjectEntry roe : listener.getRemovedRollbackList()) {
+ outboundObjectsToRemoveRollbackList.add(roe);
+ }
+ }
+
+ private void fillInboundPersistedLists(String json) {
+ try {
+ JSONArray array = new JSONArray(json);
+ for (int i = 0; i < array.length(); i++) {
+ JSONObject obj = array.getJSONObject(i);
+
+ if (obj.getString("method").equals("persistObject")) {
+
+ String parentUUID = obj.getString("parentUUID");
+ String type = obj.getString("type");
+ String uuid = obj.getString("uuid");
+ int index = obj.getInt("index");
+
+ inboundObjectsToAdd.add(new
PersistedSPObject(parentUUID, type, uuid, index));
+
+ } else if
(obj.getString("method").equals("persistProperty")) {
+
+ String uuid = obj.getString("uuid");
+ String propertyName = obj.getString("propertyName");
+ DataType type =
DataType.valueOf(obj.getString("type"));
+ Object oldValue = null;
+ try {
+ oldValue = SPJSONMessageDecoder.getWithType(obj,
type, "oldValue");
+ } catch (Exception e) {}
+ Object newValue =
SPJSONMessageDecoder.getWithType(obj, type, "newValue");
+ boolean unconditional = false;
+
+ PersistedSPOProperty property = new
PersistedSPOProperty(uuid, propertyName, type, oldValue, newValue,
unconditional);
+
+ if (inboundPropertiesToChange.keySet().contains(uuid))
{
+
inboundPropertiesToChange.asMap().get(uuid).add(property);
+ } else {
+ inboundPropertiesToChange.put(uuid, property);
+ }
+
+ } else if (obj.getString("method").equals("removeObject"))
{
+
+ String parentUUID = obj.getString("parentUUID");
+ String uuid = obj.getString("uuid");
+ SPObject objectToRemove =
SQLPowerUtils.findByUuid(session.getWorkspace(), uuid, SPObject.class);
+
+ inboundObjectsToRemove.add(new
RemovedObjectEntry(parentUUID, objectToRemove,
+
objectToRemove.getParent().getChildren().indexOf(objectToRemove)));
+ }
+ }
+ } catch (Exception ex) {
+ throw new RuntimeException("Unable to create persisted
lists: ", ex);
+ }
+ }
+
+ private boolean detectConflict() {
+ // XXX : Special cases are to be looked for here.
+ if (checkForSimultaneousEdit()) return true;
+ // No special cases found.
+ return false;
+ }
+
+ private boolean checkForSimultaneousEdit() {
+ String targetDBuuid = session.getWorkspace().getUUID();
+ for (PersistedSPObject inSpo : inboundObjectsToAdd) {
+ for (PersistedSPObject outSpo : outboundObjectsToAdd) {
+ if (inSpo.getParentUUID().equals(outSpo.getParentUUID())
&& !inSpo.equals(targetDBuuid)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Creates and executes an HttpPost request containing the json of
whatever
+ * transaction was completed last.
+ * @param jsonArray Typically created by calling toString() on a
JSONArray
+ * @return A JSONMessage holding the successfulness and message body
of the server's response
+ */
+ private JSONMessage postJsonArray(String jsonArray) {
+ try {
+ URI serverURI = new URI("http", null,
+ projectLocation.getServiceInfo().getServerAddress(),
+ projectLocation.getServiceInfo().getPort(),
+ projectLocation.getServiceInfo().getPath()
+ "/project/" + projectLocation.getUUID(),
+ "currentRevision=" + currentRevision, null);
+ HttpPost postRequest = new HttpPost(serverURI);
+ postRequest.setEntity(new StringEntity(jsonArray));
+ postRequest.setHeader("Content-Type", "application/json");
+ HttpUriRequest request = postRequest;
+ return outboundHttpClient.execute(request, new
JSONResponseHandler());
+ } catch (Exception ex) {
+ throw new RuntimeException("Unable to post json to server: " +
jsonArray + "\n"+ ex.getMessage());
+ }
+ }
+
+ /**
+ * Creates and executes an HttpGet request for an update from the
server.
+ * @return A JSONMessage holding the successfulness and message body
of the server's response
+ */
+ private JSONMessage getJsonArray(HttpClient client) {
+ try {
+ URI uri = new URI("http", null,
+ projectLocation.getServiceInfo().getServerAddress(),
+ projectLocation.getServiceInfo().getPort(),
+ projectLocation.getServiceInfo().getPath() +
contextRelativePath,
+ "oldRevisionNo=" + currentRevision, null);
+ HttpUriRequest request = new HttpGet(uri);
+ return client.execute(request, new JSONResponseHandler());
+ } catch (Exception ex) {
+ throw new RuntimeException("Unable to get json from server: "
+ ex.getMessage());
+ }
+ }
+}
=======================================
---
/trunk/src/ca/sqlpower/architect/enterprise/ArchitectClientSideSession.java
Mon Mar 1 15:40:10 2010
+++
/trunk/src/ca/sqlpower/architect/enterprise/ArchitectClientSideSession.java
Fri Mar 12 13:32:07 2010
@@ -2,10 +2,7 @@
import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
-import java.io.BufferedReader;
import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.ParseException;
@@ -29,7 +26,6 @@
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpUriRequest;
-import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.BasicCookieStore;
import org.apache.http.impl.client.BasicResponseHandler;
import org.apache.http.impl.client.DefaultHttpClient;
@@ -50,7 +46,6 @@
import ca.sqlpower.architect.ddl.DDLGenerator;
import ca.sqlpower.architect.swingui.ArchitectSwingSession;
import ca.sqlpower.architect.swingui.ArchitectSwingSessionContext;
-import ca.sqlpower.dao.HttpMessageSender;
import ca.sqlpower.dao.SPPersistenceException;
import ca.sqlpower.dao.SPPersisterListener;
import ca.sqlpower.dao.SPSessionPersister;
@@ -114,8 +109,7 @@
* server changes to the {...@link #sessionPersister}.
*/
private final SPJSONPersister jsonPersister;
- private final Updater updater;
- private final Sender sender;
+ private final NetworkConflictResolver updater;
private final SPJSONMessageDecoder jsonMessageDecoder;
private final DataSourceCollectionUpdater dataSourceCollectionUpdater =
new DataSourceCollectionUpdater();
@@ -152,18 +146,20 @@
}
outboundHttpClient =
createHttpClient(projectLocation.getServiceInfo());
-
- sender = new Sender(outboundHttpClient,
projectLocation.getServiceInfo(), projectLocation.getUUID());
- jsonPersister = new SPJSONPersister(sender);
-
dataSourceCollection = getDataSources();
-
sessionPersister = new ArchitectSessionPersister("inbound-" +
projectLocation.getUUID(), getWorkspace(),
- new SessionPersisterSuperConverter(dataSourceCollection,
getWorkspace()));
+ new SessionPersisterSuperConverter(dataSourceCollection,
getWorkspace()));
sessionPersister.setSession(this);
jsonMessageDecoder = new SPJSONMessageDecoder(sessionPersister);
- updater = new Updater(projectLocation.getUUID(),
jsonMessageDecoder);
+
+ updater = new NetworkConflictResolver(
+ projectLocation,
+ jsonMessageDecoder,
+ createHttpClient(projectLocation.getServiceInfo()),
+ outboundHttpClient, this);
+
+ jsonPersister = new SPJSONPersister(updater);
}
// -
@@ -249,13 +245,16 @@
}
public void startUpdaterThread() {
- updater.start();
final SPPersisterListener listener = new
SPPersisterListener(jsonPersister, sessionPersister,
new SessionPersisterSuperConverter(dataSourceCollection,
getWorkspace()));
-
SQLPowerUtils.listenToHierarchy(getWorkspace(), listener);
+ updater.setListener(listener);
+ updater.setPersister(sessionPersister);
+ updater.setConverter(new
SessionPersisterSuperConverter(dataSourceCollection, getWorkspace()));
+ updater.start();
+
addSessionLifecycleListener(new
SessionLifecycleListener<ArchitectSession>() {
public void
sessionClosing(SessionLifecycleEvent<ArchitectSession> e) {
SQLPowerUtils.unlistenToHierarchy(getWorkspace(), listener);
@@ -542,228 +541,6 @@
new UsernamePasswordCredentials(serviceInfo.getUsername(),
serviceInfo.getPassword()));
return httpClient;
}
-
- // Contained classes
--------------------------------------------------------------------
-
- private boolean persistingToServer = false;
-
- /**
- * Sends outgoing JSON
- */
- private class Sender extends HttpMessageSender<JSONObject> {
-
- private List<JSONArray> messageQueue;
-
- /**
- * This variable keeps track of the number of calls to flush that were
made while
- * executing flush. It is used so that each flush call will be executed
individually.
- * The method flush does not continue flushing until the queue is empty
because the
- * last message in the queue may be incomplete, and transactions should
be sent
- * atomically.
- */
- private int flushDepth = 0;
-
- public Sender(HttpClient httpClient, SPServerInfo serverInfo, String
rootUUID) {
- super(httpClient, serverInfo, rootUUID);
- messageQueue = new ArrayList<JSONArray>();
- }
-
- public void clear() {
- if (messageQueue.size() > 0) {
- messageQueue.remove(0);
- }
- }
-
- public void flush() throws SPPersistenceException {
- try {
- synchronized (this) {
- if (persistingToServer) {
- flushDepth++;
- return;
- } else {
- if (messageQueue.size() > 0) {
- persistingToServer = true;
- } else {
- // If flush was called but no message was
given,
- // don't bother sending anything. (This is
not a
- // special precaution, it does happen.)
- persistingToServer = false;
- return;
- }
- }
- }
-
- URI serverURI = getServerURI();
- HttpPost postRequest = new HttpPost(serverURI);
- postRequest.setEntity(new
StringEntity(messageQueue.get(0).toString()));
-
- postRequest.setHeader("Content-Type", "application/json");
- HttpUriRequest request = postRequest;
- clear();
- final JSONMessage response =
getHttpClient().execute(request, new JSONResponseHandler());
-
- runInForeground(new Runnable() {
- public void run() {
- try {
- if (response.isSuccessful()) {
- // Message was sent successfully and accepted by
the server.
- currentRevision = (new
JSONObject(response.getBody())).getInt("currentRevision");
- } else {
- // Message was sent successfully but rejected by
the server. We must rollback our
- // changes and update to the head revision.
- logger.debug("Response unsuccessful");
- throw new RuntimeException("Out of sync with
server");
- }
- } catch (JSONException e) {
- throw new RuntimeException(e);
- } finally {
- synchronized (sender) {
- persistingToServer = false;
- try {
- if (flushDepth > 0) {
- flushDepth--;
- flush();
- }
- } catch (SPPersistenceException e) {
- throw new RuntimeException(e);
- }
- }
- }
- }
- });
- } catch (UnsupportedEncodingException e) {
- throw new SPPersistenceException(null, e);
- } catch (URISyntaxException e) {
- throw new SPPersistenceException(null, e);
- } catch (ClientProtocolException e) {
- throw new SPPersistenceException(null, e);
- } catch (IOException e) {
- throw new SPPersistenceException(null, e);
- } catch (RuntimeException e) {
- throw new SPPersistenceException(null, e);
- }
- }
-
- public void send(JSONObject content) throws
SPPersistenceException {
- // Transactions can start with a begin or a rollback
- if
(content.toString().equals("{\"uuid\":null,\"method\":\"begin\"}") ||
-
content.toString().equals("{\"uuid\":null,\"method\":\"rollback\"}")) {
- messageQueue.add(new JSONArray());
- }
- messageQueue.get(messageQueue.size() - 1).put(content);
- }
-
- public URI getServerURI() throws URISyntaxException {
- String contextPath = getServerInfo().getPath();
- return new URI("http", null, getServerInfo().getServerAddress(),
getServerInfo().getPort(),
- contextPath + "/project/" +
getProjectLocation().getUUID(), "currentRevision=" + currentRevision, null);
- }
- }
-
-
- /**
- * Polls this session's server for updates until interrupted. There
should
- * be exactly one instance of this class per ArchitectServerSession.
- */
- private class Updater extends Thread {
-
- /**
- * How long we will pause after an update error before
attempting to
- * contact the server again.
- */
- private long retryDelay = 1000;
-
- private final SPJSONMessageDecoder jsonDecoder;
-
- /**
- * Used by the Updater to handle inbound HTTP updates
- */
- private final HttpClient inboundHttpClient;
-
- final String contextRelativePath = "/project/" +
projectLocation.getUUID();
-
- private volatile boolean cancelled;
-
- /**
- * Creates, but does not start, the updater thread.
- *
- * @param projectUUID
- * the ID of the workspace this updater is responsible for.
This is
- * used in creating the thread's name.
- */
- Updater(String projectUUID, SPJSONMessageDecoder jsonDecoder) {
- super("updater-" + projectUUID);
- this.jsonDecoder = jsonDecoder;
- inboundHttpClient =
createHttpClient(projectLocation.getServiceInfo());
- }
-
- public void interrupt() {
- super.interrupt();
- cancelled = true;
- }
-
- boolean updating = false;
-
- @Override
- public void run() {
- logger.info("Updater thread starting");
- try {
- while (!this.isInterrupted() && !cancelled) {
- try {
- while (updating) {
- synchronized (this) {
- wait();
- }
- }
-
- URI uri = getServerURI(projectLocation.getServiceInfo(),
contextRelativePath,
- "oldRevisionNo=" +
currentRevision);
- HttpUriRequest request = new
HttpGet(uri);
-
- JSONMessage message =
inboundHttpClient.execute(request, new JSONResponseHandler());
- final JSONObject json = new
JSONObject(message.getBody());
- final String jsonArray = json.getString("data");
-
- updating = true;
- runInForeground(new Runnable() {
- public void run() {
- try {
- synchronized (sender) {
- if (!persistingToServer) {
- int newRevision =
json.getInt("currentRevision");
- if (currentRevision <
newRevision) {
- currentRevision =
newRevision;
-
jsonDecoder.decode(jsonArray);
- }
- }
- }
- } catch (SPPersistenceException e) {
- logger.error("Update from server
failed!", e);
- throw new RuntimeException("Please hit
the refresh button that does not exist", e);
- // TODO discard session and reload
- } catch (JSONException e) {
- logger.error("Update from server
failed!", e);
- } finally {
- updating = false;
- synchronized (updater) {
- updater.notify();
- }
- }
- }
- });
-
- } catch (Exception ex) {
- logger.error("Failed to contact server. Will retry in " + retryDelay
+ " ms.", ex);
- Thread.sleep(retryDelay);
- }
- }
- } catch (InterruptedException ex) {
- logger.info("Updater thread exiting normally due to
interruption.");
- }
-
- inboundHttpClient.getConnectionManager().shutdown();
- }
- }
private class DataSourceCollectionUpdater implements
DatabaseListChangeListener, PropertyChangeListener {
@@ -941,77 +718,6 @@
"data-sources/" + type + "/" + ds.getName());
}
}
-
- private static class JSONMessage {
- private final String message;
- private final boolean successful;
-
- public JSONMessage(String message, boolean successful) {
- this.message = message;
- this.successful = successful;
- }
-
- public String getBody() {
- return message;
- }
-
- public boolean isSuccessful() {
- return successful;
- }
- }
-
- private static class JSONResponseHandler implements
ResponseHandler<JSONMessage> {
-
- /*
- * Unsuccessful responses should have information sent in a header,
- * either as "unsuccessfulResponse" or "exceptionStackTrace"
- */
-
- public JSONMessage handleResponse(HttpResponse response) {
- try {
-
- BufferedReader reader = new BufferedReader(
- new
InputStreamReader(response.getEntity().getContent()));
- StringBuffer buffer = new StringBuffer();
-
- String line;
- while ((line = reader.readLine()) != null) {
- buffer.append(line).append("\n");
- }
- JSONObject message = new JSONObject(buffer.toString());
-
- // Does the response contain data? If so, return it.
Communication
- // with the resource has been successful.
- if (message.getString("responseKind").equals("data")) {
- return new JSONMessage(message.getString("data"),
true);
- } else {
- // Has the request been unsuccessful?
- if
(message.getString("responseKind").equals("unsuccessful")) {
- return new JSONMessage(message.getString("data"),
false);
- } else {
- // Does the response contain an exception? If so,
reconstruct, and then
- // re-throw it. There has been an exception on the
server.
- if
(message.getString("responseKind").equals("exceptionStackTrace")) {
-
- JSONArray stackTraceStrings = new
JSONArray(message.getString("data"));
- StringBuffer stackTraceMessage = new
StringBuffer();
- for (int i = 0; i <
stackTraceStrings.length(); i++) {
-
stackTraceMessage.append("\n").append(stackTraceStrings.get(i));
- }
-
- throw new
Exception(stackTraceMessage.toString());
-
- } else {
- // This exception represents a(n epic)
client-server miscommunication
- throw new Exception("Unable to parse
response ");
- }
- }
- }
- } catch (Exception ex) {
- throw new RuntimeException(ex);
- }
- }
- }
public void createRevisionSession(int revisionNo,
ArchitectSwingSession swingSession) {
// TODO Auto-generated method stub
=======================================
--- /trunk/src/ca/sqlpower/architect/swingui/dbtree/DBTreeModel.java Wed
Mar 10 08:21:50 2010
+++ /trunk/src/ca/sqlpower/architect/swingui/dbtree/DBTreeModel.java Fri
Mar 12 13:32:07 2010
@@ -242,6 +242,8 @@
private int transactionCount = 0;
public void childAdded(SPChildEvent e) {
+ if
(!SQLPowerUtils.getAncestorList(e.getSource()).contains(root)
&& !e.getSource().equals(root)) return;
+
if (!root.getSession().isForegroundThread())
throw new IllegalStateException("Adding a child " +
e.getChild() + " to " + e.getSource() +
" not on the foreground thread.");
@@ -285,6 +287,8 @@
}
public void childRemoved(SPChildEvent e) {
+ if
(!SQLPowerUtils.getAncestorList(e.getSource()).contains(root)
&& !e.getSource().equals(root)) return;
+
if (!root.getSession().isForegroundThread())
throw new IllegalStateException("Removing a child " +
e.getChild() + " to " + e.getSource() +
" not on the foreground thread.");
@@ -350,6 +354,9 @@
}
public void propertyChanged(PropertyChangeEvent e) {
+ if (!SQLPowerUtils.getAncestorList(((SPObject)
e.getSource())).contains(root) && !e.getSource().equals(root)) return;
+
+
if (!root.getSession().isForegroundThread())
throw new IllegalStateException("Changing the property" +
e.getPropertyName() + " on " + e.getSource() +
" not on the foreground thread.");
@@ -442,7 +449,12 @@
public DBTreeModel(SQLObjectRoot root) throws SQLObjectException {
this.root = root;
this.treeModelListeners = new LinkedList();
- SQLPowerUtils.listenToHierarchy(root, treeListener);
+ SQLPowerUtils.listenToHierarchy(root, treeListener);
+
+ for (SPObject ancestor : SQLPowerUtils.getAncestorList(root)) {
+ ancestor.addSPListener(treeListener);
+ }
+
setupTreeForNode(root);
}
@@ -486,7 +498,7 @@
if (logger.isDebugEnabled())
logger.debug("DBTreeModel.getChildCount("+parent+")"); //$NON-NLS-1$
//$NON-NLS-2$
if (parent instanceof FolderNode) {
- return ((FolderNode) parent).getChildren().size();
+ return ((FolderNode) parent).getChildren().size();
} else if (parent instanceof SQLTable) {
return foldersInTables.get((SQLTable) parent).size();
}
