Xinwen Deng / aem-src

/*
 * Decompiled with CFR 0_118.
 * 
 * Could not load the following classes:
 *  javax.jcr.Session
 *  org.apache.jackrabbit.util.ISO8601
 *  org.osgi.framework.BundleContext
 *  org.osgi.framework.ServiceRegistration
 */
package com.day.cq.replication.impl;

import com.day.cq.replication.Agent;
import com.day.cq.replication.AgentConfig;
import com.day.cq.replication.ContentBuilder;
import com.day.cq.replication.ListenerLogDelegator;
import com.day.cq.replication.PollingTransportHandler;
import com.day.cq.replication.ReplicationAction;
import com.day.cq.replication.ReplicationActionType;
import com.day.cq.replication.ReplicationContent;
import com.day.cq.replication.ReplicationContentFacade;
import com.day.cq.replication.ReplicationContentFactory;
import com.day.cq.replication.ReplicationException;
import com.day.cq.replication.ReplicationListener;
import com.day.cq.replication.ReplicationLog;
import com.day.cq.replication.ReplicationOptions;
import com.day.cq.replication.ReplicationQueue;
import com.day.cq.replication.ReplicationResult;
import com.day.cq.replication.ReplicationTransaction;
import com.day.cq.replication.ReverseReplication;
import com.day.cq.replication.TransportContext;
import com.day.cq.replication.TransportException;
import com.day.cq.replication.TransportHandler;
import com.day.cq.replication.UnrecoverableReplicationException;
import com.day.cq.replication.impl.AgentConfigImpl;
import com.day.cq.replication.impl.AgentConfigStatefulWrapper;
import com.day.cq.replication.impl.AgentMBean;
import com.day.cq.replication.impl.AgentMBeanImpl;
import com.day.cq.replication.impl.AgentManagerImpl;
import com.day.cq.replication.impl.ReplicationContentFilterChainImpl;
import com.day.cq.replication.impl.ReplicationLogImpl;
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.ReplicationContentWrapper;
import com.day.cq.replication.impl.queue.ReplicationQueueImpl;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import javax.jcr.Session;
import javax.management.ObjectName;
import org.apache.jackrabbit.util.ISO8601;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceRegistration;

