0002699: Copy directly to staging when nodes are on same server

JumpMind · Jul 29, 2016 · 80ae9da · 80ae9da
1 parent 2417187
commit 80ae9da
Show file tree

Hide file tree

Showing 3 changed files with 36 additions and 1 deletion.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java
@@ -145,6 +145,7 @@ private ParameterConstants() {
     public final static String CONCURRENT_RESERVATION_TIMEOUT = "http.concurrent.reservation.timeout.ms";
 
     public final static String OUTGOING_BATCH_PEEK_AHEAD_BATCH_COMMIT_SIZE = "outgoing.batches.peek.ahead.batch.commit.size";
+    public final static String OUTGOING_BATCH_COPY_TO_INCOMING_STAGING = "outgoing.batches.copy.to.incoming.staging";
     public final static String ROUTING_FLUSH_JDBC_BATCH_SIZE = "routing.flush.jdbc.batch.size";
     public final static String ROUTING_WAIT_FOR_DATA_TIMEOUT_SECONDS = "routing.wait.for.data.timeout.seconds";
     public final static String ROUTING_MAX_GAPS_TO_QUALIFY_IN_SQL = "routing.max.gaps.to.qualify.in.sql";

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java
@@ -20,6 +20,7 @@
  */
 package org.jumpmind.symmetric.service.impl;
 
+import java.io.BufferedReader;
 import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -48,6 +49,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
 import org.jumpmind.db.io.DatabaseXmlUtil;
 import org.jumpmind.db.model.Column;
@@ -61,6 +63,7 @@
 import org.jumpmind.db.sql.ISqlRowMapper;
 import org.jumpmind.db.sql.ISqlTransaction;
 import org.jumpmind.db.sql.Row;
+import org.jumpmind.symmetric.AbstractSymmetricEngine;
 import org.jumpmind.symmetric.ISymmetricEngine;
 import org.jumpmind.symmetric.SymmetricException;
 import org.jumpmind.symmetric.Version;
@@ -922,7 +925,29 @@ protected OutgoingBatch sendOutgoingBatch(ProcessInfo processInfo, Node targetNo
 
             IStagedResource extractedBatch = getStagedResource(currentBatch);
             if (extractedBatch != null) {
-                if (mode == ExtractMode.FOR_SYM_CLIENT && writer != null) {
+                if (mode == ExtractMode.FOR_SYM_CLIENT && writer != null) {                   
+                    if (!isRetry && parameterService.is(ParameterConstants.OUTGOING_BATCH_COPY_TO_INCOMING_STAGING)) {
+                        ISymmetricEngine targetEngine = AbstractSymmetricEngine.findEngineByUrl(targetNode.getSyncUrl());
+                        if (targetEngine != null) {
+                            try {
+                                long memoryThresholdInBytes = parameterService.getLong(ParameterConstants.STREAM_TO_FILE_THRESHOLD);
+                                Node sourceNode = nodeService.findIdentity();
+                                IStagedResource targetResource = targetEngine.getStagingManager().create(memoryThresholdInBytes, 
+                                        Constants.STAGING_CATEGORY_INCOMING, Batch.getStagedLocation(false, sourceNode.getNodeId()), 
+                                        currentBatch.getBatchId());
+                                BufferedReader sourceReader = extractedBatch.getReader();
+                                BufferedWriter targetWriter = targetResource.getWriter();
+                                IOUtils.copy(sourceReader, targetWriter);
+                                extractedBatch.close();
+                                targetResource.close();
+                                targetResource.setState(State.READY);
+                                isRetry = true;
+                            } catch (Exception e) {
+                                throw new RuntimeException(e);
+                            }
+                        }
+                    }
+
                     Channel channel = configurationService.getChannel(currentBatch.getChannelId());
                     DataContext ctx = new DataContext();
                     SimpleStagingDataReader dataReader = new SimpleStagingDataReader(BatchType.EXTRACT, currentBatch.getBatchId(), 

diff --git a/symmetric-core/src/main/resources/symmetric-default.properties b/symmetric-core/src/main/resources/symmetric-default.properties
@@ -1002,6 +1002,15 @@ routing.stale.gap.busy.expire.time.ms=1200000
 # Tags: extract
 outgoing.batches.peek.ahead.batch.commit.size=10
 
+# When sending an outgoing batch, copy directly from the outgoing staging to the incoming staging
+# when both nodes are on the same server.  This also requires the staging to be enabled
+# (stream.to.file.enabled=true).  The HTTP transport is still used to send a batch
+# "retry" instruction that causes the target node to read from staging.
+#
+# DatabaseOverridable: true
+# Type: boolean
+# Tags: extract
+outgoing.batches.copy.to.incoming.staging=true
 
 # Update the outgoing batch status to QY (querying) and SE (sending) only when
 # the last update to the batch is in the past by at least the specified number of milliseconds.