improved reopen handling and concurrency

orientechnologies · Dec 22, 2015 · c0c2b3d · c0c2b3d
1 parent a9fa986
commit c0c2b3d
Showing 1 changed file with 56 additions and 46 deletions.
diff --git a/client/src/main/java/com/orientechnologies/orient/client/remote/OStorageRemote.java b/client/src/main/java/com/orientechnologies/orient/client/remote/OStorageRemote.java
@@ -157,6 +157,13 @@ public <T> T networkOperation(final OStorageRemoteOperation<T> operation, final
       OChannelBinaryAsynchClient network=null;
       try {
         network = getAvailableNetwork(getNextAvailableServerURL(false));
+
+        if (tokens.get(network.getServerURL()) == null && getSessionId() > 0) {
+          // TODO: Remove this workaround in favor of a proper per server authentication.
+          setSessionId(network.getServerURL(), -1, null);
+          openRemoteDatabase(network);
+        }
+
         return operation.execute(network);
       } catch (OModificationOperationProhibitedException mope) {
         handleDBFreeze();
@@ -1865,58 +1872,67 @@ protected synchronized String openRemoteDatabase() throws IOException {
     return openRemoteDatabase(currentURL);
   }
 
-  protected String openRemoteDatabase(String currentURL) {
-    do {
-      do {
-        OChannelBinaryAsynchClient network = null;
-        try {
-          network = getAvailableNetwork(currentURL);
-          try {
-            network.writeByte(OChannelBinaryProtocol.REQUEST_DB_OPEN);
-            network.writeInt(getSessionId());
+  protected void openRemoteDatabase(OChannelBinaryAsynchClient network) throws IOException {
+    stateLock.acquireWriteLock();
+    try {
 
-            // @SINCE 1.0rc8
-            sendClientInfo(network);
+      try {
+        network.writeByte(OChannelBinaryProtocol.REQUEST_DB_OPEN);
+        network.writeInt(getSessionId());
 
-            network.writeString(name);
-            network.writeString(connectionUserName);
-            network.writeString(connectionUserPassword);
+        // @SINCE 1.0rc8
+        sendClientInfo(network);
 
-          } finally {
-            endRequest(network);
-          }
+        network.writeString(name);
+        network.writeString(connectionUserName);
+        network.writeString(connectionUserPassword);
 
-          final int sessionId;
+      } finally {
+        endRequest(network);
+      }
 
-          try {
-            network.beginResponse(getSessionId(), false);
-            sessionId = network.readInt();
-            byte[] token = network.readBytes();
-            if (token.length == 0) {
-              token = null;
-            } else {
-              network.getServiceThread().setTokenBased(true);
-            }
-            setSessionId(currentURL, sessionId, token);
+      final int sessionId;
 
-            OLogManager.instance().debug(this, "Client connected to %s with session id=%d", currentURL, sessionId);
+      try {
+        network.beginResponse(getSessionId(), false);
+        sessionId = network.readInt();
+        byte[] token = network.readBytes();
+        if (token.length == 0) {
+          token = null;
+        } else {
+          network.getServiceThread().setTokenBased(true);
+        }
+        setSessionId(network.getServerURL(), sessionId, token);
 
-            readDatabaseInformation(network);
+        OLogManager.instance().debug(this, "Client connected to %s with session id=%d", network.getServerURL(), sessionId);
 
-            // READ CLUSTER CONFIGURATION
-            updateClusterConfiguration(currentURL, network.readBytes());
+        readDatabaseInformation(network);
 
-            // read OrientDB release info
-            if (network.getSrvProtocolVersion() >= 14)
-              network.readString();
+        // READ CLUSTER CONFIGURATION
+        updateClusterConfiguration(network.getServerURL(), network.readBytes());
 
-            status = STATUS.OPEN;
+        // read OrientDB release info
+        if (network.getSrvProtocolVersion() >= 14)
+          network.readString();
 
-            return currentURL;
+        status = STATUS.OPEN;
 
-          } finally {
-            endResponse(network);
-          }
+      } finally {
+        endResponse(network);
+      }
+    }finally {
+      stateLock.releaseWriteLock();
+    }
+  }
+
+  protected String openRemoteDatabase(String currentURL) {
+    do {
+      do {
+        OChannelBinaryAsynchClient network = null;
+        try {
+          network = getAvailableNetwork(currentURL);
+          openRemoteDatabase(network);
+          return currentURL;
         } catch (OIOException e) {
           if (network != null) {
             // REMOVE THE NETWORK CONNECTION IF ANY
@@ -2190,12 +2206,6 @@ protected OChannelBinaryAsynchClient getAvailableNetwork(final String iCurrentUR
         cause = e;
       }
 
-      if (tokens.get(lastURL) == null && getSessionId() > 0) {
-        // TODO: Remove this workaround in favor of a proper per server authentication.
-        setSessionId(lastURL, -1, null);
-        throw new OTokenException("Missing token, reconnect is required");
-      }
-
       if (network == null) {
         lastURL = useNewServerURL(lastURL);
         if (lastURL == null) {