clusternode.java

来自「jsr170接口的java实现。是个apache的开源项目。」· Java 代码 · 共 1,278 行 · 第 1/3 页

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.jackrabbit.core.cluster;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.jackrabbit.core.config.ClusterConfig;
import org.apache.jackrabbit.core.config.ConfigurationException;
import org.apache.jackrabbit.core.config.JournalConfig;
import org.apache.jackrabbit.core.NodeId;
import org.apache.jackrabbit.core.journal.Journal;
import org.apache.jackrabbit.core.journal.RecordConsumer;
import org.apache.jackrabbit.core.journal.Record;
import org.apache.jackrabbit.core.journal.JournalException;
import org.apache.jackrabbit.core.journal.FileRevision;
import org.apache.jackrabbit.core.nodetype.InvalidNodeTypeDefException;
import org.apache.jackrabbit.core.nodetype.NodeTypeDef;
import org.apache.jackrabbit.core.observation.EventState;
import org.apache.jackrabbit.core.observation.EventStateCollection;
import org.apache.jackrabbit.core.state.ChangeLog;
import org.apache.jackrabbit.core.state.ItemState;
import org.apache.jackrabbit.core.state.NodeState;
import org.apache.jackrabbit.core.state.PropertyState;
import org.apache.jackrabbit.name.QName;
import org.apache.jackrabbit.name.Path;
import EDU.oswego.cs.dl.util.concurrent.Mutex;

import javax.jcr.RepositoryException;
import javax.jcr.Session;
import javax.jcr.observation.Event;
import java.util.List;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import java.util.HashSet;
import java.io.File;

/**
 * Default clustered node implementation.
 */
