[328595] TimeoutException during raw replication

https://bugs.eclipse.org/bugs/show_bug.cgi?id=328595

4 files changed, 198 insertions, 123 deletions

diff --git a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientProtocol.java b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientProtocol.java
index 66f28411cd..b87f401ddd 100644
--- a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientProtocol.java
+++ b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientProtocol.java
@@ -343,7 +343,7 @@ public class CDOClientProtocol extends SignalProtocol<CDOSession> implements CDO
 
   public void replicateRepositoryRaw(CDORawReplicationContext context, OMMonitor monitor)
   {
-    send(new ReplicateRepositoryRawRequest(this, context, monitor));
+    send(new ReplicateRepositoryRawRequest(this, context), monitor);
   }
 
   public CDOChangeSetData[] loadChangeSets(CDOBranchPointRange... ranges)
@@ -404,6 +404,22 @@ public class CDOClientProtocol extends SignalProtocol<CDOSession> implements CDO
     }
   }
 
+  private Boolean send(ReplicateRepositoryRawRequest request, OMMonitor monitor)
+  {
+    try
+    {
+      return request.send(monitor);
+    }
+    catch (RuntimeException ex)
+    {
+      throw ex;
+    }
+    catch (Exception ex)
+    {
+      throw new TransportException(ex);
+    }
+  }
+
   private CommitTransactionResult send(CommitTransactionRequest request, OMMonitor monitor)
   {
     try
diff --git a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientRequestWithMonitoring.java b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientRequestWithMonitoring.java
new file mode 100644
index 0000000000..d96641f432
--- /dev/null
+++ b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CDOClientRequestWithMonitoring.java
@@ -0,0 +1,167 @@
+/***************************************************************************
+ * Copyright (c) 2004 - 2010 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ *    Simon McDuff - bug 201266
+ *    Simon McDuff - bug 215688
+ *    Simon McDuff - bug 213402
+ *    Andre Dietisheim - bug 256649
+ **************************************************************************/
+package org.eclipse.emf.cdo.internal.net4j.protocol;
+
+import org.eclipse.emf.cdo.common.branch.CDOBranchManager;
+import org.eclipse.emf.cdo.common.commit.CDOCommitInfoManager;
+import org.eclipse.emf.cdo.common.id.CDOIDProvider;
+import org.eclipse.emf.cdo.common.model.CDOPackageRegistry;
+import org.eclipse.emf.cdo.common.model.lob.CDOLobStore;
+import org.eclipse.emf.cdo.common.protocol.CDODataInput;
+import org.eclipse.emf.cdo.common.protocol.CDODataOutput;
+import org.eclipse.emf.cdo.common.revision.CDOListFactory;
+import org.eclipse.emf.cdo.common.revision.CDORevisionFactory;
+import org.eclipse.emf.cdo.internal.common.protocol.CDODataInputImpl;
+import org.eclipse.emf.cdo.internal.common.protocol.CDODataOutputImpl;
+
+import org.eclipse.emf.internal.cdo.revision.CDOListWithElementProxiesImpl;
+
+import org.eclipse.net4j.signal.RequestWithMonitoring;
+import org.eclipse.net4j.util.io.ExtendedDataInputStream;
+import org.eclipse.net4j.util.io.ExtendedDataOutputStream;
+import org.eclipse.net4j.util.io.StringIO;
+import org.eclipse.net4j.util.om.monitor.OMMonitor;
+
+import org.eclipse.emf.spi.cdo.InternalCDOSession;
+
+import java.io.IOException;
+
+/**
+ * @author Eike Stepper
+ */
+public abstract class CDOClientRequestWithMonitoring<RESULT> extends RequestWithMonitoring<RESULT>
+{
+  private ExtendedDataOutputStream requestStream;
+
+  private ExtendedDataInputStream confirmationStream;
+
+  public CDOClientRequestWithMonitoring(CDOClientProtocol protocol, short signalID)
+  {
+    super(protocol, signalID);
+  }
+
+  @Override
+  public CDOClientProtocol getProtocol()
+  {
+    return (CDOClientProtocol)super.getProtocol();
+  }
+
+  protected ExtendedDataOutputStream getRequestStream()
+  {
+    return requestStream;
+  }
+
+  protected ExtendedDataInputStream getConfirmationStream()
+  {
+    return confirmationStream;
+  }
+
+  protected InternalCDOSession getSession()
+  {
+    return (InternalCDOSession)getProtocol().getSession();
+  }
+
+  protected CDOIDProvider getIDProvider()
+  {
+    return null;
+  }
+
+  @Override
+  protected int getMonitorProgressSeconds()
+  {
+    org.eclipse.emf.cdo.net4j.CDOSession session = (org.eclipse.emf.cdo.net4j.CDOSession)getSession();
+    return session.options().getProgressInterval();
+  }
+
+  @Override
+  protected final void requesting(ExtendedDataOutputStream out, OMMonitor monitor) throws Exception
+  {
+    requestStream = out;
+    requesting(new CDODataOutputImpl(out)
+    {
+      @Override
+      public CDOPackageRegistry getPackageRegistry()
+      {
+        return getSession().getPackageRegistry();
+      }
+
+      @Override
+      public CDOIDProvider getIDProvider()
+      {
+        return CDOClientRequestWithMonitoring.this.getIDProvider();
+      }
+
+      @Override
+      protected StringIO getPackageURICompressor()
+      {
+        return getProtocol().getPackageURICompressor();
+      }
+    }, monitor);
+  }
+
+  @Override
+  protected final RESULT confirming(ExtendedDataInputStream in, OMMonitor monitor) throws Exception
+  {
+    confirmationStream = in;
+    return confirming(new CDODataInputImpl(in)
+    {
+      @Override
+      protected StringIO getPackageURICompressor()
+      {
+        return getProtocol().getPackageURICompressor();
+      }
+
+      @Override
+      protected CDOPackageRegistry getPackageRegistry()
+      {
+        return getSession().getPackageRegistry();
+      }
+
+      @Override
+      protected CDOBranchManager getBranchManager()
+      {
+        return getSession().getBranchManager();
+      }
+
+      @Override
+      protected CDOCommitInfoManager getCommitInfoManager()
+      {
+        return getSession().getCommitInfoManager();
+      }
+
+      @Override
+      protected CDORevisionFactory getRevisionFactory()
+      {
+        return getSession().getRevisionManager().getFactory();
+      }
+
+      @Override
+      protected CDOLobStore getLobStore()
+      {
+        return getSession().getLobStore();
+      }
+
+      @Override
+      protected CDOListFactory getListFactory()
+      {
+        return CDOListWithElementProxiesImpl.FACTORY;
+      }
+    }, monitor);
+  }
+
+  protected abstract void requesting(CDODataOutput out, OMMonitor monitor) throws IOException;
+
+  protected abstract RESULT confirming(CDODataInput in, OMMonitor monitor) throws IOException;
+}
diff --git a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CommitTransactionRequest.java b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CommitTransactionRequest.java
index 45d2aacb23..5af184e81d 100644
--- a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CommitTransactionRequest.java
+++ b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/CommitTransactionRequest.java
@@ -15,10 +15,8 @@
 package org.eclipse.emf.cdo.internal.net4j.protocol;
 
 import org.eclipse.emf.cdo.CDOObject;
-import org.eclipse.emf.cdo.common.branch.CDOBranchManager;
 import org.eclipse.emf.cdo.common.branch.CDOBranchPoint;
 import org.eclipse.emf.cdo.common.commit.CDOCommitData;
-import org.eclipse.emf.cdo.common.commit.CDOCommitInfoManager;
 import org.eclipse.emf.cdo.common.id.CDOID;
 import org.eclipse.emf.cdo.common.id.CDOIDAndVersion;
 import org.eclipse.emf.cdo.common.id.CDOIDMetaRange;
@@ -26,40 +24,28 @@ import org.eclipse.emf.cdo.common.id.CDOIDProvider;
 import org.eclipse.emf.cdo.common.id.CDOIDTemp;
 import org.eclipse.emf.cdo.common.id.CDOIDUtil;
 import org.eclipse.emf.cdo.common.model.CDOPackageInfo;
-import org.eclipse.emf.cdo.common.model.CDOPackageRegistry;
 import org.eclipse.emf.cdo.common.model.CDOPackageUnit;
 import org.eclipse.emf.cdo.common.model.lob.CDOBlob;
 import org.eclipse.emf.cdo.common.model.lob.CDOClob;
 import org.eclipse.emf.cdo.common.model.lob.CDOLob;
-import org.eclipse.emf.cdo.common.model.lob.CDOLobStore;
 import org.eclipse.emf.cdo.common.protocol.CDODataInput;
 import org.eclipse.emf.cdo.common.protocol.CDODataOutput;
 import org.eclipse.emf.cdo.common.protocol.CDOProtocolConstants;
-import org.eclipse.emf.cdo.common.revision.CDOListFactory;
 import org.eclipse.emf.cdo.common.revision.CDORevision;
-import org.eclipse.emf.cdo.common.revision.CDORevisionFactory;
 import org.eclipse.emf.cdo.common.revision.CDORevisionKey;
 import org.eclipse.emf.cdo.common.revision.delta.CDORevisionDelta;
-import org.eclipse.emf.cdo.internal.common.protocol.CDODataInputImpl;
-import org.eclipse.emf.cdo.internal.common.protocol.CDODataOutputImpl;
 import org.eclipse.emf.cdo.internal.net4j.bundle.OM;
 import org.eclipse.emf.cdo.spi.common.model.InternalCDOPackageInfo;
 import org.eclipse.emf.cdo.spi.common.model.InternalCDOPackageRegistry.MetaInstanceMapper;
 import org.eclipse.emf.cdo.transaction.CDOTransaction;
 
-import org.eclipse.emf.internal.cdo.revision.CDOListWithElementProxiesImpl;
-
-import org.eclipse.net4j.signal.RequestWithMonitoring;
-import org.eclipse.net4j.util.io.ExtendedDataInputStream;
 import org.eclipse.net4j.util.io.ExtendedDataOutputStream;
 import org.eclipse.net4j.util.io.IOUtil;
-import org.eclipse.net4j.util.io.StringIO;
 import org.eclipse.net4j.util.om.monitor.OMMonitor;
 import org.eclipse.net4j.util.om.trace.ContextTracer;
 
 import org.eclipse.emf.ecore.EClass;
 import org.eclipse.emf.spi.cdo.CDOSessionProtocol.CommitTransactionResult;
-import org.eclipse.emf.spi.cdo.InternalCDOSession;
 
 import java.io.IOException;
 import java.io.OutputStreamWriter;
@@ -69,7 +55,7 @@ import java.util.List;
 /**
  * @author Eike Stepper
  */
-public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransactionResult>
+public class CommitTransactionRequest extends CDOClientRequestWithMonitoring<CommitTransactionResult>
 {
   private static final ContextTracer TRACER = new ContextTracer(OM.DEBUG_PROTOCOL, CommitTransactionRequest.class);
 
@@ -85,8 +71,6 @@ public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransa
 
   private Collection<CDOLob<?>> lobs;
 
-  private ExtendedDataOutputStream stream;
-
   public CommitTransactionRequest(CDOClientProtocol protocol, int transactionID, String comment, boolean releaseLocks,
       CDOIDProvider idProvider, CDOCommitData commitData, Collection<CDOLob<?>> lobs)
   {
@@ -107,47 +91,19 @@ public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransa
   }
 
   @Override
-  public CDOClientProtocol getProtocol()
-  {
-    return (CDOClientProtocol)super.getProtocol();
-  }
-
-  protected InternalCDOSession getSession()
+  protected int getMonitorTimeoutSeconds()
   {
-    return (InternalCDOSession)getProtocol().getSession();
+    org.eclipse.emf.cdo.net4j.CDOSession session = (org.eclipse.emf.cdo.net4j.CDOSession)getSession();
+    return session.options().getCommitTimeout();
   }
 
+  @Override
   protected CDOIDProvider getIDProvider()
   {
     return idProvider;
   }
 
   @Override
-  protected final void requesting(ExtendedDataOutputStream out, OMMonitor monitor) throws Exception
-  {
-    stream = out;
-    requesting(new CDODataOutputImpl(out)
-    {
-      @Override
-      public CDOPackageRegistry getPackageRegistry()
-      {
-        return getSession().getPackageRegistry();
-      }
-
-      @Override
-      public CDOIDProvider getIDProvider()
-      {
-        return CommitTransactionRequest.this.getIDProvider();
-      }
-
-      @Override
-      protected StringIO getPackageURICompressor()
-      {
-        return getProtocol().getPackageURICompressor();
-      }
-    }, monitor);
-  }
-
   protected void requesting(CDODataOutput out, OMMonitor monitor) throws IOException
   {
     requestingTransactionInfo(out);
@@ -220,11 +176,12 @@ public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransa
       }
     }
 
-    requestingLobs(out);
+    requestingLobs();
   }
 
