ReplicationHelper.java

ReplicationHelper.java

package com.wemblog;

import java.util.Iterator;

import javax.jcr.Node;
import javax.jcr.RepositoryException;
import javax.jcr.Session;

import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceMetadata;
import org.apache.sling.api.resource.ResourceResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.day.cq.dam.api.Asset;
import com.day.cq.replication.Agent;
import com.day.cq.replication.AgentManager;
import com.day.cq.replication.ReplicationActionType;
import com.day.cq.replication.ReplicationException;
import com.day.cq.replication.ReplicationQueue;
import com.day.cq.replication.ReplicationStatus;
import com.day.cq.replication.Replicator;
import com.day.cq.wcm.api.Page;

/**
 * Helper Method to do tree activation
 * 
 * @author Yogesh Upadhyay
 * @since 1.0.12
 * 
 */
public class ReplicationHelper {

  private Logger logger = LoggerFactory.getLogger(ReplicationHelper.class);

  private final Replicator replicator;

  private final ResourceResolver resolver;

  private final Session session;

  private boolean onlyModified=false;

  private boolean reactivate=false;

  private boolean ignoreDeactivated=false;

  private boolean dryRun;

  private long lastUpdate;

  private long tCount = 0;
  private long aCount = 0;

  private AgentManager agentManager;
  
  private ReplicationActionType replicationActionType = ReplicationActionType.ACTIVATE;

  /**
   * Default Constructor
   * 
   * @param replicator
   * @param resolver
   * @param agentManager
   */

  public ReplicationHelper(final Replicator replicator, final ResourceResolver resolver,
      final AgentManager agentManager) {
    this.replicator = replicator;
    this.resolver = resolver;
    this.session = resolver.adaptTo(Session.class);
    this.agentManager = agentManager;
  }

  public void setOnlyModified(final boolean onlyModified) {
    this.onlyModified = onlyModified;
  }

  public void setReactivate(final boolean reactivate) {
    this.reactivate = reactivate;
  }

  public void setIgnoreDeactivated(final boolean ignoreDeactivated) {
    this.ignoreDeactivated = ignoreDeactivated;
  }
  
  public void setReplicationActionType(final ReplicationActionType actionType){
    this.replicationActionType = actionType;
  }

  /**
   * Process Replication based on path
   * 
   * @param path
   */
  public void process(String path) {
    if (path == null || path.length() == 0) {
      return;
    }
    // snip off all trailing slashes
    while (path.endsWith("/")) {
      path = path.substring(0, path.length() - 1);
    }
    // reject root and 1 level paths
    if (path.lastIndexOf('/') <= 0) {
      if (path.length() == 0) {
        path = "/";
      }
      return;

    }
    Resource res = resolver.getResource(path);
    if (res == null) {
      return;
    }
    long startTime = System.currentTimeMillis();
    try {
      process(res);
      long endTime = System.currentTimeMillis();
      logger.debug("Time taken to do tree activation ", endTime - startTime);

    } catch (Exception e) {
      logger.error("Error during tree activation of " + path, e);
    }

  }

  /**
   * Get all throttled Agents
   * 
   * @return
   */
  private Agent getThrottleAgent() {
    // get the first enabled agents
    AgentManager agentMgr = this.agentManager;
    for (Agent agent : agentMgr.getAgents().values()) {
      if (agent.isEnabled()) {
        return agent;
      }
    }
    return null;
  }

  /**
   * Process replication Job based on resource
   * 
   * @param res
   * @return
   * @throws RepositoryException
   * @throws ReplicationException
   */
  private boolean process(Resource res) throws RepositoryException,
      ReplicationException {

    // we can only tree-activate hierarchy nodes
    Node node = res.adaptTo(Node.class);
    if (!node.isNodeType("nt:hierarchyNode")) {
      return false;
    }
    Page page = res.adaptTo(Page.class);
    Asset asset = res.adaptTo(Asset.class);
    long lastModified;
    if (page != null) {
      lastModified = page.getLastModified() == null ? -1 : page
          .getLastModified().getTimeInMillis();
    } else if (asset != null) {
      lastModified = asset.getLastModified() == 0 ? -1 : asset
          .getLastModified();
    } else {
      ResourceMetadata data = res.getResourceMetadata();
      lastModified = data.getModificationTime();
    }
  
    ReplicationStatus rs = res.adaptTo(ReplicationStatus.class);
    long lastPublished = 0;
    boolean isDeactivated = false;
    boolean isActivated = false;
    if (rs != null && rs.getLastPublished() != null) {
      lastPublished = rs.getLastPublished().getTimeInMillis();
      isDeactivated = rs.isDeactivated();
      isActivated = rs.isActivated();
    }
    boolean isModified = lastModified > lastPublished;
    boolean doActivate = false;

    if (!isModified && onlyModified) {
      doActivate = false;
    }else {
      try {
        replicator.checkPermission(this.session,
            this.replicationActionType, res.getPath());
        doActivate = true;
      } catch (ReplicationException e) {
        logger.error(e.getMessage());
      }
    }

    tCount++;

    Agent agent = getThrottleAgent();
    ReplicationQueue queue = agent == null ? null : agent.getQueue();
    int num = queue == null ? 0 : queue.entries().size();
    int test = 0;

    //Check for queue full or not
    while (num > 3) {
      try {
        logger.error("Queue is Full waiting for queue to be empty");
        Thread.sleep(500);
      } catch (InterruptedException e) {
        // ignore
      }
      num = queue.entries().size();
      test++;
    }

    if (doActivate) {
      if (!dryRun) {
        try {
          logger.debug("Replicating file "+res.getPath());
          replicator.replicate(session, this.replicationActionType,
              res.getPath());
        } catch (ReplicationException e) {
          logger.error("Error during tree activation of " + res.getPath(), e);
        }
      }
      aCount++;
    }

    long now = System.currentTimeMillis();
    if (now - lastUpdate > 1000L) {
      lastUpdate = now;

    }
    Iterator<Resource> iter = resolver.listChildren(res);
    while (iter.hasNext()) {
      process(iter.next());
    }
    return true;
  }

}