Making the content syncs more resilient

gradle.properties

@@ -5,9 +5,11 @@ org.gradle.jvmargs=-XX:MaxPermSize=512m -XX:+CMSClassUnloadingEnabled -XX:+CMSPe
 bundleInstallRoot = /apps/grabbit/install
 
 group = com.twcable.grabbit
-version = 7.1.5
+version = 7.1.6
 
 # Please keep alphabetical
+apache_httpclient_version = 4.5.4
+apache_httpcore_version = 4.4.8
 cglib_nodep_version = 2.2.2
 commons_collections_version = 3.2.1
 commons_io_version = 1.4
@@ -41,6 +43,7 @@ servlet_api_version = 2.5
 slf4j_version = 1.7.6
 sling_api_version = 2.9.0
 sling_base_version = 2.2.2
+sling_commons_json_version = 2.0.6
 sling_commons_testing_version = 2.0.12
 sling_commons_version = 2.2.0
 sling_event_version = 3.1.4

gradle/bundle.gradle

@@ -27,6 +27,9 @@ jar.manifest {
     attributes 'Bundle-Name': project.bundleName
     attributes 'Bundle-SymbolicName': project.symbolicName
     attributes 'Bundle-Description': project.bundleDescription
+    instruction 'Import-Package', 'org.apache.sling.commons.json; version="[2.0.4,3.0)"'
+    instruction 'Import-Package', 'org.apache.http; version="[4.0,5.0)"'
+    instruction 'Import-Package', 'org.apache.http.client; version="[4.0,5.0)"'
     instruction 'Import-Package', 'groovy.json; version="[2.3,3.0)"'
     instruction 'Import-Package', 'groovy.json.internal; version="[2.3,3.0)"'
     instruction 'Import-Package', 'org.springframework.batch.core.scope; version="[2.2,3.0)"'

gradle/dependencies.gradle

@@ -23,6 +23,11 @@ dependencies {
     }
     compile "org.apache.servicemix.bundles:org.apache.servicemix.bundles.okio:${okio_version}"
 
+    compile "org.apache.httpcomponents:httpcore-osgi:${apache_httpcore_version}"
+    compile "org.apache.httpcomponents:httpclient-osgi:${apache_httpclient_version}"
+
+    // JSON Commons Library
+    compile "org.apache.sling:org.apache.sling.commons.json:${sling_commons_json_version}"
 
     // Apache Sling libraries
     compile "org.apache.sling:org.apache.sling.api:${sling_api_version}"

gradle/packageExclusions.gradle

@@ -39,6 +39,15 @@ configurations.cq_package {
     exclude group: 'org.apache.sling', module:'org.apache.sling.jcr.resource'
     exclude group: 'org.apache.sling', module: 'org.apache.sling.jcr.api'
 
+    // Exclude the Sling Commons JSON Library
+    exclude group: 'org.apache.sling', module: 'org.apache.sling.commons.json'
+
+    // HTTP Core OSGI
+    // 4.4.8
+    exclude group: 'org.apache.httpcomponents', module: 'httpcore-osgi'
+    // 4.5.4
+    exclude group: 'org.apache.httpcomponents', module: 'httpclient-osgi'
+
     // 6.x exclude bundles
     exclude group: 'com.google.guava', module: 'guava'
     exclude module: 'cq-workflow-console'

src/main/groovy/com/twcable/grabbit/GrabbitConfiguration.groovy

@@ -20,6 +20,7 @@ import com.google.common.collect.ImmutableMap
 import com.twcable.grabbit.util.CryptoUtil
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import org.apache.commons.lang3.builder.ToStringBuilder
 import org.yaml.snakeyaml.Yaml
 
 import javax.annotation.Nonnull
@@ -110,7 +111,7 @@ class GrabbitConfiguration {
 
         if (errorBuilder.hasErrors()) throw errorBuilder.build()
 
-        return new GrabbitConfiguration(
+        GrabbitConfiguration config = new GrabbitConfiguration(
             serverUsername,
             serverPassword,
             serverScheme,
@@ -119,6 +120,10 @@ class GrabbitConfiguration {
             deltaContent,
             pathConfigurations.asImmutable()
         )
+
+        log.info "# GrabbitConfiguration #\n${config}"
+
+        return config
     }
 
     private static final Pattern prePattern = Pattern.compile(/^(\/|\.\/|\\).*$/)
@@ -269,4 +274,8 @@ class GrabbitConfiguration {
         }
     }
 
+    @Override
+    public String toString() {
+        return ToStringBuilder.reflectionToString(this);
+    }
 }

src/main/groovy/com/twcable/grabbit/client/batch/steps/http/CreateHttpConnectionTasklet.groovy

@@ -23,6 +23,12 @@ import groovy.util.logging.Slf4j
 import okhttp3.*
 import okhttp3.OkHttpClient.Builder as HttpClientBuilder
 import okhttp3.Request.Builder as RequestBuilder
+import org.apache.http.HttpEntity
+import org.apache.http.entity.AbstractHttpEntity
+import org.apache.http.entity.StringEntity
+import org.apache.sling.commons.json.JSONArray
+import org.apache.sling.commons.json.JSONException
+import org.apache.sling.commons.json.JSONObject
 import org.springframework.batch.core.ExitStatus
 import org.springframework.batch.core.StepContribution
 import org.springframework.batch.core.scope.context.ChunkContext
@@ -69,23 +75,66 @@ class CreateHttpConnectionTasklet implements Tasklet {
 
 
     Connection createConnection(@Nonnull final Map jobParameters) {
+        log.info "createConnection : START"
 
         final String username = (String)jobParameters.get(ClientBatchJob.SERVER_USERNAME)
         final String password = (String)jobParameters.get(ClientBatchJob.SERVER_PASSWORD)
 
-        final Request request = new RequestBuilder()
-                .url(getURLForRequest(jobParameters))
+        final Request request = new RequestBuilder().post(formatRequestBody(jobParameters))
+                .url(getPostURLForRequest(jobParameters))
                 .addHeader('Authorization', Credentials.basic(username, password))
                 .build()
+//        final Request request = new RequestBuilder()
+//                .url(getURLForRequest(jobParameters))
+//                .addHeader('Authorization', Credentials.basic(username, password))
+//                .build()
 
 
         final OkHttpClient client = getNewHttpClient()
 
         final Response response = client.newCall(request).execute()
+        log.info "createConnection : response=${response}"
+        log.info "createConnection : body=${response.body()}"
         //We return response information in a connection like this because it's clear, but also because Response is a final class that we can not easily mock
         return new Connection(response.body().byteStream(), response.networkResponse(), response.code())
     }
 
+    /**
+     * Format the JSON request body to send the params of the invalidationPath
+     * @param invalidationPath
+     * @return
+     */
+    private RequestBody formatRequestBody(@Nonnull final Map jobParameters) {
+        try {
+            //addQueryParameter will encode these values for us
+            String path = (String)jobParameters.get(ClientBatchJob.PATH)
+            String after = (String)jobParameters.get(ClientBatchJob.CONTENT_AFTER_DATE) ?: '';
+
+            final String excludePathParam = jobParameters.get(ClientBatchJob.EXCLUDE_PATHS)
+            final excludePaths = (excludePathParam != null && !excludePathParam.isEmpty() ? excludePathParam.split(/\*/) : Collections.EMPTY_LIST) as Collection<String>
+            List<String> excludePathsList = new ArrayList<>()
+            for(String excludePath : excludePaths) {
+                excludePathsList.add(excludePath)
+            }
+
+            JSONObject json = new JSONObject();
+            json.put("path", path);
+            json.put("after", after);
+
+            JSONArray excludePathsArray = new JSONArray(excludePathsList);
+
+            json.put("excludePaths", excludePathsArray);
+
+            RequestBody requestBody = RequestBody.create(MediaType.parse("application/json"), json.toString());
+            return requestBody;
+        } catch (IOException ioe) {
+            log.error(ioe.getMessage(), ioe);
+        } catch (JSONException e) {
+            log.error("Exception occurred forming JSON", e);
+        }
+        return null;
+    }
+
 
     HttpUrl getURLForRequest(@Nonnull final Map jobParameters) {
         HttpUrlBuilder urlBuilder = new HttpUrl.Builder()
@@ -108,6 +157,17 @@ class CreateHttpConnectionTasklet implements Tasklet {
         return urlBuilder.build()
     }
 
+    HttpUrl getPostURLForRequest(@Nonnull final Map jobParameters) {
+        HttpUrlBuilder urlBuilder = new HttpUrl.Builder()
+
+        urlBuilder.scheme((String)jobParameters.get(ClientBatchJob.SCHEME))
+        urlBuilder.host((String)jobParameters.get(ClientBatchJob.HOST))
+        urlBuilder.port(Integer.parseInt((String)jobParameters.get(ClientBatchJob.PORT)))
+        urlBuilder.encodedPath('/grabbit/content')
+
+        return urlBuilder.build()
+    }
+
 
     private OkHttpClient getNewHttpClient() {
         return new HttpClientBuilder()

src/main/groovy/com/twcable/grabbit/client/batch/steps/jcrnodes/JcrNodesReader.groovy

@@ -37,11 +37,19 @@ class JcrNodesReader implements ItemReader<ProtoNode> {
 
     @Override
     NodeProtos.Node read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
-        ProtoNode nodeProto = ProtoNode.parseDelimitedFrom(theInputStream())
-        if (!nodeProto) {
-            log.info "Received all data from Server"
-            return null
+        ProtoNode nodeProto
+        try {
+            log.trace "JcrNodesReader.read() : START"
+            nodeProto = ProtoNode.parseDelimitedFrom(theInputStream())
+            if (!nodeProto) {
+                log.info "Received all data from Server"
+                return null
+            }
+        } catch (Exception e) {
+            log.error "Exception occurred parsing from the inputStream\n${e}", e
         }
+        log.debug "read() : NodeProto: \n${nodeProto.name}"
+
         return nodeProto
     }
 

src/main/groovy/com/twcable/grabbit/client/batch/steps/jcrnodes/JcrNodesWriter.groovy

@@ -17,16 +17,26 @@
 package com.twcable.grabbit.client.batch.steps.jcrnodes
 
 import com.twcable.grabbit.client.batch.ClientBatchJobContext
-import com.twcable.grabbit.jcr.JCRNodeDecorator
+
 import com.twcable.grabbit.jcr.ProtoNodeDecorator
 import com.twcable.grabbit.proto.NodeProtos.Node as ProtoNode
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import org.apache.commons.lang3.StringUtils
 import org.springframework.batch.core.ItemWriteListener
 import org.springframework.batch.item.ItemWriter
 import org.springframework.util.StopWatch
 
+import javax.jcr.AccessDeniedException
+import javax.jcr.InvalidItemStateException
+import javax.jcr.ItemExistsException
+import javax.jcr.ReferentialIntegrityException
+import javax.jcr.RepositoryException
 import javax.jcr.Session
+import javax.jcr.lock.LockException
+import javax.jcr.nodetype.ConstraintViolationException
+import javax.jcr.nodetype.NoSuchNodeTypeException
+import javax.jcr.version.VersionException
 
 /**
  * A Custom ItemWriter that will write the provided Jcr Nodes to the {@link JcrNodesWriter#theSession()}
@@ -41,16 +51,24 @@ class JcrNodesWriter implements ItemWriter<ProtoNode>, ItemWriteListener {
     @Override
     void beforeWrite(List nodeProtos) {
         //no-op
+        log.trace "beforeWrite() : About to write on the client some nodeProtos"
     }
 
 
     @Override
     void afterWrite(List nodeProtos) {
         log.info "Saving ${nodeProtos.size()} nodes"
-        log.debug """Saving Nodes : ${(nodeProtos as List<ProtoNode>).collectMany { ProtoNode pNode ->
-            [ pNode.name , pNode.mandatoryChildNodeList.collect { it.name - pNode.name }]
-        }.flatten()}"""
-        theSession().save()
+        if (log.isDebugEnabled()) {
+            log.debug """Saving Nodes : ${(nodeProtos as List<ProtoNode>).collectMany { ProtoNode pNode ->
+            [ pNode.name , pNode.mandatoryChildNodeList.collect { it.name - pNode.name }]}.flatten()}"""
+        }
+        try {
+            theSession().save()
+//        } catch(InvalidItemStateException|ConstraintViolationException|AccessDeniedException|ItemExistsException|ReferentialIntegrityException|VersionException|LockException|NoSuchNodeTypeException|RepositoryException e){
+        } catch (Exception e) {
+            log.error("Exception occurred when trying to save nodes on the client\n${e}", e)
+        }
+
         withStopWatch("Refreshing session: ${theSession()}") {
             theSession().refresh(false)
         }
@@ -60,6 +78,11 @@ class JcrNodesWriter implements ItemWriter<ProtoNode>, ItemWriteListener {
     @Override
     void onWriteError(Exception exception, List nodeProtos) {
         log.error "Exception writing JCR Nodes to current JCR Session : ${theSession()}. ", exception
+        StringBuilder sb = new StringBuilder();
+        for (Object nodeProto : nodeProtos) {
+            sb.append(((ProtoNode)nodeProto).name).append("\n=================\n");
+        }
+        log.warn("Items where the error occurred are: \n" + sb.toString());
     }
 
 
@@ -69,8 +92,10 @@ class JcrNodesWriter implements ItemWriter<ProtoNode>, ItemWriteListener {
      */
     @Override
     void write(List<? extends ProtoNode> nodeProtos) throws Exception {
+        log.trace "client write() : START"
         Session session = theSession()
         for (ProtoNode nodeProto : nodeProtos) {
+            log.debug "writeToJcr : nodeProto=${nodeProto.name}"
             writeToJcr(nodeProto, session)
         }
     }

src/main/groovy/com/twcable/grabbit/jcr/JCRNodeDecorator.groovy

@@ -18,25 +18,24 @@ package com.twcable.grabbit.jcr
 import com.twcable.grabbit.proto.NodeProtos.Node as ProtoNode
 import com.twcable.grabbit.proto.NodeProtos.Node.Builder as ProtoNodeBuilder
 import com.twcable.grabbit.proto.NodeProtos.Property as ProtoProperty
+import com.twcable.grabbit.server.services.impl.TreeTraverser
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
 import javax.annotation.Nonnull
 import javax.annotation.Nullable
 import javax.jcr.ItemNotFoundException
+import javax.jcr.Node
 import javax.jcr.Node as JCRNode
 import javax.jcr.PathNotFoundException
 import javax.jcr.Property as JcrProperty
 import javax.jcr.RepositoryException
 import javax.jcr.nodetype.ItemDefinition
-import org.apache.jackrabbit.commons.flat.TreeTraverser
 import org.apache.jackrabbit.value.DateValue
 
 
 import static org.apache.jackrabbit.JcrConstants.JCR_CREATED
 import static org.apache.jackrabbit.JcrConstants.JCR_LASTMODIFIED
 import static org.apache.jackrabbit.JcrConstants.JCR_PRIMARYTYPE
-import static org.apache.jackrabbit.commons.flat.TreeTraverser.ErrorHandler
-import static org.apache.jackrabbit.commons.flat.TreeTraverser.InclusionPolicy
 import static org.apache.jackrabbit.oak.spi.security.authorization.accesscontrol.AccessControlConstants.AC_NODETYPE_NAMES
 import static org.apache.jackrabbit.oak.spi.security.authorization.accesscontrol.AccessControlConstants.NT_REP_ACL
 
@@ -84,7 +83,7 @@ class JCRNodeDecorator {
 
 
     Iterator<JCRNode> getChildNodeIterator() {
-        return TreeTraverser.nodeIterator(innerNode, ErrorHandler.IGNORE, new NoRootInclusionPolicy(this))
+        return TreeTraverser.nodeIterator(innerNode, TreeTraverser.ErrorHandler.ALL, new NoRootInclusionPolicy(this) as TreeTraverser.InclusionPolicy<? super Node>)
     }
 
 
@@ -121,6 +120,8 @@ class JCRNodeDecorator {
 
 
     String getPrimaryType() {
+        if (innerNode == null || innerNode.getProperty(JCR_PRIMARYTYPE) == null)
+            return ""
         innerNode.getProperty(JCR_PRIMARYTYPE).string
     }
 
@@ -158,7 +159,9 @@ class JCRNodeDecorator {
      ProtoNode toProtoNode() {
         final ProtoNodeBuilder protoNodeBuilder = ProtoNode.newBuilder()
         protoNodeBuilder.setName(path)
+        log.trace "toProtoNode() : about to collect all properties"
         protoNodeBuilder.addAllProperties(getProtoProperties())
+        log.trace "toProtoNode() : collected all properties"
         requiredChildNodes.each {
             protoNodeBuilder.addMandatoryChildNode(it.toProtoNode())
         }
@@ -201,6 +204,7 @@ class JCRNodeDecorator {
      */
     boolean isAuthorizablePart() {
         try {
+            if (getParent() == null) return false
             JCRNodeDecorator parent = new JCRNodeDecorator(getParent())
             while(!parent.isAuthorizableType()) {
                 parent = new JCRNodeDecorator(parent.getParent())
@@ -268,7 +272,7 @@ class JCRNodeDecorator {
     }
 
     @CompileStatic
-    private static class NoRootInclusionPolicy implements InclusionPolicy<JCRNode> {
+    private static class NoRootInclusionPolicy implements TreeTraverser.InclusionPolicy<JCRNode> {
 
         final JCRNodeDecorator rootNode