public class AgentImpl
implements Agent,
QueueListener {
    private String configError;
    private AgentConfig config;
    private final String id;
    private final ServiceTracker serviceTracker;
    private final AgentManagerImpl agentMgr;
    private final ReplicationLogImpl log;
    private ReplicationQueue queue;
    private final ReplicationContentFactory contentFactory;
    private InternalTransportContext ctx;
    private ServiceRegistration mbeanServiceRegistration;
    private String distributionAgentName = null;
    private boolean expectReverseReplicationErrors = false;

    public AgentImpl(String id, AgentConfig config, ServiceTracker serviceTracker, AgentManagerImpl agentMgr, ReplicationContentFactory binFactory) {
        this.id = id;
        this.serviceTracker = serviceTracker;
        this.agentMgr = agentMgr;
        this.contentFactory = binFactory;
        this.log = new ReplicationLogImpl(id, ReplicationLog.Level.ERROR);
        this.update(config);
    }

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

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

    ReplicationContentFactory getContentFactory() {
        return this.contentFactory;
    }

    public void update(AgentConfig newConfig) {
        String[] uris;
        if (this.stop()) {
            this.log.info("Agent stopped for (re)configuration");
        }
        this.configError = null;
        this.config = newConfig;
        this.log.info("Agent configuration updated at %s", this.config.getConfigPath());
        try {
            this.config.checkValid();
        }
        catch (IllegalArgumentException e) {
            this.configError = e.getMessage();
            this.log.error("Agent not valid. %s", this.configError);
            return;
        }
        if (this.serviceTracker.getTransportHandler(this.config) == null) {
            this.log.info("Agent not valid yet, no transport handler for %s", this.config.getTransportURI());
        }
        if (this.serviceTracker.getContentBuilder(this.config.getSerializationType()) == null) {
            this.log.info("Agent not valid yet. No content builder for %s", this.config.getSerializationType());
        }
        if (newConfig instanceof AgentConfigImpl) {
            this.distributionAgentName = ((AgentConfigImpl)newConfig).getDistributionAgentName();
        }
        if ((uris = this.config.getAllTransportURIs()) != null && uris.length > 1) {
            this.config = new AgentConfigStatefulWrapper(this.config);
        }
        this.ctx = new InternalTransportContext(this.config);
        if (this.config.getLogLevel() != null) {
            this.log.setLevel(this.config.getLogLevel());
        }
        Object[] arrobject = new Object[2];
        arrobject[0] = this.isEnabled() ? "enabled" : "disabled";
        arrobject[1] = this.isValid() ? "" : "in";
        this.log.info("Agent updated. State is %s and %svalid", arrobject);
        this.start();
    }

    public boolean start() {
        if (this.isEnabled() && this.isValid() && this.queue == null) {
            this.queue = this.distributionAgentName != null ? this.agentMgr.createDistributionQueue(this.distributionAgentName) : this.agentMgr.createQueue(this.id, this);
            this.log.info("Agent started.");
            this.queue.forceRetry();
            return true;
        }
        return false;
    }

    public void init(BundleContext bc) {
        try {
            Hashtable<String, String> mbeanProps = new Hashtable<String, String>();
            mbeanProps.put("jmx.objectname", "com.adobe.granite.replication:type=agent,id=" + ObjectName.quote(this.id));
            AgentMBeanImpl mbean = new AgentMBeanImpl(this);
            this.mbeanServiceRegistration = bc.registerService(AgentMBean.class.getName(), (Object)mbean, mbeanProps);
        }
        catch (Throwable t) {
            this.log.warn("Unable to register mbean: {}", t.getLocalizedMessage());
        }
    }

    public void destroy() {
        this.destroy(true);
    }

    public void destroy(boolean clear) {
        if (this.mbeanServiceRegistration != null) {
            this.mbeanServiceRegistration.unregister();
            this.mbeanServiceRegistration = null;
        }
        if (this.queue != null) {
            if (this.queue instanceof ReplicationQueueImpl) {
                ((ReplicationQueueImpl)this.queue).close(clear);
            }
            this.queue = null;
            this.log.info("Agent Destroyed.");
        }
        if (this.contentFactory != null) {
            this.contentFactory.close();
        }
    }

    public boolean stop() {
        if (this.queue != null) {
            if (this.queue instanceof ReplicationQueueImpl) {
                ((ReplicationQueueImpl)this.queue).close(false);
            }
            this.queue = null;
            this.log.info("Agent stopped.");
            return true;
        }
        return false;
    }

    @Override
    public boolean isEnabled() {
        return this.config.isEnabled();
    }

    @Override
    public boolean isValid() {
        return this.configError == null && this.serviceTracker.getContentBuilder(this.config.getSerializationType()) != null;
    }

    @Override
    public void checkValid() {
        if (this.configError != null) {
            throw new IllegalArgumentException(this.configError);
        }
        if (this.serviceTracker.getContentBuilder(this.config.getSerializationType()) == null) {
            throw new IllegalArgumentException("ContentBuilder not available");
        }
    }

    @Override
    public ReplicationLog getLog() {
        return this.log;
    }

    @Override
    public ReplicationContent buildContent(Session session, ReplicationAction action) throws ReplicationException {
        return this.buildContent(session, action, null);
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    @Override
    public ReplicationContent buildContent(Session session, ReplicationAction action, Map<String, Object> parameters) throws ReplicationException {
        this.log.info("Creating content for path %s", action.getPath());
        ContentBuilder builder = this.serviceTracker.getContentBuilder(this.config.getSerializationType());
        if (builder == null) {
            throw new ReplicationException("ContentBuilder not available");
        }
        action.setConfig(this.config);
        action.setLog(this.log);
        try {
            HashMap<String, Object> builderParameters = new HashMap<String, Object>();
            if (parameters != null) {
                builderParameters.putAll(parameters);
            }
            builderParameters.put(ContentBuilder.PARAMETER_CONTENT_FILER_CHAIN, new ReplicationContentFilterChainImpl(this.agentMgr, action));
            ReplicationContent content = builder.create(session, action, this.contentFactory, builderParameters);
            content.acquire(this.getId());
            ReplicationContent replicationContent = content;
            return replicationContent;
        }
        finally {
            action.setLog(null);
            action.setConfig(null);
        }
    }

    @Override
    public ReplicationContent getContent(ReplicationContentFacade facade) throws ReplicationException {
        ReplicationContent content = this.contentFactory.create(facade);
        content.acquire(this.getId());
        return content;
    }

    @Override
    public boolean process(ReplicationQueueImpl.EntryData data) {
        try {
            ReplicationContent content = this.contentFactory.create(data.getContent());
            return this.doReplicate(data.getAction(), content, null, this.ctx);
        }
        catch (ReplicationException e) {
            this.log.error("Error during replication of %s: %s", data.getAction(), e.toString());
            return false;
        }
    }

    @Override
    public List<QueueListener.JobStatus> process(List<ReplicationQueueImpl.EntryData> dataObjects) {
        ArrayList<ReplicationContentWrapper.Entry> contentObjects = new ArrayList<ReplicationContentWrapper.Entry>();
        for (ReplicationQueueImpl.EntryData entry : dataObjects) {
            ReplicationContent content = this.contentFactory.create(entry.getContent());
            ReplicationContentWrapper.Entry e = new ReplicationContentWrapper.Entry(entry, content);
            contentObjects.add(e);
        }
        ReplicationContentWrapper wrapper = new ReplicationContentWrapper(contentObjects);
        ReplicationAction template = dataObjects.get(0).getAction();
        ReplicationAction action = new ReplicationAction(template.getType(), "batch:");
        QueueListener.JobStatus defaultResult = QueueListener.JobStatus.SUCCEEDED;
        try {
            if (!this.doReplicate(action, wrapper, null, this.ctx)) {
                defaultResult = QueueListener.JobStatus.FAILED;
            }
        }
        catch (ReplicationException e) {
            defaultResult = QueueListener.JobStatus.FAILED;
            this.log.error("Error during replication of %s: %s", action, e.toString());
        }
        ArrayList<QueueListener.JobStatus> resultList = new ArrayList<QueueListener.JobStatus>();
        for (ReplicationContentWrapper.Entry entry2 : contentObjects) {
            if (entry2.failed) {
                resultList.add(QueueListener.JobStatus.CANCELLED);
                continue;
            }
            resultList.add(defaultResult);
        }
        return resultList;
    }

    @Override
    public void abort(ReplicationQueueImpl.EntryData data) {
        ReplicationContent content = this.contentFactory.create(data.getContent());
        if (content != null) {
            content.release(this.getId());
        }
    }

    private synchronized boolean doReplicate(ReplicationAction action, ReplicationContent content, ReplicationOptions opts, InternalTransportContext ctx) throws ReplicationException {
        return this.doTransport(action, content, opts, ctx);
    }

    private boolean doTransport(final ReplicationAction action, final ReplicationContent content, ReplicationOptions opts, InternalTransportContext ctx) throws ReplicationException {
        TransportHandler handler = this.serviceTracker.getTransportHandler(this.config);
        if (handler == null) {
            throw new ReplicationException("Unable to replicate content with no transport handler. Check configuration.");
        }
        ReplicationListener listener = opts == null ? null : opts.getListener();
        final ListenerLogDelegator txLog = new ListenerLogDelegator(this.log, listener);
        ReplicationTransaction tx = new ReplicationTransaction(){

            @Override
            public ReplicationAction getAction() {
                return action;
            }

            @Override
            public ReplicationContent getContent() {
                return content;
            }

            @Override
            public ReplicationLog getLog() {
                return txLog;
            }
        };
        try {
            if (listener != null) {
                listener.onStart(this, action);
            }
            ReplicationResult result = handler.deliver(ctx, tx);
            if (listener != null) {
                listener.onEnd(this, action, result);
            }
            if (result.isSuccess()) {
                if (content != null) {
                    content.release(this.getId());
                }
            } else {
                ctx.error();
            }
            return result.isSuccess();
        }
        catch (UnrecoverableReplicationException e) {
            txLog.warn("Unrecoverable error occurred during replication. Aborting: " + e.getMessage(), e);
            if (listener != null) {
                listener.onError(this, action, e);
            }
            if (content != null) {
                content.release(this.getId());
            }
            return true;
        }
        catch (ReplicationException e) {
            if (listener != null) {
                listener.onError(this, action, e);
            }
            throw e;
        }
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     */
    @Override
    public void replicate(ReplicationAction action, ReplicationContent content, ReplicationOptions opts) throws ReplicationException {
        if (opts.isSynchronous() || this.distributionAgentName != null) {
            block10 : {
                InternalTransportContext ctx = new InternalTransportContext(this.config);
                try {
                    if (this.distributionAgentName != null) {
                        this.doTransport(action, content, opts, ctx);
                        break block10;
                    }
                    this.doReplicate(action, content, opts, ctx);
                }
                finally {
                    ctx.discard();
                }
            }
            if (content != null) {
                content.release(this.getId());
            }
        } else {
            ReplicationQueueImpl q = (ReplicationQueueImpl)this.getQueue();
            if (q != null) {
                try {
                    q.addEntry(action, content == null ? null : content.getFacade(), this.config.getRetryDelay());
                }
                catch (Exception e) {
                    throw new ReplicationException(e);
                }
            }
        }
    }

    @Override
    public ReverseReplication[] poll(ReplicationAction action) throws ReplicationException {
        TransportHandler handler = this.serviceTracker.getTransportHandler(this.config);
        if (!(handler instanceof PollingTransportHandler)) {
            String msg = "Unable to replicate content with no polling transport handler. Check configuration.";
            throw new ReplicationException(msg);
        }
        Calendar lastPoll = this.agentMgr.getLastPollTime(this.getId());
        String rev = lastPoll == null ? null : ISO8601.format((Calendar)lastPoll);
        final ReplicationAction newAction = new ReplicationAction(action.getType(), action.getPath(), action.getTime(), action.getUserId(), rev);
        ReplicationTransaction tx = new ReplicationTransaction(){

            @Override
            public ReplicationAction getAction() {
                return newAction;
            }

            @Override
            public ReplicationContent getContent() {
                return null;
            }

            @Override
            public ReplicationLog getLog() {
                return new ReverseReplicationLog(AgentImpl.this.log);
            }
        };
        ReplicationResult result = ReplicationResult.OK;
        ReverseReplication[] ret = new ReverseReplication[]{};
        try {
            ret = ((PollingTransportHandler)((Object)handler)).poll(this.ctx, tx, this.contentFactory);
        }
        catch (TransportException e) {
            result = e.getResult();
        }
        for (ReverseReplication r2 : ret) {
            r2.getContent().acquire(this.getId());
        }
        if (result.isSuccess()) {
            this.log.info("Fetched %d contents from %s since %s", ret.length, this.getConfiguration().getTransportURI(), rev);
            this.expectReverseReplicationErrors = false;
            return ret;
        }
        this.expectReverseReplicationErrors = true;
        for (ReverseReplication r2 : ret) {
            r2.getContent().release(this.getId());
        }
        this.ctx.error();
        return null;
    }

    @Override
    public void setNextPollTimeline(Calendar time) {
        this.agentMgr.setLastPollTime(this.getId(), time);
    }

    @Override
    public Calendar getLastPollTimeline() {
        return this.agentMgr.getLastPollTime(this.getId());
    }

    @Override
    public boolean isInMaintenanceMode() {
        return this.config.isInMaintenanceMode();
    }

    @Override
    public boolean isCacheInvalidator() {
        ContentBuilder cb = this.serviceTracker.getContentBuilder(this.config.getSerializationType());
        return cb != null && cb.getName().equals("flush");
    }

    @Override
    public AgentConfig getConfiguration() {
        return this.config;
    }

    private static class InternalTransportContext
    implements TransportContext {
        private final AgentConfig config;
        private final Map<String, TransportContext.Discardable> attrs = new HashMap<String, TransportContext.Discardable>();

        InternalTransportContext(AgentConfig config) {
            this.config = config;
        }

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

        @Override
        public AgentConfig getConfig() {
            return this.config;
        }

        @Override
        public TransportContext.Discardable getAttribute(String name) {
            return this.attrs.get(name);
        }

        @Override
        public TransportContext.Discardable setAttribute(String name, TransportContext.Discardable attr) {
            if (attr == null) {
                return this.attrs.remove(name);
            }
            return this.attrs.put(name, attr);
        }

        public void discard() {
            for (TransportContext.Discardable d : this.attrs.values()) {
                d.discard();
            }
            this.attrs.clear();
        }

        public void error() {
            this.discard();
            if (this.config instanceof AgentConfigStatefulWrapper) {
                ((AgentConfigStatefulWrapper)this.config).skipToNextTransportUri();
            }
        }
    }

    private class ReverseReplicationLog
    extends ListenerLogDelegator {
        public ReverseReplicationLog(ReplicationLog log) {
            super(log, null);
        }

        @Override
        public void info(String message) {
            if (AgentImpl.this.expectReverseReplicationErrors) {
                this.debug("Expected information: " + message);
            } else {
                super.info(message);
            }
        }

        @Override
        public void warn(String message) {
            if (AgentImpl.this.expectReverseReplicationErrors) {
                this.debug("Expected warning: " + message);
            } else {
                super.warn(message);
            }
        }

        @Override
        public void error(String message) {
            if (AgentImpl.this.expectReverseReplicationErrors) {
                this.debug("Expected error: " + message);
            } else {
                super.warn(message);
            }
        }
    }

}