KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > alfresco > repo > content > replication > ContentStoreReplicator


1 /*
2  * Copyright (C) 2006 Alfresco, Inc.
3  *
4  * Licensed under the Mozilla Public License version 1.1
5  * with a permitted attribution clause. You may obtain a
6  * copy of the License at
7  *
8  * http://www.alfresco.org/legal/license.txt
9  *
10  * Unless required by applicable law or agreed to in writing,
11  * software distributed under the License is distributed on an
12  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
13  * either express or implied. See the License for the specific
14  * language governing permissions and limitations under the
15  * License.
16  */
17 package org.alfresco.repo.content.replication;
18
19 import java.util.Set JavaDoc;
20
21 import org.alfresco.error.AlfrescoRuntimeException;
22 import org.alfresco.repo.content.ContentStore;
23 import org.alfresco.repo.node.index.IndexRecovery;
24 import org.alfresco.service.cmr.repository.ContentReader;
25 import org.alfresco.service.cmr.repository.ContentWriter;
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.quartz.Job;
29 import org.quartz.JobExecutionContext;
30 import org.quartz.JobExecutionException;
31
32 /**
33  * This component performs one-way replication between to content stores.
34  * <p>
35  * It ensure that the content from the first store is copied to the second
36  * store where required, therefore primarily acting as a backup or
37  * replication mechanism.
38  * <p>
39  * Once started, this process runs continuously on a low-priority thread
40  * and cannot be restarted.
41  *
42  * @author Derek Hulley
43  */
44 public class ContentStoreReplicator
45 {
46     private static Log logger = LogFactory.getLog(ContentStoreReplicator.class);
47
48     private ContentStore sourceStore;
49     private ContentStore targetStore;
50     
51     /** used to ensure that this instance gets started once only */
52     private boolean started;
53     /** set this on to keep replicating and never stop. The default is <code>true</code>. */
54     private boolean runContinuously;
55     /** the time to wait between passes */
56     private long waitTime;
57
58     public ContentStoreReplicator()
59     {
60         this.started = false;
61         this.runContinuously = true;
62         this.waitTime = 60000L;
63     }
64     
65     /**
66      * Set the source that content must be taken from
67      *
68      * @param sourceStore the content source
69      */
70     public void setSourceStore(ContentStore sourceStore)
71     {
72         this.sourceStore = sourceStore;
73     }
74     
75     /**
76      * Set the target that content must be written to
77      *
78      * @param targetStore the content target
79      */
80     public void setTargetStore(ContentStore targetStore)
81     {
82         this.targetStore = targetStore;
83     }
84
85     /**
86      * Set whether the thread should run continuously or terminate after
87      * a first pass.
88      *
89      * @param runContinuously true to run continously (default)
90      */
91     public void setRunContinuously(boolean runContinuously)
92     {
93         this.runContinuously = runContinuously;
94     }
95
96     /**
97      * Set the time to wait between replication passes (in seconds)
98      *
99      * @param waitTime the time between passes (in seconds). Default is 60s.
100      */
101     public void setWaitTime(long waitTime)
102     {
103         // convert to millis
104 this.waitTime = waitTime * 1000L;
105     }
106
107     /**
108      * Kick off the replication thread. This method can be used once.
109      */
110     public synchronized void start()
111     {
112         if (started)
113         {
114             throw new AlfrescoRuntimeException("This ContentStoreReplicator has already been started");
115         }
116         // create a low-priority, daemon thread to do the work
117 Runnable JavaDoc runnable = new ReplicationRunner();
118         Thread JavaDoc thread = new Thread JavaDoc(runnable);
119         thread.setPriority(Thread.MIN_PRIORITY);
120         thread.setDaemon(true);
121         // start it
122 thread.start();
123     }
124     
125     /**
126      * Stateful thread runnable that performs the replication.
127      *
128      * @author Derek Hulley
129      */
130     private class ReplicationRunner implements Runnable JavaDoc
131     {
132         public void run()
133         {
134             // keep this thread going permanently
135 while (true)
136             {
137                 try
138                 {
139                     ContentStoreReplicator.this.replicate();
140                     // check if the process should terminate
141 if (!runContinuously)
142                     {
143                         // the thread has caught up with all the available work and should not
144 // run continuously
145 if (logger.isDebugEnabled())
146                         {
147                             logger.debug("Thread quitting - first pass of replication complete:");
148                         }
149                         break;
150                     }
151                     // pause the the required wait time
152 synchronized(ContentStoreReplicator.this)
153                     {
154                         ContentStoreReplicator.this.wait(waitTime);
155                     }
156                 }
157                 catch (InterruptedException JavaDoc e)
158                 {
159                     // ignore
160 }
161                 catch (Throwable JavaDoc e)
162                 {
163                     // report
164 logger.error("Replication failure", e);
165                 }
166             }
167         }
168     }
169     
170     /**
171      * Perform a full replication of all source to target URLs.
172      */
173     private void replicate()
174     {
175         // get all the URLs from the source
176 Set JavaDoc<String JavaDoc> sourceUrls = sourceStore.getUrls();
177         // get all the URLs from the target
178 Set JavaDoc<String JavaDoc> targetUrls = targetStore.getUrls();
179         // remove source URLs that are present in the target
180 sourceUrls.removeAll(targetUrls);
181         
182         // ensure that each remaining source URL is present in the target
183 for (String JavaDoc contentUrl : sourceUrls)
184         {
185             replicate(contentUrl);
186         }
187     }
188     
189     /**
190      * Checks if the target store has the URL, and if not, replicates the content.
191      * <p>
192      * Any failures are reported and not thrown, but the target URL is removed for
193      * good measure.
194      *
195      * @param contentUrl the URL to replicate
196      */
197     private void replicate(String JavaDoc contentUrl)
198     {
199         try
200         {
201             // check that the target doesn't have it
202 if (targetStore.exists(contentUrl))
203             {
204                 // ignore this as the target has it already
205 if (logger.isDebugEnabled())
206                 {
207                     logger.debug("No replication required - URL exists in target store: \n" +
208                             " source store: " + sourceStore + "\n" +
209                             " target store: " + targetStore + "\n" +
210                             " content URL: " + contentUrl);
211                 }
212                 return;
213             }
214             // get a writer to the target store - this can fail if the content is there now
215 ContentWriter writer = targetStore.getWriter(null, contentUrl);
216             // get the source reader
217 ContentReader reader = sourceStore.getReader(contentUrl);
218             if (!reader.exists())
219             {
220                 // the content may have disappeared from the source store
221 if (logger.isDebugEnabled())
222                 {
223                     logger.debug("Source store no longer has URL - no replication possible: \n" +
224                             " source store: " + sourceStore + "\n" +
225                             " target store: " + targetStore + "\n" +
226                             " content URL: " + contentUrl);
227                 }
228                 return;
229             }
230             // copy from the reader to the writer
231 writer.putContent(reader);
232         }
233         catch (Throwable JavaDoc e)
234         {
235             logger.error("Failed to replicate URL - removing target content: \n" +
236                     " source store: " + sourceStore + "\n" +
237                     " target store: " + targetStore + "\n" +
238                     " content URL: " + contentUrl,
239                     e);
240             targetStore.delete(contentUrl);
241         }
242     }
243
244     /**
245      * Kicks off the {@link ContentStoreReplicator content store replicator}.
246      *
247      * @author Derek Hulley
248      */
249     public class ContentStoreReplicatorJob implements Job
250     {
251         /** KEY_CONTENT_STORE_REPLICATOR = 'contentStoreReplicator' */
252         public static final String JavaDoc KEY_CONTENT_STORE_REPLICATOR = "contentStoreReplicator";
253         
254         /**
255          * Forces a full index recovery using the {@link IndexRecovery recovery component} passed
256          * in via the job detail.
257          */
258         public void execute(JobExecutionContext context) throws JobExecutionException
259         {
260             ContentStoreReplicator contentStoreReplicator = (ContentStoreReplicator) context.getJobDetail()
261                     .getJobDataMap().get(KEY_CONTENT_STORE_REPLICATOR);
262             if (contentStoreReplicator == null)
263             {
264                 throw new JobExecutionException("Missing job data: " + KEY_CONTENT_STORE_REPLICATOR);
265             }
266             // reindex
267 contentStoreReplicator.start();
268         }
269     }
270 }
271
Popular Tags