Xinwen Deng / aem-src

/*
 * Decompiled with CFR 0_118.
 * 
 * Could not load the following classes:
 *  javax.jcr.RepositoryException
 *  org.apache.sling.commons.scheduler.ScheduleOptions
 *  org.apache.sling.commons.scheduler.Scheduler
 *  org.apache.sling.event.jobs.Job
 *  org.apache.sling.event.jobs.JobManager
 *  org.apache.sling.event.jobs.JobManager$QueryType
 *  org.apache.sling.event.jobs.Queue
 *  org.apache.sling.event.jobs.Statistics
 *  org.apache.sling.event.jobs.consumer.JobExecutionContext
 *  org.apache.sling.event.jobs.consumer.JobExecutionContext$ResultBuilder
 *  org.apache.sling.event.jobs.consumer.JobExecutionResult
 *  org.slf4j.Logger
 *  org.slf4j.LoggerFactory
 */
package com.day.cq.replication.impl.queue;

import com.day.cq.replication.Agent;
import com.day.cq.replication.AgentConfig;
import com.day.cq.replication.ReplicationAction;
import com.day.cq.replication.ReplicationActionType;
import com.day.cq.replication.ReplicationContentFacade;
import com.day.cq.replication.ReplicationException;
import com.day.cq.replication.ReplicationQueue;
import com.day.cq.replication.impl.queue.QueueListener;
import com.day.cq.replication.impl.queue.ReplicationJob;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.jcr.RepositoryException;
import org.apache.sling.commons.scheduler.ScheduleOptions;
import org.apache.sling.commons.scheduler.Scheduler;
import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.JobManager;
import org.apache.sling.event.jobs.Queue;
import org.apache.sling.event.jobs.Statistics;
import org.apache.sling.event.jobs.consumer.JobExecutionContext;
import org.apache.sling.event.jobs.consumer.JobExecutionResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ReplicationQueueImpl
implements ReplicationQueue {
    private final Logger log;
    private final JobManager jobManager;
    private final String name;
    private QueueListener agent;
    private final String queueTopic;
    private volatile long processingSince = 0;
    private Scheduler scheduler;
    private final BatchInfo batchInfo = new BatchInfo();

    public ReplicationQueueImpl(JobManager mgr, Scheduler scheduler, String name) {
        this.scheduler = scheduler;
        this.name = name;
        this.log = LoggerFactory.getLogger((String)(Agent.class.getName() + "." + name + ".queue"));
        this.jobManager = mgr;
        this.queueTopic = "com/day/cq/replication/job/" + name;
    }

    private Queue getJobQueue() {
        Queue q = this.jobManager.getQueue(this.queueTopic);
        if (q == null) {
            this.log.debug("Queue {} not found.", (Object)this.queueTopic);
        }
        return q;
    }

    public void open(QueueListener agent) {
        this.agent = agent;
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    public void close(boolean clear) {
        List<ReplicationJob> processList;
        BatchInfo batchInfo = this.batchInfo;
        synchronized (batchInfo) {
            processList = this.batchInfo.drain();
            this.batchInfo.isRunning = false;
            this.batchInfo.notify();
        }
        boolean first = true;
        for (ReplicationJob job : processList) {
            JobExecutionContext context = (JobExecutionContext)job.getProperty(":sling:jobs:asynchandler");
            try {
                if (first) {
                    first = false;
                    context.asyncProcessingFinished(context.result().failed());
                    continue;
                }
                context.asyncProcessingFinished(context.result().failed(0));
            }
            catch (IllegalStateException var7_9) {}
        }
        if (clear) {
            this.clear();
        }
        this.agent = null;
    }

    @Override
    public String getName() {
        return this.name;
    }

    @Override
    public boolean isPaused() {
        Queue q = this.getJobQueue();
        return q != null && q.isSuspended();
    }

    @Override
    public void setPaused(boolean paused) {
        Queue q = this.getJobQueue();
        if (q != null) {
            if (paused) {
                this.getJobQueue().suspend();
            } else {
                this.getJobQueue().resume();
            }
        }
    }

    @Override
    public List<ReplicationQueue.Entry> entries() {
        LinkedList<ReplicationQueue.Entry> entries = new LinkedList<ReplicationQueue.Entry>();
        int i = 0;
        for (Job evt : this.jobManager.findJobs(JobManager.QueryType.ALL, this.queueTopic, -1, new Map[0])) {
            ReplicationJob job = new ReplicationJob(evt);
            entries.add(new EntryImpl(job.getEventId(), job, i++));
        }
        return entries;
    }

    @Override
    public List<ReplicationQueue.Entry> entries(String path) {
        LinkedList<ReplicationQueue.Entry> entries = new LinkedList<ReplicationQueue.Entry>();
        int i = 0;
        for (Job evt : this.jobManager.findJobs(JobManager.QueryType.ALL, this.queueTopic, -1, new Map[0])) {
            ReplicationJob job = new ReplicationJob(evt);
            if (path.equals(job.getPath())) {
                entries.add(new EntryImpl(job.getEventId(), job, i));
            }
            ++i;
        }
        return entries;
    }

    public boolean hasEntries(String path) {
        HashMap<String, String> pathTemplate = new HashMap<String, String>();
        pathTemplate.put("cq:path", path);
        return !this.jobManager.findJobs(JobManager.QueryType.ALL, this.queueTopic, 1, new Map[]{pathTemplate}).isEmpty();
    }

    @Override
    public ReplicationQueue.Entry getEntry(String path, Calendar after) {
        long time = after == null ? 0 : after.getTimeInMillis();
        for (ReplicationQueue.Entry e : this.entries(path)) {
            if (e.getAction().getTime() < time) continue;
            return e;
        }
        return null;
    }

    @Override
    public void clear() {
        LinkedHashMap<String, EntryData> entries = new LinkedHashMap<String, EntryData>();
        for (Job evt : this.jobManager.findJobs(JobManager.QueryType.ALL, this.queueTopic, -1, new Map[0])) {
            ReplicationJob job = new ReplicationJob(evt);
            entries.put(job.getEventId(), new EntryData(job.getAction(), job.getContent()));
        }
        this.clear(entries);
    }

    @Override
    public void clear(Set<String> ids) {
        LinkedHashMap<String, EntryData> entries = new LinkedHashMap<String, EntryData>();
        for (String id : ids) {
            Job evt = this.jobManager.getJobById(id);
            if (evt == null) continue;
            ReplicationJob job = new ReplicationJob(evt);
            entries.put(job.getEventId(), new EntryData(job.getAction(), job.getContent()));
        }
        this.clear(entries);
    }

    private void clear(Map<String, EntryData> entries) {
        for (Map.Entry<String, EntryData> entry : entries.entrySet()) {
            if (!this.jobManager.removeJobById(entry.getKey())) continue;
            this.agent.abort(entry.getValue());
        }
    }

    @Override
    public boolean isBlocked() {
        return this.getStatus().getNextRetryTime() > 0;
    }

    @Override
    public long nextRetryPeriod() {
        ReplicationQueue.Status st = this.getStatus();
        return st.getNextRetryTime() - st.getStatusTime();
    }

    @Override
    public long lastProcessTime() {
        return this.getStatus().getLastProcessTime();
    }

    @Override
    public ReplicationQueue.Status getStatus() {
        Long nextRetryTime;
        Queue q = this.getJobQueue();
        if (q == null) {
            return new StatusImpl(System.currentTimeMillis(), -1, -1, -1);
        }
        return new StatusImpl(System.currentTimeMillis(), this.processingSince, q.getStatistics().getLastFinishedJobTime(), (nextRetryTime = (Long)q.getState("isSleepingUntil")) != null ? nextRetryTime : -1);
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    @Override
    public void forceRetry() {
        ReplicationQueueImpl replicationQueueImpl = this;
        synchronized (replicationQueueImpl) {
            List<Job> jobs = this.getJobs(0, 1);
            if (!jobs.isEmpty()) {
                this.process(new ReplicationJob(jobs.get(0)));
            }
        }
    }

    private List<Job> getJobs(int skip, int limit) {
        int actualSkip = skip < 0 ? 0 : skip;
        int actualLimit = limit < 0 ? -1 : actualSkip + limit;
        Collection jobs = this.jobManager.findJobs(JobManager.QueryType.ALL, this.queueTopic, (long)actualLimit, new Map[0]);
        ArrayList<Job> result = new ArrayList<Job>();
        int i = 0;
        for (Job job : jobs) {
            if (i >= actualSkip) {
                result.add(job);
            }
            ++i;
        }
        return result;
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    public QueueListener.JobStatus process(ReplicationJob job) {
        this.log.info("Processing job for agent {}", (Object)job.getQueueName());
        try {
            long start = System.currentTimeMillis();
            EntryData data = new EntryData(job.getAction(), job.getContent());
            boolean ret = false;
            try {
                this.processingSince = System.currentTimeMillis();
                ret = this.agent.process(data);
            }
            finally {
                this.processingSince = 0;
            }
            long end = System.currentTimeMillis();
            Object[] arrobject = new Object[3];
            arrobject[0] = job.getQueueName();
            arrobject[1] = String.valueOf(end - start);
            arrobject[2] = ret ? "Ok." : "Failed.";
            this.log.info("Job for agent {} processed in {}ms. {}", arrobject);
            return ret ? QueueListener.JobStatus.SUCCEEDED : QueueListener.JobStatus.FAILED;
        }
        catch (Throwable e) {
            this.log.error("Error during processing of replication.", e);
            return QueueListener.JobStatus.FAILED;
        }
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    private List<QueueListener.JobStatus> process(List<ReplicationJob> jobs) {
        ArrayList<EntryData> entries = new ArrayList<EntryData>();
        for (ReplicationJob rj : jobs) {
            entries.add(new EntryData(rj.getAction(), rj.getContent()));
        }
        try {
            this.processingSince = System.currentTimeMillis();
            List<QueueListener.JobStatus> list = this.agent.process(entries);
            return list;
        }
        finally {
            this.processingSince = 0;
        }
    }

    public void addEntry(ReplicationAction action, ReplicationContentFacade content, long retryDelay) throws Exception {
        HashMap<String, Object> properties = new HashMap<String, Object>();
        properties.put("event.job.retrydelay", retryDelay);
        if (content != null) {
            properties.put("replicationContent", content);
            this.log.debug("Creating replication job (action: {}) with content {} ", new Object[]{action, content});
        } else {
            this.log.debug("Creating replication job (action: {}) without content", new Object[]{action});
        }
        properties.put("cq:path", action.getPath());
        if (action.getPaths().length > 1) {
            properties.put("cq:paths", action.getPaths());
        }
        if (action.getRevision() != null) {
            properties.put("cq:revision", action.getRevision());
        }
        properties.put("cq:time", action.getTime());
        properties.put("cq:type", action.getType().name());
        properties.put("cq:user", action.getUserId());
        Job job = this.jobManager.addJob("com/day/cq/replication/job/" + this.name, properties);
        if (job == null) {
            throw new ReplicationException("Replication job could not be created by JobManager");
        }
        this.log.debug("replication job {} added", (Object)job);
        if (content != null) {
            this.verify(job);
        }
        this.log.info("Replication request queued for {} at {}.", (Object)this.name, (Object)action.getPath());
    }

    private void verify(Job job) throws Exception {
        block4 : {
            Job retrievedJob = this.jobManager.getJobById(job.getId());
            if (retrievedJob == null) {
                return;
            }
            Object contentProperty = retrievedJob.getProperty("replicationContent");
            try {
                ReplicationContentFacade replicationContentFacade = (ReplicationContentFacade)contentProperty;
                if (replicationContentFacade != null) {
                    this.log.debug("replication job content persisted correctly");
                    break block4;
                }
                throw new Exception("replication job content persisted as null");
            }
            catch (ClassCastException cce) {
                throw new Exception("content property was not stored correctly");
            }
        }
    }

    private boolean sendBatch(AgentConfig config) {
        if (this.batchInfo.size >= config.getBatchMaxSize()) {
            return true;
        }
        return false;
    }

    private void checkBatchTimer(final AgentConfig config) {
        if (this.batchInfo.queuedJobs.size() > 0 && !this.batchInfo.isRunning) {
            this.batchInfo.isRunning = true;
            this.log.info("Starting batch timer with {}", (Object)config.getBatchWaitTime());
            Runnable r = new Runnable(){

                /*
                 * WARNING - Removed try catching itself - possible behaviour change.
                 */
                @Override
                public void run() {
                    List<ReplicationJob> processList = null;
                    BatchInfo batchInfo = ReplicationQueueImpl.this.batchInfo;
                    synchronized (batchInfo) {
                        try {
                            ReplicationQueueImpl.this.batchInfo.wait(1000 * config.getBatchWaitTime());
                        }
                        catch (InterruptedException var3_3) {
                            // empty catch block
                        }
                        if (ReplicationQueueImpl.access$200((ReplicationQueueImpl)ReplicationQueueImpl.this).isRunning) {
                            ReplicationQueueImpl.access$200((ReplicationQueueImpl)ReplicationQueueImpl.this).isRunning = false;
                            if (ReplicationQueueImpl.access$200((ReplicationQueueImpl)ReplicationQueueImpl.this).queuedJobs.size() > 0) {
                                processList = ReplicationQueueImpl.this.batchInfo.drain();
                            }
                        }
                    }
                    if (processList != null) {
                        ReplicationQueueImpl.this.startBatchReplication(processList, null);
                    }
                }
            };
            this.scheduler.schedule((Object)r, this.scheduler.NOW());
        }
    }

    private QueueListener.JobStatus startBatchReplication(List<ReplicationJob> processList, ReplicationJob lastJob) {
        long start = System.currentTimeMillis();
        List<QueueListener.JobStatus> result = this.process(processList);
        long end = System.currentTimeMillis();
        Iterator<QueueListener.JobStatus> iter = result.iterator();
        boolean first = true;
        for (ReplicationJob rj : processList) {
            QueueListener.JobStatus jobResult = iter.next();
            this.log.info("Job for agent {} processed in {}ms: {}", new Object[]{rj.getQueueName(), String.valueOf(end - start), jobResult.name()});
            JobExecutionContext jobContext = (JobExecutionContext)rj.getProperty(":sling:jobs:asynchandler");
            if (rj == lastJob) {
                return jobResult;
            }
            try {
                if (first || jobResult != QueueListener.JobStatus.FAILED) {
                    first = false;
                    if (jobResult == QueueListener.JobStatus.CANCELLED) {
                        jobContext.asyncProcessingFinished(jobContext.result().cancelled());
                        continue;
                    }
                    if (jobResult == QueueListener.JobStatus.FAILED) {
                        jobContext.asyncProcessingFinished(jobContext.result().failed());
                        continue;
                    }
                    if (jobResult == QueueListener.JobStatus.SUCCEEDED) {
                        jobContext.asyncProcessingFinished(jobContext.result().succeeded());
                        continue;
                    }
                    this.log.warn("async processing status cannot be determined for {}", (Object)jobResult.name());
                    continue;
                }
                jobContext.asyncProcessingFinished(jobContext.result().failed(0));
            }
            catch (IllegalStateException ise) {
                this.log.error("Error during replication of job " + rj, (Throwable)ise);
            }
        }
        return null;
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    public QueueListener.JobStatus batchProcess(ReplicationJob job, AgentConfig config) {
        QueueListener.JobStatus result;
        List<ReplicationJob> processList = null;
        boolean doSingleReplication = false;
        BatchInfo batchInfo = this.batchInfo;
        synchronized (batchInfo) {
            if (job.getContent() != null && job.getContent().getContentLength() == -1) {
                this.log.info("Job {} for agent {} has no content length - batch processing not possible", (Object)job.getAction(), (Object)job.getQueueName());
                doSingleReplication = true;
            } else if (job.getAction().getType() != ReplicationActionType.ACTIVATE) {
                this.log.info("Job {} for agent {} is neither activate, delete nor deactivate - batch processing not possible", (Object)job.getAction(), (Object)job.getQueueName());
                doSingleReplication = true;
            } else {
                this.batchInfo.add(job);
                this.log.info("Queued job {} for agent {} for batch processing...", (Object)job.getAction(), (Object)job.getQueueName());
            }
            if (doSingleReplication || this.sendBatch(config)) {
                processList = this.batchInfo.drain();
                this.batchInfo.isRunning = false;
                this.batchInfo.notify();
            }
            this.checkBatchTimer(config);
        }
        if (processList != null && (result = this.startBatchReplication(processList, job)) != null) {
            return result;
        }
        if (doSingleReplication) {
            return this.process(job);
        }
        return null;
    }

    public static final class StatusImpl
    implements ReplicationQueue.Status {
        private final long statusTime;
        private final long processingSince;
        private final long lastProcessTime;
        private final long nextRetryTime;

        public StatusImpl(long statusTime, long processingSince, long lastProcessTime, long nextRetryTime) {
            this.statusTime = statusTime;
            this.processingSince = processingSince;
            this.lastProcessTime = lastProcessTime;
            this.nextRetryTime = nextRetryTime;
        }

        @Override
        public long getStatusTime() {
            return this.statusTime;
        }

        @Override
        public long getProcessingSince() {
            return this.processingSince;
        }

        @Override
        public long getLastProcessTime() {
            return this.lastProcessTime;
        }

        @Override
        public long getNextRetryTime() {
            return this.nextRetryTime;
        }
    }

    private final class EntryImpl
    implements ReplicationQueue.Entry {
        private final String id;
        private final ReplicationJob job;
        private EntryData data;
        private final int queuePos;

        private EntryImpl(String id, ReplicationJob job, int queuePos) {
            this.id = id;
            this.job = job;
            this.queuePos = queuePos;
        }

        @Override
        public String getId() {
            return this.id;
        }

        @Override
        public ReplicationQueue getQueue() {
            return ReplicationQueueImpl.this;
        }

        @Override
        public ReplicationAction getAction() {
            try {
                EntryData data = this.getData();
                return data == null ? null : data.getAction();
            }
            catch (RepositoryException e) {
                ReplicationQueueImpl.this.log.error("Error while reading data: {}", (Object)e.toString());
                return null;
            }
        }

        @Override
        public ReplicationContentFacade getContent() {
            try {
                EntryData data = this.getData();
                return data == null ? null : data.getContent();
            }
            catch (RepositoryException e) {
                ReplicationQueueImpl.this.log.error("Error while reading data: {}", (Object)e.toString());
                return null;
            }
        }

        @Override
        public int getQueuePosition() {
            return this.queuePos;
        }

        @Override
        public int getNumProcessed() {
            Integer retryCount = (Integer)this.job.getProperty("event.job.retrycount");
            return retryCount == null ? 0 : retryCount;
        }

        @Override
        public Calendar getLastProcessTime() {
            Date date = (Date)this.job.getProperty("modificationDate");
            if (date != null) {
                Calendar c = Calendar.getInstance();
                c.setTime(date);
                return c;
            }
            return null;
        }

        public EntryData getData() throws RepositoryException {
            if (this.data == null) {
                this.data = new EntryData(this.job.getAction(), this.job.getContent());
            }
            return this.data;
        }
    }

    public static final class EntryData
    implements Serializable {
        private static final long serialVersionUID = 6772480983519180596L;
        private final ReplicationAction action;
        private final ReplicationContentFacade content;

        public EntryData(ReplicationAction action, ReplicationContentFacade content) {
            this.action = action;
            this.content = content;
        }

        public ReplicationAction getAction() {
            return this.action;
        }

        public ReplicationContentFacade getContent() {
            return this.content;
        }
    }

    private static final class BatchInfo {
        public final List<ReplicationJob> queuedJobs = new ArrayList<ReplicationJob>();
        public volatile long size;
        public volatile boolean isRunning = false;

        private BatchInfo() {
        }

        public void add(ReplicationJob job) {
            this.queuedJobs.add(job);
            if (job.getContent() != null) {
                this.size += job.getContent().getContentLength();
            }
        }

        public List<ReplicationJob> drain() {
            ArrayList<ReplicationJob> newList = new ArrayList<ReplicationJob>();
            newList.addAll(this.queuedJobs);
            this.queuedJobs.clear();
            this.size = 0;
            return newList;
        }
    }

}