-  protected void requestingLobs(CDODataOutput out) throws IOException
+  protected void requestingLobs() throws IOException
   {
+    ExtendedDataOutputStream out = getRequestStream();
     out.writeInt(lobs.size());
     for (CDOLob<?> lob : lobs)
     {
@@ -234,13 +191,13 @@ public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransa
       {
         CDOBlob blob = (CDOBlob)lob;
         out.writeLong(size);
-        IOUtil.copyBinary(blob.getContents(), stream, size);
+        IOUtil.copyBinary(blob.getContents(), out, size);
       }
       else
       {
         CDOClob clob = (CDOClob)lob;
         out.writeLong(-size);
-        IOUtil.copyCharacter(clob.getContents(), new OutputStreamWriter(stream), size);
+        IOUtil.copyCharacter(clob.getContents(), new OutputStreamWriter(out), size);
       }
     }
   }
@@ -253,54 +210,6 @@ public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransa
   }
 
   @Override
-  protected final CommitTransactionResult confirming(ExtendedDataInputStream in, OMMonitor monitor) throws Exception
-  {
-    return confirming(new CDODataInputImpl(in)
-    {
-      @Override
-      protected StringIO getPackageURICompressor()
-      {
-        return getProtocol().getPackageURICompressor();
-      }
-
-      @Override
-      protected CDOPackageRegistry getPackageRegistry()
-      {
-        return getSession().getPackageRegistry();
-      }
-
-      @Override
-      protected CDOBranchManager getBranchManager()
-      {
-        return getSession().getBranchManager();
-      }
-
-      @Override
-      protected CDOCommitInfoManager getCommitInfoManager()
-      {
-        return getSession().getCommitInfoManager();
-      }
-
-      @Override
-      protected CDORevisionFactory getRevisionFactory()
-      {
-        return getSession().getRevisionManager().getFactory();
-      }
-
-      @Override
-      protected CDOLobStore getLobStore()
-      {
-        return getSession().getLobStore();
-      }
-
-      @Override
-      protected CDOListFactory getListFactory()
-      {
-        return CDOListWithElementProxiesImpl.FACTORY;
-      }
-    }, monitor);
-  }
-
   protected CommitTransactionResult confirming(CDODataInput in, OMMonitor monitor) throws IOException
   {
     CommitTransactionResult result = confirmingCheckError(in);
@@ -383,18 +292,4 @@ public class CommitTransactionRequest extends RequestWithMonitoring<CommitTransa
       }
     }
   }
