Xinwen Deng / aem-src

/*
 * Decompiled with CFR 0_118.
 * 
 * Could not load the following classes:
 *  javax.jcr.Node
 *  javax.jcr.Property
 *  javax.jcr.RepositoryException
 *  javax.jcr.Session
 *  org.apache.felix.scr.annotations.Activate
 *  org.apache.felix.scr.annotations.Component
 *  org.apache.felix.scr.annotations.Deactivate
 *  org.apache.felix.scr.annotations.Property
 *  org.apache.felix.scr.annotations.Reference
 *  org.apache.felix.scr.annotations.ReferenceCardinality
 *  org.apache.felix.scr.annotations.ReferencePolicy
 *  org.apache.felix.scr.annotations.Service
 *  org.apache.jackrabbit.commons.JcrUtils
 *  org.apache.sling.api.resource.ResourceResolverFactory
 *  org.apache.sling.commons.scheduler.Scheduler
 *  org.apache.sling.event.jobs.Job
 *  org.apache.sling.event.jobs.JobManager
 *  org.apache.sling.event.jobs.consumer.JobExecutionContext
 *  org.apache.sling.event.jobs.consumer.JobExecutionContext$ResultBuilder
 *  org.apache.sling.event.jobs.consumer.JobExecutionResult
 *  org.apache.sling.event.jobs.consumer.JobExecutor
 *  org.apache.sling.jcr.api.SlingRepository
 *  org.apache.sling.serviceusermapping.ServiceUserMapped
 *  org.osgi.framework.BundleContext
 *  org.osgi.service.component.ComponentContext
 *  org.slf4j.Logger
 *  org.slf4j.LoggerFactory
 */
package com.day.cq.replication.impl;

