001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.camel.component.file;
018
019import java.io.IOException;
020import java.lang.reflect.Method;
021import java.nio.file.attribute.PosixFilePermission;
022import java.util.ArrayList;
023import java.util.Comparator;
024import java.util.HashMap;
025import java.util.HashSet;
026import java.util.List;
027import java.util.Map;
028import java.util.Set;
029
030import org.apache.camel.CamelContext;
031import org.apache.camel.Component;
032import org.apache.camel.Exchange;
033import org.apache.camel.Expression;
034import org.apache.camel.ExpressionIllegalSyntaxException;
035import org.apache.camel.LoggingLevel;
036import org.apache.camel.Message;
037import org.apache.camel.Processor;
038import org.apache.camel.impl.ScheduledPollEndpoint;
039import org.apache.camel.processor.idempotent.MemoryIdempotentRepository;
040import org.apache.camel.spi.BrowsableEndpoint;
041import org.apache.camel.spi.FactoryFinder;
042import org.apache.camel.spi.IdempotentRepository;
043import org.apache.camel.spi.Language;
044import org.apache.camel.spi.UriParam;
045import org.apache.camel.util.FileUtil;
046import org.apache.camel.util.IOHelper;
047import org.apache.camel.util.ObjectHelper;
048import org.apache.camel.util.ServiceHelper;
049import org.apache.camel.util.StringHelper;
050import org.slf4j.Logger;
051import org.slf4j.LoggerFactory;
052
053/**
054 * Base class for file endpoints
055 */
056public abstract class GenericFileEndpoint<T> extends ScheduledPollEndpoint implements BrowsableEndpoint {
057
058    protected static final String DEFAULT_STRATEGYFACTORY_CLASS = "org.apache.camel.component.file.strategy.GenericFileProcessStrategyFactory";
059    protected static final int DEFAULT_IDEMPOTENT_CACHE_SIZE = 1000;
060    
061    private static final Integer CHMOD_WRITE_MASK = 02;
062    private static final Integer CHMOD_READ_MASK = 04;
063    private static final Integer CHMOD_EXECUTE_MASK = 01;
064
065    protected final Logger log = LoggerFactory.getLogger(getClass());
066
067    // common options
068
069    @UriParam(defaultValue = "true")
070    protected boolean autoCreate = true;
071    @UriParam(defaultValue = "" + FileUtil.BUFFER_SIZE)
072    protected int bufferSize = FileUtil.BUFFER_SIZE;
073    @UriParam
074    protected boolean flatten;
075    @UriParam
076    protected String charset;
077    @UriParam
078    protected Expression fileName;
079
080    // producer options
081
082    @UriParam(label = "producer", defaultValue = "Override")
083    protected GenericFileExist fileExist = GenericFileExist.Override;
084    @UriParam(label = "producer")
085    protected String tempPrefix;
086    @UriParam(label = "producer")
087    protected Expression tempFileName;
088    @UriParam(label = "producer", defaultValue = "true")
089    protected boolean eagerDeleteTargetFile = true;
090    @UriParam(defaultValue = "false", label = "producer")
091    protected boolean keepLastModified;
092    @UriParam(label = "producer")
093    protected String doneFileName;
094    @UriParam(label = "producer", defaultValue = "false")
095    protected boolean allowNullBody;
096    @UriParam(label = "producer")
097    protected String chmod;
098
099    // consumer options
100
101    @UriParam
102    protected GenericFileConfiguration configuration;
103    @UriParam(label = "consumer")
104    protected GenericFileProcessStrategy<T> processStrategy;
105    @UriParam(label = "consumer")
106    protected IdempotentRepository<String> inProgressRepository = new MemoryIdempotentRepository();
107    @UriParam(label = "consumer")
108    protected String localWorkDirectory;
109    @UriParam(label = "consumer", defaultValue = "false")
110    protected boolean startingDirectoryMustExist;
111    @UriParam(label = "consumer", defaultValue = "false")
112    protected boolean directoryMustExist;
113    @UriParam(label = "consumer", defaultValue = "false")
114    protected boolean noop;
115    @UriParam(label = "consumer", defaultValue = "false")
116    protected boolean recursive;
117    @UriParam(label = "consumer", defaultValue = "false")
118    protected boolean delete;
119    @UriParam(label = "consumer")
120    protected int maxMessagesPerPoll;
121    @UriParam(label = "consumer", defaultValue = "true")
122    protected boolean eagerMaxMessagesPerPoll = true;
123    @UriParam(label = "consumer", defaultValue = "" + Integer.MAX_VALUE)
124    protected int maxDepth = Integer.MAX_VALUE;
125    @UriParam(label = "consumer")
126    protected int minDepth;
127    @UriParam(label = "consumer")
128    protected String include;
129    @UriParam(label = "consumer")
130    protected String exclude;
131    @UriParam(label = "consumer")
132    protected Expression move;
133    @UriParam(label = "consumer")
134    protected Expression moveFailed;
135    @UriParam(label = "consumer")
136    protected Expression preMove;
137    @UriParam(label = "producer")
138    protected Expression moveExisting;
139    @UriParam(label = "consumer")
140    protected Boolean idempotent;
141    @UriParam(label = "consumer")
142    protected Expression idempotentKey;
143    @UriParam(label = "consumer")
144    protected IdempotentRepository<String> idempotentRepository;
145    @UriParam(label = "consumer")
146    protected GenericFileFilter<T> filter;
147    protected volatile AntPathMatcherGenericFileFilter<T> antFilter;
148    @UriParam(label = "consumer")
149    protected String antInclude;
150    @UriParam(label = "consumer")
151    protected String antExclude;
152    @UriParam(label = "consumer")
153    protected Comparator<GenericFile<T>> sorter;
154    @UriParam(label = "consumer")
155    protected Comparator<Exchange> sortBy;
156    @UriParam(label = "consumer", enums = "none,markerFile,fileLock,rename,changed")
157    protected String readLock = "none";
158    @UriParam(label = "consumer", defaultValue = "1000")
159    protected long readLockCheckInterval = 1000;
160    @UriParam(label = "consumer", defaultValue = "10000")
161    protected long readLockTimeout = 10000;
162    @UriParam(label = "consumer", defaultValue = "true")
163    protected boolean readLockMarkerFile = true;
164    @UriParam(label = "consumer", defaultValue = "WARN")
165    protected LoggingLevel readLockLoggingLevel = LoggingLevel.WARN;
166    @UriParam(label = "consumer", defaultValue = "1")
167    protected long readLockMinLength = 1;
168    @UriParam(label = "consumer", defaultValue = "0")
169    protected long readLockMinAge;
170    @UriParam(label = "consumer")
171    protected GenericFileExclusiveReadLockStrategy<T> exclusiveReadLockStrategy;
172
173    public GenericFileEndpoint() {
174    }
175
176    public GenericFileEndpoint(String endpointUri, Component component) {
177        super(endpointUri, component);
178    }
179
180    public boolean isSingleton() {
181        return true;
182    }
183
184    public abstract GenericFileConsumer<T> createConsumer(Processor processor) throws Exception;
185
186    public abstract GenericFileProducer<T> createProducer() throws Exception;
187
188    public abstract Exchange createExchange(GenericFile<T> file);
189
190    public abstract String getScheme();
191
192    public abstract char getFileSeparator();
193
194    public abstract boolean isAbsolute(String name);
195
196    /**
197     * Return the file name that will be auto-generated for the given message if
198     * none is provided
199     */
200    public String getGeneratedFileName(Message message) {
201        return StringHelper.sanitize(message.getMessageId());
202    }
203
204    public GenericFileProcessStrategy<T> getGenericFileProcessStrategy() {
205        if (processStrategy == null) {
206            processStrategy = createGenericFileStrategy();
207            log.debug("Using Generic file process strategy: {}", processStrategy);
208        }
209        return processStrategy;
210    }
211
212    /**
213     * This implementation will <b>not</b> load the file content.
214     * Any file locking is neither in use by this implementation..
215     */
216    @Override
217    public List<Exchange> getExchanges() {
218        final List<Exchange> answer = new ArrayList<Exchange>();
219
220        GenericFileConsumer<?> consumer = null;
221        try {
222            // create a new consumer which can poll the exchanges we want to browse
223            // do not provide a processor as we do some custom processing
224            consumer = createConsumer(null);
225            consumer.setCustomProcessor(new Processor() {
226                @Override
227                public void process(Exchange exchange) throws Exception {
228                    answer.add(exchange);
229                }
230            });
231            // do not start scheduler, as we invoke the poll manually
232            consumer.setStartScheduler(false);
233            // start consumer
234            ServiceHelper.startService(consumer);
235            // invoke poll which performs the custom processing, so we can browse the exchanges
236            consumer.poll();
237        } catch (Exception e) {
238            throw ObjectHelper.wrapRuntimeCamelException(e);
239        } finally {
240            try {
241                ServiceHelper.stopService(consumer);
242            } catch (Exception e) {
243                log.debug("Error stopping consumer used for browsing exchanges. This exception will be ignored", e);
244            }
245        }
246
247        return answer;
248    }
249
250    /**
251     * A strategy method to lazily create the file strategy
252     */
253    @SuppressWarnings("unchecked")
254    protected GenericFileProcessStrategy<T> createGenericFileStrategy() {
255        Class<?> factory = null;
256        try {
257            FactoryFinder finder = getCamelContext().getFactoryFinder("META-INF/services/org/apache/camel/component/");
258            log.trace("Using FactoryFinder: {}", finder);
259            factory = finder.findClass(getScheme(), "strategy.factory.", CamelContext.class);
260        } catch (ClassNotFoundException e) {
261            log.trace("'strategy.factory.class' not found", e);
262        } catch (IOException e) {
263            log.trace("No strategy factory defined in 'META-INF/services/org/apache/camel/component/'", e);
264        }
265
266        if (factory == null) {
267            // use default
268            try {
269                log.trace("Using ClassResolver to resolve class: {}", DEFAULT_STRATEGYFACTORY_CLASS);
270                factory = this.getCamelContext().getClassResolver().resolveClass(DEFAULT_STRATEGYFACTORY_CLASS);
271            } catch (Exception e) {
272                log.trace("Cannot load class: {}", DEFAULT_STRATEGYFACTORY_CLASS, e);
273            }
274            // fallback and us this class loader
275            try {
276                if (log.isTraceEnabled()) {
277                    log.trace("Using classloader: {} to resolve class: {}", this.getClass().getClassLoader(), DEFAULT_STRATEGYFACTORY_CLASS);
278                }
279                factory = this.getCamelContext().getClassResolver().resolveClass(DEFAULT_STRATEGYFACTORY_CLASS, this.getClass().getClassLoader());
280            } catch (Exception e) {
281                if (log.isTraceEnabled()) {
282                    log.trace("Cannot load class: {} using classloader: " + this.getClass().getClassLoader(), DEFAULT_STRATEGYFACTORY_CLASS, e);
283                }
284            }
285
286            if (factory == null) {
287                throw new TypeNotPresentException(DEFAULT_STRATEGYFACTORY_CLASS + " class not found", null);
288            }
289        }
290
291        try {
292            Method factoryMethod = factory.getMethod("createGenericFileProcessStrategy", CamelContext.class, Map.class);
293            Map<String, Object> params = getParamsAsMap();
294            log.debug("Parameters for Generic file process strategy {}", params);
295            return (GenericFileProcessStrategy<T>) ObjectHelper.invokeMethod(factoryMethod, null, getCamelContext(), params);
296        } catch (NoSuchMethodException e) {
297            throw new TypeNotPresentException(factory.getSimpleName() + ".createGenericFileProcessStrategy method not found", e);
298        }
299    }
300
301    /**
302     * Chmod value must be between 000 and 777; If there is a leading digit like in 0755 we will ignore it.
303     */
304    public boolean chmodPermissionsAreValid(String chmod) {
305        if (chmod == null || chmod.length() < 3 || chmod.length() > 4) {
306            return false;
307        }
308        String permissionsString = chmod.trim().substring(chmod.length() - 3);  // if 4 digits chop off leading one
309        for (int i = 0; i < permissionsString.length(); i++) {
310            Character c = permissionsString.charAt(i);
311            if (!Character.isDigit(c) || Integer.parseInt(c.toString()) > 7) {
312                return false;
313            }
314        }
315        return true;
316    }
317
318    public Set<PosixFilePermission> getPermissions() {
319        Set<PosixFilePermission> permissions = new HashSet<PosixFilePermission>();
320        if (ObjectHelper.isEmpty(chmod)) {
321            return permissions;
322        }
323
324        String chmodString = chmod.substring(chmod.length() - 3);  // if 4 digits chop off leading one
325
326        Integer ownerValue = Integer.parseInt(chmodString.substring(0, 1));
327        Integer groupValue = Integer.parseInt(chmodString.substring(1, 2));
328        Integer othersValue = Integer.parseInt(chmodString.substring(2, 3));
329
330        if ((ownerValue & CHMOD_WRITE_MASK) > 0) {
331            permissions.add(PosixFilePermission.OWNER_WRITE);
332        }
333        if ((ownerValue & CHMOD_READ_MASK) > 0) {
334            permissions.add(PosixFilePermission.OWNER_READ);
335        }
336        if ((ownerValue & CHMOD_EXECUTE_MASK) > 0) {
337            permissions.add(PosixFilePermission.OWNER_EXECUTE);
338        }
339
340        if ((groupValue & CHMOD_WRITE_MASK) > 0) {
341            permissions.add(PosixFilePermission.GROUP_WRITE);
342        }
343        if ((groupValue & CHMOD_READ_MASK) > 0) {
344            permissions.add(PosixFilePermission.GROUP_READ);
345        }
346        if ((groupValue & CHMOD_EXECUTE_MASK) > 0) {
347            permissions.add(PosixFilePermission.GROUP_EXECUTE);
348        }
349
350        if ((othersValue & CHMOD_WRITE_MASK) > 0) {
351            permissions.add(PosixFilePermission.OTHERS_WRITE);
352        }
353        if ((othersValue & CHMOD_READ_MASK) > 0) {
354            permissions.add(PosixFilePermission.OTHERS_READ);
355        }
356        if ((othersValue & CHMOD_EXECUTE_MASK) > 0) {
357            permissions.add(PosixFilePermission.OTHERS_EXECUTE);
358        }
359
360        return permissions;
361    }
362
363    public String getChmod() {
364        return chmod;
365    }
366
367    /**
368     * Specify the file permissions which is sent by the producer, the chmod value must be between 000 and 777;
369     * If there is a leading digit like in 0755 we will ignore it.
370     */
371    public void setChmod(String chmod) throws Exception {
372        if (ObjectHelper.isNotEmpty(chmod) && chmodPermissionsAreValid(chmod)) {
373            this.chmod = chmod.trim();
374        } else {
375            throw new IllegalArgumentException("chmod option [" + chmod + "] is not valid");
376        }
377    }
378
379    public boolean isNoop() {
380        return noop;
381    }
382
383    /**
384     * If true, the file is not moved or deleted in any way.
385     * This option is good for readonly data, or for ETL type requirements.
386     * If noop=true, Camel will set idempotent=true as well, to avoid consuming the same files over and over again.
387     */
388    public void setNoop(boolean noop) {
389        this.noop = noop;
390    }
391
392    public boolean isRecursive() {
393        return recursive;
394    }
395
396    /**
397     * If a directory, will look for files in all the sub-directories as well.
398     */
399    public void setRecursive(boolean recursive) {
400        this.recursive = recursive;
401    }
402
403    public String getInclude() {
404        return include;
405    }
406
407    /**
408     * Is used to include files, if filename matches the regex pattern.
409     */
410    public void setInclude(String include) {
411        this.include = include;
412    }
413
414    public String getExclude() {
415        return exclude;
416    }
417
418    /**
419     * Is used to exclude files, if filename matches the regex pattern.
420     */
421    public void setExclude(String exclude) {
422        this.exclude = exclude;
423    }
424
425    public String getAntInclude() {
426        return antInclude;
427    }
428
429    /**
430     * Ant style filter inclusion.
431     * Multiple inclusions may be specified in comma-delimited format.
432     */
433    public void setAntInclude(String antInclude) {
434        this.antInclude = antInclude;
435        if (this.antFilter == null) {
436            this.antFilter = new AntPathMatcherGenericFileFilter<T>();
437        }
438        this.antFilter.setIncludes(antInclude);
439    }
440
441    public String getAntExclude() {
442        return antExclude;
443    }
444
445    /**
446     * Ant style filter exclusion. If both antInclude and antExclude are used, antExclude takes precedence over antInclude.
447     * Multiple exclusions may be specified in comma-delimited format.
448     */
449    public void setAntExclude(String antExclude) {
450        this.antExclude = antExclude;
451        if (this.antFilter == null) {
452            this.antFilter = new AntPathMatcherGenericFileFilter<T>();
453        }
454        this.antFilter.setExcludes(antExclude);
455    }
456
457    /**
458     * Sets case sensitive flag on {@link org.apache.camel.component.file.AntPathMatcherFileFilter}
459     */
460    public void setAntFilterCaseSensitive(boolean antFilterCaseSensitive) {
461        if (this.antFilter == null) {
462            this.antFilter = new AntPathMatcherGenericFileFilter<T>();
463        }
464        this.antFilter.setCaseSensitive(antFilterCaseSensitive);
465    }
466
467    public GenericFileFilter<T> getAntFilter() {
468        return antFilter;
469    }
470
471    public boolean isDelete() {
472        return delete;
473    }
474
475    /**
476     * If true, the file will be deleted after it is processed successfully.
477     */
478    public void setDelete(boolean delete) {
479        this.delete = delete;
480    }
481
482    public boolean isFlatten() {
483        return flatten;
484    }
485
486    /**
487     * Flatten is used to flatten the file name path to strip any leading paths, so it's just the file name.
488     * This allows you to consume recursively into sub-directories, but when you eg write the files to another directory
489     * they will be written in a single directory.
490     * Setting this to true on the producer enforces that any file name in CamelFileName header
491     * will be stripped for any leading paths.
492     */
493    public void setFlatten(boolean flatten) {
494        this.flatten = flatten;
495    }
496
497    public Expression getMove() {
498        return move;
499    }
500
501    /**
502     * Expression (such as Simple Language) used to dynamically set the filename when moving it after processing.
503     * To move files into a .done subdirectory just enter .done.
504     */
505    public void setMove(Expression move) {
506        this.move = move;
507    }
508
509    /**
510     * @see #setMove(org.apache.camel.Expression)
511     */
512    public void setMove(String fileLanguageExpression) {
513        String expression = configureMoveOrPreMoveExpression(fileLanguageExpression);
514        this.move = createFileLanguageExpression(expression);
515    }
516
517    public Expression getMoveFailed() {
518        return moveFailed;
519    }
520
521    /**
522     * Sets the move failure expression based on Simple language.
523     * For example, to move files into a .error subdirectory use: .error.
524     * Note: When moving the files to the fail location Camel will handle the error and will not pick up the file again.
525     */
526    public void setMoveFailed(Expression moveFailed) {
527        this.moveFailed = moveFailed;
528    }
529
530    public void setMoveFailed(String fileLanguageExpression) {
531        String expression = configureMoveOrPreMoveExpression(fileLanguageExpression);
532        this.moveFailed = createFileLanguageExpression(expression);
533    }
534
535    public Expression getPreMove() {
536        return preMove;
537    }
538
539    /**
540     * Expression (such as File Language) used to dynamically set the filename when moving it before processing.
541     * For example to move in-progress files into the order directory set this value to order.
542     */
543    public void setPreMove(Expression preMove) {
544        this.preMove = preMove;
545    }
546
547    public void setPreMove(String fileLanguageExpression) {
548        String expression = configureMoveOrPreMoveExpression(fileLanguageExpression);
549        this.preMove = createFileLanguageExpression(expression);
550    }
551
552    public Expression getMoveExisting() {
553        return moveExisting;
554    }
555
556    /**
557     * Expression (such as File Language) used to compute file name to use when fileExist=Move is configured.
558     * To move files into a backup subdirectory just enter backup.
559     * This option only supports the following File Language tokens: "file:name", "file:name.ext", "file:name.noext", "file:onlyname",
560     * "file:onlyname.noext", "file:ext", and "file:parent". Notice the "file:parent" is not supported by the FTP component,
561     * as the FTP component can only move any existing files to a relative directory based on current dir as base.
562     */
563    public void setMoveExisting(Expression moveExisting) {
564        this.moveExisting = moveExisting;
565    }
566
567    public void setMoveExisting(String fileLanguageExpression) {
568        String expression = configureMoveOrPreMoveExpression(fileLanguageExpression);
569        this.moveExisting = createFileLanguageExpression(expression);
570    }
571
572    public Expression getFileName() {
573        return fileName;
574    }
575
576    /**
577     * Use Expression such as File Language to dynamically set the filename.
578     * For consumers, it's used as a filename filter.
579     * For producers, it's used to evaluate the filename to write.
580     * If an expression is set, it take precedence over the CamelFileName header. (Note: The header itself can also be an Expression).
581     * The expression options support both String and Expression types.
582     * If the expression is a String type, it is always evaluated using the File Language.
583     * If the expression is an Expression type, the specified Expression type is used - this allows you,
584     * for instance, to use OGNL expressions. For the consumer, you can use it to filter filenames,
585     * so you can for instance consume today's file using the File Language syntax: mydata-${date:now:yyyyMMdd}.txt.
586     * The producers support the CamelOverruleFileName header which takes precedence over any existing CamelFileName header;
587     * the CamelOverruleFileName is a header that is used only once, and makes it easier as this avoids to temporary
588     * store CamelFileName and have to restore it afterwards.
589     */
590    public void setFileName(Expression fileName) {
591        this.fileName = fileName;
592    }
593
594    public void setFileName(String fileLanguageExpression) {
595        this.fileName = createFileLanguageExpression(fileLanguageExpression);
596    }
597
598    public String getDoneFileName() {
599        return doneFileName;
600    }
601
602    /**
603     * If provided, then Camel will write a 2nd done file when the original file has been written.
604     * The done file will be empty. This option configures what file name to use.
605     * Either you can specify a fixed name. Or you can use dynamic placeholders.
606     * The done file will always be written in the same folder as the original file.
607     * <p/>
608     * Only ${file.name} and ${file.name.noext} is supported as dynamic placeholders.
609     */
610    public void setDoneFileName(String doneFileName) {
611        this.doneFileName = doneFileName;
612    }
613
614    public Boolean isIdempotent() {
615        return idempotent != null ? idempotent : false;
616    }
617
618    public String getCharset() {
619        return charset;
620    }
621
622    /**
623     * This option is used to specify the encoding of the file.
624     * You can use this on the consumer, to specify the encodings of the files, which allow Camel to know the charset
625     * it should load the file content in case the file content is being accessed.
626     * Likewise when writing a file, you can use this option to specify which charset to write the file as well.
627     */
628    public void setCharset(String charset) {
629        IOHelper.validateCharset(charset);
630        this.charset = charset;
631    }
632
633    protected boolean isIdempotentSet() {
634        return idempotent != null;
635    }
636
637    /**
638     * Option to use the Idempotent Consumer EIP pattern to let Camel skip already processed files.
639     * Will by default use a memory based LRUCache that holds 1000 entries. If noop=true then idempotent will be enabled
640     * as well to avoid consuming the same files over and over again.
641     */
642    public void setIdempotent(Boolean idempotent) {
643        this.idempotent = idempotent;
644    }
645
646    public Expression getIdempotentKey() {
647        return idempotentKey;
648    }
649
650    /**
651     * To use a custom idempotent key. By default the absolute path of the file is used.
652     * You can use the File Language, for example to use the file name and file size, you can do:
653     * <tt>idempotentKey=${file:name}-${file:size}</tt>
654     */
655    public void setIdempotentKey(Expression idempotentKey) {
656        this.idempotentKey = idempotentKey;
657    }
658
659    public void setIdempotentKey(String expression) {
660        this.idempotentKey = createFileLanguageExpression(expression);
661    }
662
663    public IdempotentRepository<String> getIdempotentRepository() {
664        return idempotentRepository;
665    }
666
667    /**
668     * A pluggable repository org.apache.camel.spi.IdempotentRepository which by default use MemoryMessageIdRepository
669     * if none is specified and idempotent is true.
670     */
671    public void setIdempotentRepository(IdempotentRepository<String> idempotentRepository) {
672        this.idempotentRepository = idempotentRepository;
673    }
674
675    public GenericFileFilter<T> getFilter() {
676        return filter;
677    }
678
679    /**
680     * Pluggable filter as a org.apache.camel.component.file.GenericFileFilter class.
681     * Will skip files if filter returns false in its accept() method.
682     */
683    public void setFilter(GenericFileFilter<T> filter) {
684        this.filter = filter;
685    }
686
687    public Comparator<GenericFile<T>> getSorter() {
688        return sorter;
689    }
690
691    /**
692     * Pluggable sorter as a java.util.Comparator<org.apache.camel.component.file.GenericFile> class.
693     */
694    public void setSorter(Comparator<GenericFile<T>> sorter) {
695        this.sorter = sorter;
696    }
697
698    public Comparator<Exchange> getSortBy() {
699        return sortBy;
700    }
701
702    /**
703     * Built-in sort by using the File Language.
704     * Supports nested sorts, so you can have a sort by file name and as a 2nd group sort by modified date.
705     */
706    public void setSortBy(Comparator<Exchange> sortBy) {
707        this.sortBy = sortBy;
708    }
709
710    public void setSortBy(String expression) {
711        setSortBy(expression, false);
712    }
713
714    public void setSortBy(String expression, boolean reverse) {
715        setSortBy(GenericFileDefaultSorter.sortByFileLanguage(getCamelContext(), expression, reverse));
716    }
717
718    public String getTempPrefix() {
719        return tempPrefix;
720    }
721
722    /**
723     * This option is used to write the file using a temporary name and then, after the write is complete,
724     * rename it to the real name. Can be used to identify files being written and also avoid consumers
725     * (not using exclusive read locks) reading in progress files. Is often used by FTP when uploading big files.
726     */
727    public void setTempPrefix(String tempPrefix) {
728        this.tempPrefix = tempPrefix;
729        // use only name as we set a prefix in from on the name
730        setTempFileName(tempPrefix + "${file:onlyname}");
731    }
732
733    public Expression getTempFileName() {
734        return tempFileName;
735    }
736
737    /**
738     * The same as tempPrefix option but offering a more fine grained control on the naming of the temporary filename as it uses the File Language.
739     */
740    public void setTempFileName(Expression tempFileName) {
741        this.tempFileName = tempFileName;
742    }
743
744    public void setTempFileName(String tempFileNameExpression) {
745        this.tempFileName = createFileLanguageExpression(tempFileNameExpression);
746    }
747
748    public boolean isEagerDeleteTargetFile() {
749        return eagerDeleteTargetFile;
750    }
751
752    /**
753     * Whether or not to eagerly delete any existing target file.
754     * This option only applies when you use fileExists=Override and the tempFileName option as well.
755     * You can use this to disable (set it to false) deleting the target file before the temp file is written.
756     * For example you may write big files and want the target file to exists during the temp file is being written.
757     * This ensure the target file is only deleted until the very last moment, just before the temp file is being
758     * renamed to the target filename. This option is also used to control whether to delete any existing files when
759     * fileExist=Move is enabled, and an existing file exists.
760     * If this option copyAndDeleteOnRenameFails false, then an exception will be thrown if an existing file existed,
761     * if its true, then the existing file is deleted before the move operation.
762     */
763    public void setEagerDeleteTargetFile(boolean eagerDeleteTargetFile) {
764        this.eagerDeleteTargetFile = eagerDeleteTargetFile;
765    }
766
767    public GenericFileConfiguration getConfiguration() {
768        if (configuration == null) {
769            configuration = new GenericFileConfiguration();
770        }
771        return configuration;
772    }
773
774    public void setConfiguration(GenericFileConfiguration configuration) {
775        this.configuration = configuration;
776    }
777
778    public GenericFileExclusiveReadLockStrategy<T> getExclusiveReadLockStrategy() {
779        return exclusiveReadLockStrategy;
780    }
781
782    /**
783     * Pluggable read-lock as a org.apache.camel.component.file.GenericFileExclusiveReadLockStrategy implementation.
784     */
785    public void setExclusiveReadLockStrategy(GenericFileExclusiveReadLockStrategy<T> exclusiveReadLockStrategy) {
786        this.exclusiveReadLockStrategy = exclusiveReadLockStrategy;
787    }
788
789    public String getReadLock() {
790        return readLock;
791    }
792
793    /**
794     * Used by consumer, to only poll the files if it has exclusive read-lock on the file (i.e. the file is not in-progress or being written).
795     * Camel will wait until the file lock is granted.
796     * <p/>
797     * This option provides the build in strategies:
798     * <ul>
799     *     <li>none - No read lock is in use
800     *     <li>markerFile - Camel creates a marker file (fileName.camelLock) and then holds a lock on it. This option is not available for the FTP component
801     *     <li>changed - Changed is using file length/modification timestamp to detect whether the file is currently being copied or not. Will at least use 1 sec
802     *     to determine this, so this option cannot consume files as fast as the others, but can be more reliable as the JDK IO API cannot
803     *     always determine whether a file is currently being used by another process. The option readLockCheckInterval can be used to set the check frequency.</li>
804     *     <li>fileLock - is for using java.nio.channels.FileLock. This option is not avail for the FTP component. This approach should be avoided when accessing
805     *     a remote file system via a mount/share unless that file system supports distributed file locks.</li>
806     *     <li>rename - rename is for using a try to rename the file as a test if we can get exclusive read-lock.</li>
807     * </ul>
808     */
809    public void setReadLock(String readLock) {
810        this.readLock = readLock;
811    }
812
813    public long getReadLockCheckInterval() {
814        return readLockCheckInterval;
815    }
816
817    /**
818     * Interval in millis for the read-lock, if supported by the read lock.
819     * This interval is used for sleeping between attempts to acquire the read lock.
820     * For example when using the changed read lock, you can set a higher interval period to cater for slow writes.
821     * The default of 1 sec. may be too fast if the producer is very slow writing the file.
822     * <p/>
823     * Notice: For FTP the default readLockCheckInterval is 5000.
824     * <p/>
825     * The readLockTimeout value must be higher than readLockCheckInterval, but a rule of thumb is to have a timeout
826     * that is at least 2 or more times higher than the readLockCheckInterval. This is needed to ensure that amble
827     * time is allowed for the read lock process to try to grab the lock before the timeout was hit.
828     */
829    public void setReadLockCheckInterval(long readLockCheckInterval) {
830        this.readLockCheckInterval = readLockCheckInterval;
831    }
832
833    public long getReadLockTimeout() {
834        return readLockTimeout;
835    }
836
837    /**
838     * Optional timeout in millis for the read-lock, if supported by the read-lock.
839     * If the read-lock could not be granted and the timeout triggered, then Camel will skip the file.
840     * At next poll Camel, will try the file again, and this time maybe the read-lock could be granted.
841     * Use a value of 0 or lower to indicate forever. Currently fileLock, changed and rename support the timeout.
842     * <p/>
843     * Notice: For FTP the default readLockTimeout value is 20000 instead of 10000.
844     * <p/>
845     * The readLockTimeout value must be higher than readLockCheckInterval, but a rule of thumb is to have a timeout
846     * that is at least 2 or more times higher than the readLockCheckInterval. This is needed to ensure that amble
847     * time is allowed for the read lock process to try to grab the lock before the timeout was hit.
848     */
849    public void setReadLockTimeout(long readLockTimeout) {
850        this.readLockTimeout = readLockTimeout;
851    }
852
853    public boolean isReadLockMarkerFile() {
854        return readLockMarkerFile;
855    }
856
857    /**
858     * Whether to use marker file with the changed, rename, or exclusive read lock types.
859     * By default a marker file is used as well to guard against other processes picking up the same files.
860     * This behavior can be turned off by setting this option to false.
861     * For example if you do not want to write marker files to the file systems by the Camel application.
862     */
863    public void setReadLockMarkerFile(boolean readLockMarkerFile) {
864        this.readLockMarkerFile = readLockMarkerFile;
865    }
866
867    public LoggingLevel getReadLockLoggingLevel() {
868        return readLockLoggingLevel;
869    }
870
871    /**
872     * Logging level used when a read lock could not be acquired.
873     * By default a WARN is logged. You can change this level, for example to OFF to not have any logging.
874     * This option is only applicable for readLock of types: changed, fileLock, rename.
875     */
876    public void setReadLockLoggingLevel(LoggingLevel readLockLoggingLevel) {
877        this.readLockLoggingLevel = readLockLoggingLevel;
878    }
879
880    public long getReadLockMinLength() {
881        return readLockMinLength;
882    }
883
884    /**
885     * This option applied only for readLock=changed. This option allows you to configure a minimum file length.
886     * By default Camel expects the file to contain data, and thus the default value is 1.
887     * You can set this option to zero, to allow consuming zero-length files.
888     */
889    public void setReadLockMinLength(long readLockMinLength) {
890        this.readLockMinLength = readLockMinLength;
891    }
892
893    public long getReadLockMinAge() {
894        return readLockMinAge;
895    }
896
897    /**
898     * This option applied only for readLock=change.
899     * This options allows to specify a minimum age the file must be before attempting to acquire the read lock.
900     * For example use readLockMinAge=300s to require the file is at last 5 minutes old.
901     * This can speedup the changed read lock as it will only attempt to acquire files which are at least that given age.
902     */
903    public void setReadLockMinAge(long readLockMinAge) {
904        this.readLockMinAge = readLockMinAge;
905    }
906
907    public int getBufferSize() {
908        return bufferSize;
909    }
910
911    /**
912     * Write buffer sized in bytes.
913     */
914    public void setBufferSize(int bufferSize) {
915        if (bufferSize <= 0) {
916            throw new IllegalArgumentException("BufferSize must be a positive value, was " + bufferSize);
917        }
918        this.bufferSize = bufferSize;
919    }
920
921    public GenericFileExist getFileExist() {
922        return fileExist;
923    }
924
925    /**
926     * What to do if a file already exists with the same name.
927     *
928     * Override, which is the default, replaces the existing file.
929     * <ul>
930     *   <li>Append - adds content to the existing file.</li>
931     *   <li>Fail - throws a GenericFileOperationException, indicating that there is already an existing file.</li>
932     *   <li>Ignore - silently ignores the problem and does not override the existing file, but assumes everything is okay.</li>
933     *   <li>Move - option requires to use the moveExisting option to be configured as well.
934     *   The option eagerDeleteTargetFile can be used to control what to do if an moving the file, and there exists already an existing file,
935     *   otherwise causing the move operation to fail.
936     *   The Move option will move any existing files, before writing the target file.</li>
937     *   <li>TryRename Camel is only applicable if tempFileName option is in use. This allows to try renaming the file from the temporary name to the actual name,
938     *   without doing any exists check.This check may be faster on some file systems and especially FTP servers.</li>
939     * </ul>
940     */
941    public void setFileExist(GenericFileExist fileExist) {
942        this.fileExist = fileExist;
943    }
944
945    public boolean isAutoCreate() {
946        return autoCreate;
947    }
948
949    /**
950     * Automatically create missing directories in the file's pathname. For the file consumer, that means creating the starting directory.
951     * For the file producer, it means the directory the files should be written to.
952     */
953    public void setAutoCreate(boolean autoCreate) {
954        this.autoCreate = autoCreate;
955    }
956
957    public boolean isStartingDirectoryMustExist() {
958        return startingDirectoryMustExist;
959    }
960
961    /**
962     * Whether the starting directory must exist. Mind that the autoCreate option is default enabled,
963     * which means the starting directory is normally auto created if it doesn't exist.
964     * You can disable autoCreate and enable this to ensure the starting directory must exist. Will thrown an exception if the directory doesn't exist.
965     */
966    public void setStartingDirectoryMustExist(boolean startingDirectoryMustExist) {
967        this.startingDirectoryMustExist = startingDirectoryMustExist;
968    }
969
970    public boolean isDirectoryMustExist() {
971        return directoryMustExist;
972    }
973
974    /**
975     * Similar to startingDirectoryMustExist but this applies during polling recursive sub directories.
976     */
977    public void setDirectoryMustExist(boolean directoryMustExist) {
978        this.directoryMustExist = directoryMustExist;
979    }
980
981    public GenericFileProcessStrategy<T> getProcessStrategy() {
982        return processStrategy;
983    }
984
985    /**
986     * A pluggable org.apache.camel.component.file.GenericFileProcessStrategy allowing you to implement your own readLock option or similar.
987     * Can also be used when special conditions must be met before a file can be consumed, such as a special ready file exists.
988     * If this option is set then the readLock option does not apply.
989     */
990    public void setProcessStrategy(GenericFileProcessStrategy<T> processStrategy) {
991        this.processStrategy = processStrategy;
992    }
993
994    public String getLocalWorkDirectory() {
995        return localWorkDirectory;
996    }
997
998    /**
999     * When consuming, a local work directory can be used to store the remote file content directly in local files,
1000     * to avoid loading the content into memory. This is beneficial, if you consume a very big remote file and thus can conserve memory.
1001     */
1002    public void setLocalWorkDirectory(String localWorkDirectory) {
1003        this.localWorkDirectory = localWorkDirectory;
1004    }
1005
1006    public int getMaxMessagesPerPoll() {
1007        return maxMessagesPerPoll;
1008    }
1009
1010    /**
1011     * Tlo define a maximum messages to gather per poll.
1012     * By default no maximum is set. Can be used to set a limit of e.g. 1000 to avoid when starting up the server that there are thousands of files.
1013     * Set a value of 0 or negative to disabled it.
1014     * Notice: If this option is in use then the File and FTP components will limit before any sorting.
1015     * For example if you have 100000 files and use maxMessagesPerPoll=500, then only the first 500 files will be picked up, and then sorted.
1016     * You can use the eagerMaxMessagesPerPoll option and set this to false to allow to scan all files first and then sort afterwards.
1017     */
1018    public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
1019        this.maxMessagesPerPoll = maxMessagesPerPoll;
1020    }
1021
1022    public boolean isEagerMaxMessagesPerPoll() {
1023        return eagerMaxMessagesPerPoll;
1024    }
1025
1026    /**
1027     * Allows for controlling whether the limit from maxMessagesPerPoll is eager or not.
1028     * If eager then the limit is during the scanning of files. Where as false would scan all files, and then perform sorting.
1029     * Setting this option to false allows for sorting all files first, and then limit the poll. Mind that this requires a
1030     * higher memory usage as all file details are in memory to perform the sorting.
1031     */
1032    public void setEagerMaxMessagesPerPoll(boolean eagerMaxMessagesPerPoll) {
1033        this.eagerMaxMessagesPerPoll = eagerMaxMessagesPerPoll;
1034    }
1035
1036    public int getMaxDepth() {
1037        return maxDepth;
1038    }
1039
1040    /**
1041     * The maximum depth to traverse when recursively processing a directory.
1042     */
1043    public void setMaxDepth(int maxDepth) {
1044        this.maxDepth = maxDepth;
1045    }
1046
1047    public int getMinDepth() {
1048        return minDepth;
1049    }
1050
1051    /**
1052     * The minimum depth to start processing when recursively processing a directory.
1053     * Using minDepth=1 means the base directory. Using minDepth=2 means the first sub directory.
1054     */
1055    public void setMinDepth(int minDepth) {
1056        this.minDepth = minDepth;
1057    }
1058
1059    public IdempotentRepository<String> getInProgressRepository() {
1060        return inProgressRepository;
1061    }
1062
1063    /**
1064     * A pluggable in-progress repository org.apache.camel.spi.IdempotentRepository.
1065     * The in-progress repository is used to account the current in progress files being consumed. By default a memory based repository is used.
1066     */
1067    public void setInProgressRepository(IdempotentRepository<String> inProgressRepository) {
1068        this.inProgressRepository = inProgressRepository;
1069    }
1070
1071    public boolean isKeepLastModified() {
1072        return keepLastModified;
1073    }
1074
1075    /**
1076     * Will keep the last modified timestamp from the source file (if any).
1077     * Will use the Exchange.FILE_LAST_MODIFIED header to located the timestamp.
1078     * This header can contain either a java.util.Date or long with the timestamp.
1079     * If the timestamp exists and the option is enabled it will set this timestamp on the written file.
1080     * Note: This option only applies to the file producer. You cannot use this option with any of the ftp producers.
1081     */
1082    public void setKeepLastModified(boolean keepLastModified) {
1083        this.keepLastModified = keepLastModified;
1084    }
1085
1086    public boolean isAllowNullBody() {
1087        return allowNullBody;
1088    }
1089
1090    /**
1091     * Used to specify if a null body is allowed during file writing.
1092     * If set to true then an empty file will be created, when set to false, and attempting to send a null body to the file component,
1093     * a GenericFileWriteException of 'Cannot write null body to file.' will be thrown.
1094     * If the `fileExist` option is set to 'Override', then the file will be truncated, and if set to `append` the file will remain unchanged.
1095     */
1096    public void setAllowNullBody(boolean allowNullBody) {
1097        this.allowNullBody = allowNullBody;
1098    }
1099
1100    /**
1101     * Configures the given message with the file which sets the body to the
1102     * file object.
1103     */
1104    public void configureMessage(GenericFile<T> file, Message message) {
1105        message.setBody(file);
1106
1107        if (flatten) {
1108            // when flatten the file name should not contain any paths
1109            message.setHeader(Exchange.FILE_NAME, file.getFileNameOnly());
1110        } else {
1111            // compute name to set on header that should be relative to starting directory
1112            String name = file.isAbsolute() ? file.getAbsoluteFilePath() : file.getRelativeFilePath();
1113
1114            // skip leading endpoint configured directory
1115            String endpointPath = getConfiguration().getDirectory() + getFileSeparator();
1116
1117            // need to normalize paths to ensure we can match using startsWith
1118            endpointPath = FileUtil.normalizePath(endpointPath);
1119            String copyOfName = FileUtil.normalizePath(name);
1120            if (ObjectHelper.isNotEmpty(endpointPath) && copyOfName.startsWith(endpointPath)) {
1121                name = name.substring(endpointPath.length());
1122            }
1123
1124            // adjust filename
1125            message.setHeader(Exchange.FILE_NAME, name);
1126        }
1127    }
1128
1129    /**
1130     * Set up the exchange properties with the options of the file endpoint
1131     */
1132    public void configureExchange(Exchange exchange) {
1133        // Now we just set the charset property here
1134        if (getCharset() != null) {
1135            exchange.setProperty(Exchange.CHARSET_NAME, getCharset());
1136        }
1137    }
1138
1139    /**
1140     * Strategy to configure the move, preMove, or moveExisting option based on a String input.
1141     *
1142     * @param expression the original string input
1143     * @return configured string or the original if no modifications is needed
1144     */
1145    protected String configureMoveOrPreMoveExpression(String expression) {
1146        // if the expression already have ${ } placeholders then pass it unmodified
1147        if (StringHelper.hasStartToken(expression, "simple")) {
1148            return expression;
1149        }
1150
1151        // remove trailing slash
1152        expression = FileUtil.stripTrailingSeparator(expression);
1153
1154        StringBuilder sb = new StringBuilder();
1155
1156        // if relative then insert start with the parent folder
1157        if (!isAbsolute(expression)) {
1158            sb.append("${file:parent}");
1159            sb.append(getFileSeparator());
1160        }
1161        // insert the directory the end user provided
1162        sb.append(expression);
1163        // append only the filename (file:name can contain a relative path, so we must use onlyname)
1164        sb.append(getFileSeparator());
1165        sb.append("${file:onlyname}");
1166
1167        return sb.toString();
1168    }
1169
1170    protected Map<String, Object> getParamsAsMap() {
1171        Map<String, Object> params = new HashMap<String, Object>();
1172
1173        if (isNoop()) {
1174            params.put("noop", Boolean.toString(true));
1175        }
1176        if (isDelete()) {
1177            params.put("delete", Boolean.toString(true));
1178        }
1179        if (move != null) {
1180            params.put("move", move);
1181        }
1182        if (moveFailed != null) {
1183            params.put("moveFailed", moveFailed);
1184        }
1185        if (preMove != null) {
1186            params.put("preMove", preMove);
1187        }
1188        if (exclusiveReadLockStrategy != null) {
1189            params.put("exclusiveReadLockStrategy", exclusiveReadLockStrategy);
1190        }
1191        if (readLock != null) {
1192            params.put("readLock", readLock);
1193        }
1194        if (readLockCheckInterval > 0) {
1195            params.put("readLockCheckInterval", readLockCheckInterval);
1196        }
1197        if (readLockTimeout > 0) {
1198            params.put("readLockTimeout", readLockTimeout);
1199        }
1200        params.put("readLockMarkerFile", readLockMarkerFile);
1201        params.put("readLockMinLength", readLockMinLength);
1202        params.put("readLockLoggingLevel", readLockLoggingLevel);
1203        params.put("readLockMinAge", readLockMinAge);
1204
1205        return params;
1206    }
1207
1208    private Expression createFileLanguageExpression(String expression) {
1209        Language language;
1210        // only use file language if the name is complex (eg. using $)
1211        if (expression.contains("$")) {
1212            language = getCamelContext().resolveLanguage("file");
1213        } else {
1214            language = getCamelContext().resolveLanguage("constant");
1215        }
1216        return language.createExpression(expression);
1217    }
1218
1219    /**
1220     * Creates the associated name of the done file based on the given file name.
1221     * <p/>
1222     * This method should only be invoked if a done filename property has been set on this endpoint.
1223     *
1224     * @param fileName the file name
1225     * @return name of the associated done file name
1226     */
1227    protected String createDoneFileName(String fileName) {
1228        String pattern = getDoneFileName();
1229        ObjectHelper.notEmpty(pattern, "doneFileName", pattern);
1230
1231        // we only support ${file:name} or ${file:name.noext} as dynamic placeholders for done files
1232        String path = FileUtil.onlyPath(fileName);
1233        String onlyName = FileUtil.stripPath(fileName);
1234
1235        pattern = pattern.replaceFirst("\\$\\{file:name\\}", onlyName);
1236        pattern = pattern.replaceFirst("\\$simple\\{file:name\\}", onlyName);
1237        pattern = pattern.replaceFirst("\\$\\{file:name.noext\\}", FileUtil.stripExt(onlyName));
1238        pattern = pattern.replaceFirst("\\$simple\\{file:name.noext\\}", FileUtil.stripExt(onlyName));
1239
1240        // must be able to resolve all placeholders supported
1241        if (StringHelper.hasStartToken(pattern, "simple")) {
1242            throw new ExpressionIllegalSyntaxException(fileName + ". Cannot resolve reminder: " + pattern);
1243        }
1244
1245        String answer = pattern;
1246        if (ObjectHelper.isNotEmpty(path) && ObjectHelper.isNotEmpty(pattern)) {
1247            // done file must always be in same directory as the real file name
1248            answer = path + getFileSeparator() + pattern;
1249        }
1250
1251        if (getConfiguration().needToNormalize()) {
1252            // must normalize path to cater for Windows and other OS
1253            answer = FileUtil.normalizePath(answer);
1254        }
1255
1256        return answer;
1257    }
1258
1259    /**
1260     * Is the given file a done file?
1261     * <p/>
1262     * This method should only be invoked if a done filename property has been set on this endpoint.
1263     *
1264     * @param fileName the file name
1265     * @return <tt>true</tt> if its a done file, <tt>false</tt> otherwise
1266     */
1267    protected boolean isDoneFile(String fileName) {
1268        String pattern = getDoneFileName();
1269        ObjectHelper.notEmpty(pattern, "doneFileName", pattern);
1270
1271        if (!StringHelper.hasStartToken(pattern, "simple")) {
1272            // no tokens, so just match names directly
1273            return pattern.equals(fileName);
1274        }
1275
1276        // the static part of the pattern, is that a prefix or suffix?
1277        // its a prefix if ${ start token is not at the start of the pattern
1278        boolean prefix = pattern.indexOf("${") > 0;
1279
1280        // remove dynamic parts of the pattern so we only got the static part left
1281        pattern = pattern.replaceFirst("\\$\\{file:name\\}", "");
1282        pattern = pattern.replaceFirst("\\$simple\\{file:name\\}", "");
1283        pattern = pattern.replaceFirst("\\$\\{file:name.noext\\}", "");
1284        pattern = pattern.replaceFirst("\\$simple\\{file:name.noext\\}", "");
1285
1286        // must be able to resolve all placeholders supported
1287        if (StringHelper.hasStartToken(pattern, "simple")) {
1288            throw new ExpressionIllegalSyntaxException(fileName + ". Cannot resolve reminder: " + pattern);
1289        }
1290
1291        if (prefix) {
1292            return fileName.startsWith(pattern);
1293        } else {
1294            return fileName.endsWith(pattern);
1295        }
1296    }
1297
1298    @Override
1299    protected void doStart() throws Exception {
1300        // validate that the read lock options is valid for the process strategy
1301        if (!"none".equals(readLock) && !"off".equals(readLock)) {
1302            if (readLockTimeout > 0 && readLockMinAge > 0 && readLockTimeout <= readLockCheckInterval + readLockMinAge) {
1303                throw new IllegalArgumentException("The option readLockTimeout must be higher than readLockCheckInterval + readLockMinAge"
1304                    + ", was readLockTimeout=" + readLockTimeout + ", readLockCheckInterval+readLockMinAge=" + (readLockCheckInterval + readLockMinAge)
1305                    + ". A good practice is to let the readLockTimeout be at least readLockMinAge + 2 times the readLockCheckInterval"
1306                    + " to ensure that the read lock procedure has enough time to acquire the lock.");
1307            }
1308            if (readLockTimeout > 0 && readLockTimeout <= readLockCheckInterval) {
1309                throw new IllegalArgumentException("The option readLockTimeout must be higher than readLockCheckInterval"
1310                        + ", was readLockTimeout=" + readLockTimeout + ", readLockCheckInterval=" + readLockCheckInterval
1311                        + ". A good practice is to let the readLockTimeout be at least 3 times higher than the readLockCheckInterval"
1312                        + " to ensure that the read lock procedure has enough time to acquire the lock.");
1313            }
1314        }
1315
1316        ServiceHelper.startServices(inProgressRepository, idempotentRepository);
1317        super.doStart();
1318    }
1319
1320    @Override
1321    protected void doStop() throws Exception {
1322        super.doStop();
1323        ServiceHelper.stopServices(inProgressRepository, idempotentRepository);
1324    }
1325}