-
-  @Override
-  protected int getMonitorProgressSeconds()
-  {
-    org.eclipse.emf.cdo.net4j.CDOSession session = (org.eclipse.emf.cdo.net4j.CDOSession)getSession();
-    return session.options().getProgressInterval();
-  }
-
-  @Override
-  protected int getMonitorTimeoutSeconds()
-  {
-    org.eclipse.emf.cdo.net4j.CDOSession session = (org.eclipse.emf.cdo.net4j.CDOSession)getSession();
-    return session.options().getCommitTimeout();
-  }
 }
diff --git a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/ReplicateRepositoryRawRequest.java b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/ReplicateRepositoryRawRequest.java
index d699c32f77..d89632f0bd 100644
--- a/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/ReplicateRepositoryRawRequest.java
+++ b/plugins/org.eclipse.emf.cdo.net4j/src/org/eclipse/emf/cdo/internal/net4j/protocol/ReplicateRepositoryRawRequest.java
@@ -22,28 +22,25 @@ import java.io.IOException;
 /**
  * @author Eike Stepper
  */
-public class ReplicateRepositoryRawRequest extends CDOClientRequest<Boolean>
+public class ReplicateRepositoryRawRequest extends CDOClientRequestWithMonitoring<Boolean>
 {
   private CDORawReplicationContext context;
 
-  private OMMonitor monitor;
-
-  public ReplicateRepositoryRawRequest(CDOClientProtocol protocol, CDORawReplicationContext context, OMMonitor monitor)
+  public ReplicateRepositoryRawRequest(CDOClientProtocol protocol, CDORawReplicationContext context)
   {
     super(protocol, CDOProtocolConstants.SIGNAL_REPLICATE_REPOSITORY_RAW);
     this.context = context;
-    this.monitor = monitor;
   }
 
   @Override
-  protected void requesting(CDODataOutput out) throws IOException
+  protected void requesting(CDODataOutput out, OMMonitor monitor) throws IOException
   {
     out.writeInt(context.getLastReplicatedBranchID());
     out.writeLong(context.getLastReplicatedCommitTime());
   }
 
   @Override
-  protected Boolean confirming(CDODataInput in) throws IOException
+  protected Boolean confirming(CDODataInput in, OMMonitor monitor) throws IOException
   {
     context.replicateRaw(in, monitor);
     return true;