import com.day.cq.replication.Agent;
import com.day.cq.replication.AgentConfig;
import com.day.cq.replication.AgentConfigGroup;
import com.day.cq.replication.AgentManager;
import com.day.cq.replication.ConfigManager;
import com.day.cq.replication.ReplicationAction;
import com.day.cq.replication.ReplicationActionType;
import com.day.cq.replication.ReplicationContentFactory;
import com.day.cq.replication.ReplicationContentFilter;
import com.day.cq.replication.ReplicationContentFilterFactory;
import com.day.cq.replication.ReplicationQueue;
import com.day.cq.replication.impl.AgentImpl;
import com.day.cq.replication.impl.ReplicationContentFactoryProvider;
import com.day.cq.replication.impl.ServiceTracker;
import com.day.cq.replication.impl.distribution.DistributionQueueImpl;
import com.day.cq.replication.impl.queue.QueueListener;
import com.day.cq.replication.impl.queue.ReplicationJob;
import com.day.cq.replication.impl.queue.ReplicationQueueImpl;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.jcr.Node;
import javax.jcr.Property;
import javax.jcr.RepositoryException;
import javax.jcr.Session;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.ReferencePolicy;
import org.apache.felix.scr.annotations.Service;
import org.apache.jackrabbit.commons.JcrUtils;
import org.apache.sling.api.resource.ResourceResolverFactory;
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.consumer.JobExecutionContext;
import org.apache.sling.event.jobs.consumer.JobExecutionResult;
import org.apache.sling.event.jobs.consumer.JobExecutor;
import org.apache.sling.jcr.api.SlingRepository;
import org.apache.sling.serviceusermapping.ServiceUserMapped;
import org.osgi.framework.BundleContext;
import org.osgi.service.component.ComponentContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Component(metatype=1)
@Service(value={AgentManager.class, JobExecutor.class})
@org.apache.felix.scr.annotations.Property(name="job.topics", value={"com/day/cq/replication/job/*"})
@Reference(name="contentFilterFactory", referenceInterface=ReplicationContentFilterFactory.class, cardinality=ReferenceCardinality.OPTIONAL_MULTIPLE, policy=ReferencePolicy.DYNAMIC)
public class AgentManagerImpl
implements AgentManager,
JobExecutor,
ConfigManager.ConfigEventListener {
    private static final Logger log = LoggerFactory.getLogger(AgentManagerImpl.class);
    private static final String CONTENT_PATH = "/var/replication/data";
    private ComponentContext componentContext;
    @Reference
    private JobManager jobManager = null;
    @Reference
    private Scheduler scheduler = null;
    @Reference
    private SlingRepository repository = null;
    @Reference
    private ServiceTracker serviceTracker = null;
    @Reference
    private ReplicationContentFactoryProvider factoryProvider = null;
    @Reference
    private ConfigManager configManager = null;
    @org.apache.felix.scr.annotations.Property(name="serviceUser.target", value={"(subServiceName=replicationService)"})
    @Reference(name="serviceUser")
    private ServiceUserMapped serviceUserMapped;
    @Reference
    private ResourceResolverFactory resourceResolverFactory = null;
    private final Map<String, ReplicationContentFilterFactory> filterFactoryCache = new ConcurrentHashMap<String, ReplicationContentFilterFactory>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private Map<String, AgentImpl> agents = new LinkedHashMap<String, AgentImpl>();
    private Map<String, Agent> cachedAgents;

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    @Activate
    protected void activate(ComponentContext ctx) throws RepositoryException {
        this.componentContext = ctx;
        this.lock.writeLock().lock();
        Session session = null;
        try {
            for (String cfgId : this.configManager.getConfigurations().keySet()) {
                try {
                    this.update(cfgId);
                }
                catch (RepositoryException e) {
                    log.error("Error while initializing replication agent.", (Throwable)e);
                }
            }
            this.cacheAgents();
            this.configManager.registerListener(this);
            session = this.repository.loginService("replicationService", null);
            JcrUtils.getOrCreateByPath((String)"/var/replication/data", (String)"sling:Folder", (String)"sling:Folder", (Session)session, (boolean)false);
            session.save();
            log.info("AgentManager service activated");
        }
        finally {
            if (session != null) {
                session.logout();
            }
            this.lock.writeLock().unlock();
        }
    }

    @Deactivate
    protected void deactivate() {
        this.componentContext = null;
        this.configManager.unregisterListener(this);
        this.lock.writeLock().lock();
        try {
            for (AgentImpl agent : this.agents.values()) {
                agent.destroy(false);
            }
            this.agents.clear();
            this.cachedAgents = Collections.emptyMap();
        }
        finally {
            this.lock.writeLock().unlock();
        }
        this.filterFactoryCache.clear();
    }

    @Override
    public Map<String, Agent> getAgents() {
        return this.cachedAgents;
    }

    private void cacheAgents() {
        this.cachedAgents = new LinkedHashMap<String, Agent>();
        for (Agent agent : this.agents.values()) {
            this.cachedAgents.put(agent.getId(), agent);
        }
        this.cachedAgents = Collections.unmodifiableMap(this.cachedAgents);
    }

    @Override
    public void onConfigEvent(ConfigManager.ConfigEvent evt) {
        this.lock.writeLock().lock();
        try {
            switch (evt.getType()) {
                case ADDED: 
                case UPDATED: {
                    this.update(evt.getId());
                    break;
                }
                case REMOVED: {
                    this.remove(evt.getId());
                }
            }
            this.cacheAgents();
        }
        catch (RepositoryException e) {
            log.error("Error while adding/updating agent: {}", (Object)evt.getId(), (Object)e);
        }
        finally {
            this.lock.writeLock().unlock();
        }
    }

    private void update(String cfgId) throws RepositoryException {
        AgentConfig config = this.configManager.getConfigurations().get(cfgId);
        if (config == null || !config.getConfigGroup().isActive()) {
            this.remove(cfgId);
        } else {
            AgentImpl agent = this.agents.remove(cfgId);
            if (agent == null) {
                String id = config.getAgentId();
                log.info("AgentConfig at {} added.", (Object)cfgId);
                ReplicationContentFactory factory = this.factoryProvider.create(id);
                agent = new AgentImpl(id, config, this.serviceTracker, this, factory);
                agent.init(this.componentContext.getBundleContext());
            } else {
                log.info("AgentConfig at {} updated.", (Object)cfgId);
                agent.update(config);
            }
            this.agents.put(cfgId, agent);
        }
    }

    private void remove(String path) {
        AgentImpl agent = this.agents.remove(path);
        if (agent != null) {
            log.info("AgentConfig at {} removed.", (Object)path);
            agent.destroy();
        }
    }

    private Node getOutboxNode(Session session, String agentId) throws RepositoryException {
        String path = "/var/replication/data/" + agentId;
        return JcrUtils.getOrCreateByPath((String)path, (String)"sling:Folder", (String)"sling:Folder", (Session)session, (boolean)true);
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    protected Calendar getLastPollTime(String agentId) {
        this.lock.writeLock().lock();
        try {
            Session session = null;
            try {
                session = this.repository.loginService("replicationService", null);
                Node ob = this.getOutboxNode(session, agentId);
                if (ob.hasProperty("cq:lastPoll")) {
                    Calendar calendar = ob.getProperty("cq:lastPoll").getDate();
                    return calendar;
                }
            }
            catch (RepositoryException e) {
                log.error("Error while fetching last poll time for agent {}", (Object)agentId, (Object)e);
            }
            finally {
                if (session != null) {
                    session.logout();
                }
            }
            Calendar e = null;
            return e;
        }
        finally {
            this.lock.writeLock().unlock();
        }
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    protected void setLastPollTime(String agentId, Calendar time) {
        this.lock.writeLock().lock();
        Session session = null;
        try {
            session = this.repository.loginService("replicationService", null);
            Node ob = this.getOutboxNode(session, agentId);
            ob.setProperty("cq:lastPoll", time);
            session.save();
        }
        catch (RepositoryException e) {
            log.error("Error while fetching last poll time for agent {}", (Object)agentId, (Object)e);
        }
        finally {
            this.lock.writeLock().unlock();
            if (session != null) {
                session.logout();
            }
        }
    }

    public JobExecutionResult process(Job newJob, JobExecutionContext context) {
        ReplicationJob job = new ReplicationJob(newJob, context);
        String queueName = job.getQueueName();
        Agent a = this.getAgents().get(queueName);
        if (a == null) {
            log.error("Job contains unknown agent: {}. Discarding Job {}", (Object)queueName, (Object)job);
            return context.result().cancelled();
        }
        ReplicationQueueImpl queue = (ReplicationQueueImpl)a.getQueue();
        if (queue == null) {
            log.error("Job contains agent which is not started: {}. waiting {}", (Object)queueName, (Object)job);
            return context.result().failed();
        }
        QueueListener.JobStatus result = a.getConfiguration().isBatchMode() && ReplicationActionType.ACTIVATE.equals((Object)job.getAction().getType()) ? queue.batchProcess(job, a.getConfiguration()) : queue.process(job);
        if (result == null) {
            return null;
        }
        if (result == QueueListener.JobStatus.CANCELLED) {
            return context.result().cancelled();
        }
        if (result == QueueListener.JobStatus.FAILED) {
            return context.result().failed();
        }
        return context.result().succeeded();
    }

    public ReplicationQueueImpl createQueue(String name, QueueListener listener) {
        ReplicationQueueImpl q = new ReplicationQueueImpl(this.jobManager, this.scheduler, name);
        q.open(listener);
        return q;
    }

    public DistributionQueueImpl createDistributionQueue(String distributionAgentName) {
        return new DistributionQueueImpl(this.resourceResolverFactory, distributionAgentName);
    }

    protected void bindContentFilterFactory(ReplicationContentFilterFactory factory, Map<String, Object> properties) {
        String key = (String)properties.get("service.pid");
        this.filterFactoryCache.put(key, factory);
        log.info("bound new filter factory [{}]. got [{}] factories.", (Object)key, (Object)this.filterFactoryCache.size());
    }

    protected void unbindContentFilterFactory(ReplicationContentFilterFactory factory, Map<String, Object> properties) {
        String key = (String)properties.get("service.pid");
        this.filterFactoryCache.remove(key);
        log.info("unbound filter factory [{}]. got [{}] factories.", (Object)key, (Object)this.filterFactoryCache.size());
    }

    public List<ReplicationContentFilter> createContentFilterChain(ReplicationAction action) {
        ArrayList<ReplicationContentFilter> filters = new ArrayList<ReplicationContentFilter>();
        Collection<ReplicationContentFilterFactory> factories = this.filterFactoryCache.values();
        for (ReplicationContentFilterFactory factory : factories) {
            ReplicationContentFilter filter = factory.createFilter(action);
            if (null == filter) continue;
            filters.add(filter);
        }
        return filters;
    }

    protected void bindJobManager(JobManager jobManager) {
        this.jobManager = jobManager;
    }

    protected void unbindJobManager(JobManager jobManager) {
        if (this.jobManager == jobManager) {
            this.jobManager = null;
        }
    }

    protected void bindScheduler(Scheduler scheduler) {
        this.scheduler = scheduler;
    }

    protected void unbindScheduler(Scheduler scheduler) {
        if (this.scheduler == scheduler) {
            this.scheduler = null;
        }
    }

    protected void bindRepository(SlingRepository slingRepository) {
        this.repository = slingRepository;
    }

    protected void unbindRepository(SlingRepository slingRepository) {
        if (this.repository == slingRepository) {
            this.repository = null;
        }
    }

    protected void bindServiceTracker(ServiceTracker serviceTracker) {
        this.serviceTracker = serviceTracker;
    }

    protected void unbindServiceTracker(ServiceTracker serviceTracker) {
        if (this.serviceTracker == serviceTracker) {
            this.serviceTracker = null;
        }
    }

    protected void bindFactoryProvider(ReplicationContentFactoryProvider replicationContentFactoryProvider) {
        this.factoryProvider = replicationContentFactoryProvider;
    }

    protected void unbindFactoryProvider(ReplicationContentFactoryProvider replicationContentFactoryProvider) {
        if (this.factoryProvider == replicationContentFactoryProvider) {
            this.factoryProvider = null;
        }
    }

    protected void bindConfigManager(ConfigManager configManager) {
        this.configManager = configManager;
    }

    protected void unbindConfigManager(ConfigManager configManager) {
        if (this.configManager == configManager) {
            this.configManager = null;
        }
    }

    protected void bindServiceUser(ServiceUserMapped serviceUserMapped) {
        this.serviceUserMapped = serviceUserMapped;
    }

    protected void unbindServiceUser(ServiceUserMapped serviceUserMapped) {
        if (this.serviceUserMapped == serviceUserMapped) {
            this.serviceUserMapped = null;
        }
    }

    protected void bindResourceResolverFactory(ResourceResolverFactory resourceResolverFactory) {
        this.resourceResolverFactory = resourceResolverFactory;
    }

    protected void unbindResourceResolverFactory(ResourceResolverFactory resourceResolverFactory) {
        if (this.resourceResolverFactory == resourceResolverFactory) {
            this.resourceResolverFactory = null;
        }
    }

}