public class ClusterNode implements Runnable,
        NamespaceEventChannel, NodeTypeEventChannel, RecordConsumer  {

    /**
     * System property specifying a node id to use.
     */
    public static final String SYSTEM_PROPERTY_NODE_ID = "org.apache.jackrabbit.core.cluster.node_id";

    /**
     * Revision counter parameter name.
     */
    private static final String REVISION_NAME = "revision";

    /**
     * Used for padding short string representations.
     */
    private static final String SHORT_PADDING = "0000";

    /**
     * Producer identifier.
     */
    private static final String PRODUCER_ID = "JR";

    /**
     * Status constant.
     */
    private static final int NONE = 0;

    /**
     * Status constant.
     */
    private static final int STARTED = 1;

    /**
     * Status constant.
     */
    private static final int STOPPED = 2;

    /**
     * Bit indicating this is a registration operation.
     */
    private static final int NTREG_REGISTER = 0;

    /**
     * Bit indicating this is a reregistration operation.
     */
    private static final int NTREG_REREGISTER = (1 << 30);

    /**
     * Bit indicating this is an unregistration operation.
     */
    private static final int NTREG_UNREGISTER = (1 << 31);

    /**
     * Mask used in node type registration operations.
     */
    private static final int NTREG_MASK = (NTREG_REREGISTER | NTREG_UNREGISTER);

    /**
     * Logger.
     */
    private static Logger log = LoggerFactory.getLogger(ClusterNode.class);

    /**
     * Cluster context.
     */
    private ClusterContext clusterContext;

    /**
     * Cluster node id.
     */
    private String clusterNodeId;

    /**
     * Synchronization delay, in milliseconds.
     */
    private long syncDelay;

    /**
     * Journal used.
     */
    private Journal journal;

    /**
     * Mutex used when syncing.
     */
    private final Mutex syncLock = new Mutex();

    /**
     * Status flag, one of {@link #NONE}, {@link #STARTED} or {@link #STOPPED}.
     */
    private int status;

    /**
     * Map of available lock listeners, indexed by workspace name.
     */
    private final Map wspLockListeners = new HashMap();

    /**
     * Map of available update listeners, indexed by workspace name.
     */
    private final Map wspUpdateListeners = new HashMap();

    /**
     * Versioning update listener.
     */
    private UpdateEventListener versionUpdateListener;

    /**
     * Namespace listener.
     */
    private NamespaceEventListener namespaceListener;

    /**
     * Node type listener.
     */
    private NodeTypeEventListener nodeTypeListener;

    /**
     * Instance revision file.
     */
    private FileRevision instanceRevision;

    /**
     * Workspace name used when consuming records.
     */
    private String workspace;

    /**
     * Change log used when consuming records.
     */
    private ChangeLog changeLog;

    /**
     * List of recorded events; used when consuming records.
     */
    private List events;

    /**
     * Last used session for event sources.
     */
    private Session lastSession;

    /**
     * Initialize this cluster node.
     *
     * @throws ClusterException if an error occurs
     */
    public void init(ClusterContext clusterContext) throws ClusterException {
        this.clusterContext = clusterContext;

        init();
    }

    /**
     * Initialize this cluster node (overridable).
     *
     * @throws ClusterException if an error occurs
     */
    protected void init() throws ClusterException {
        ClusterConfig cc = clusterContext.getClusterConfig();
        clusterNodeId = getClusterNodeId(cc.getId());
        syncDelay = cc.getSyncDelay();

        JournalConfig jc = cc.getJournalConfig();

        String revisionName = jc.getParameters().getProperty(REVISION_NAME);
        if (revisionName == null) {
            String msg = "Revision not specified.";
            throw new ClusterException(msg);
        }
        try {
            instanceRevision = new FileRevision(new File(revisionName));

            journal = (Journal) jc.newInstance();
            journal.init(clusterNodeId, clusterContext.getNamespaceResovler());
            journal.register(this);
        } catch (ConfigurationException e) {
            throw new ClusterException(e.getMessage(), e.getCause());
        } catch (JournalException e) {
            throw new ClusterException(e.getMessage(), e.getCause());
        }
    }

    /**
     * Starts this cluster node.
     *
     * @throws ClusterException if an error occurs
     */
    public synchronized void start() throws ClusterException {
        if (status == NONE) {
            sync();

            Thread t = new Thread(this, "ClusterNode-" + clusterNodeId);
            t.setDaemon(true);
            t.start();

            status = STARTED;
        }
    }

    /**
     * Run loop that will sync this node after some delay.
     */
    public void run() {
        for (;;) {
            synchronized (this) {
                try {
                    wait(syncDelay);
                } catch (InterruptedException e) {}

                if (status == STOPPED) {
                    return;
                }
            }
            try {
                sync();
            } catch (ClusterException e) {
                String msg = "Periodic sync of journal failed: " + e.getMessage();
                log.error(msg);
            } catch (Exception e) {
                String msg = "Unexpected error while syncing of journal: " + e.getMessage();
                log.error(msg, e);
            } catch (Error e) {
                String msg = "Unexpected error while syncing of journal: " + e.getMessage();
                log.error(msg, e);
                throw e;
            }
        }
    }

    /**
     * Synchronize contents from journal.
     *
     * @throws ClusterException if an error occurs
     */
    public void sync() throws ClusterException {
        try {
            syncLock.acquire();
        } catch (InterruptedException e) {
            String msg = "Interrupted while waiting for mutex.";
            throw new ClusterException(msg);
        }

        try {
            journal.sync();
        } catch (JournalException e) {
            throw new ClusterException(e.getMessage(), e.getCause());
        } finally {
            syncLock.release();
        }
    }

    /**
     * Stops this cluster node.
     */
    public synchronized void stop() {
        if (status == STARTED) {
            status = STOPPED;

            journal.close();
            notifyAll();
        }
    }

    /**
     * Create an {@link UpdateEventChannel} for some workspace.
     *
     * @param workspace workspace name
     * @return lock event channel
     */
    public UpdateEventChannel createUpdateChannel(String workspace) {
        return new WorkspaceUpdateChannel(workspace);
    }

    /**
     * Create a {@link LockEventChannel} for some workspace.
     *
     * @param workspace workspace name
     * @return lock event channel
     */
    public LockEventChannel createLockChannel(String workspace) {
        return new WorkspaceLockChannel(workspace);
    }

    /**
     * Return the instance id to be used for this node in the cluster.
     * @param id configured id, <code>null</code> to take random id
     */
    private String getClusterNodeId(String id) {
        if (id == null) {
            id = System.getProperty(SYSTEM_PROPERTY_NODE_ID);
            if (id == null) {
                id = toHexString((short) (Math.random() * (Short.MAX_VALUE - Short.MIN_VALUE)));
            }
        }
        return id;
    }

    /**
     * Return a zero-padded short string representation.
     *
     * @param n short
     * @return string representation
     */
    private static String toHexString(short n) {
        String s = Integer.toHexString(n);
        int padlen = SHORT_PADDING.length() - s.length();
        if (padlen < 0) {
            s = s.substring(-padlen);
        } else if (padlen > 0) {
            s = SHORT_PADDING.substring(0, padlen) + s;
        }
        return s;
    }

    //-----------------------------------------------< NamespaceEventListener >

    /**
     * {@inheritDoc}
     */
    public void remapped(String oldPrefix, String newPrefix, String uri) {
        if (status != STARTED) {
            log.info("not started: namespace operation ignored.");
            return;
        }
        Record record = null;
        boolean succeeded = false;

        try {
            record = journal.getProducer(PRODUCER_ID).append();
            record.writeString(null);
            write(record, oldPrefix, newPrefix, uri);
            record.writeChar('\0');
            record.update();
            setRevision(record.getRevision());
            succeeded = true;
        } catch (JournalException e) {
            String msg = "Unable to create log entry: " + e.getMessage();
            log.error(msg);
        } catch (Throwable e) {
            String msg = "Unexpected error while creating log entry.";
            log.error(msg, e);
        } finally {
            if (!succeeded && record != null) {
                record.cancelUpdate();
            }
        }
    }

    public void setListener(NamespaceEventListener listener) {
        namespaceListener = listener;
    }

    //------------------------------------------------< NodeTypeEventListener >

    /**
     * {@inheritDoc}
     */
    public void registered(Collection ntDefs) {
        if (status != STARTED) {
            log.info("not started: nodetype operation ignored.");
            return;
        }
        Record record = null;
        boolean succeeded = false;

        try {
            record = journal.getProducer(PRODUCER_ID).append();