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 */
017 package org.apache.camel.component.file;
018
019 import org.apache.camel.Exchange;
020 import org.apache.camel.impl.LoggingExceptionHandler;
021 import org.apache.camel.spi.ExceptionHandler;
022 import org.apache.camel.spi.Synchronization;
023 import org.apache.commons.logging.Log;
024 import org.apache.commons.logging.LogFactory;
025
026 /**
027 * On completion strategy that performs the nessasary work after the {@link Exchange} has been processed.
028 * <p/>
029 * The work is for example to move the processed file into a backup folder, delete the file or
030 * in case of processing failure do a rollback.
031 *
032 * @version $Revision: 790297 $
033 */
034 public class GenericFileOnCompletion<T> implements Synchronization {
035
036 private final transient Log log = LogFactory.getLog(GenericFileOnCompletion.class);
037 private GenericFileEndpoint<T> endpoint;
038 private GenericFileOperations<T> operations;
039 private ExceptionHandler exceptionHandler;
040
041 public GenericFileOnCompletion(GenericFileEndpoint<T> endpoint, GenericFileOperations<T> operations) {
042 this.endpoint = endpoint;
043 this.operations = operations;
044 }
045
046 @SuppressWarnings("unchecked")
047 public void onComplete(Exchange exchange) {
048 onCompletion(exchange);
049 }
050
051 @SuppressWarnings("unchecked")
052 public void onFailure(Exchange exchange) {
053 onCompletion(exchange);
054 }
055
056 public ExceptionHandler getExceptionHandler() {
057 if (exceptionHandler == null) {
058 exceptionHandler = new LoggingExceptionHandler(getClass());
059 }
060 return exceptionHandler;
061 }
062
063 public void setExceptionHandler(ExceptionHandler exceptionHandler) {
064 this.exceptionHandler = exceptionHandler;
065 }
066
067 protected void onCompletion(Exchange exchange) {
068 GenericFileProcessStrategy<T> processStrategy = endpoint.getGenericFileProcessStrategy();
069
070 // after processing
071 final GenericFile<T> file = (GenericFile<T>) exchange.getProperty(FileComponent.FILE_EXCHANGE_FILE);
072 boolean failed = exchange.isFailed();
073
074 if (log.isDebugEnabled()) {
075 log.debug("Done processing file: " + file + " using exchange: " + exchange);
076 }
077
078 // commit or rollback
079 boolean committed = false;
080 try {
081 if (!failed) {
082 // commit the file strategy if there was no failure or already handled by the DeadLetterChannel
083 processStrategyCommit(processStrategy, exchange, file);
084 committed = true;
085 } else {
086 if (exchange.getException() != null) {
087 // if the failure was an exception then handle it
088 handleException(exchange.getException());
089 }
090 }
091 } finally {
092 if (!committed) {
093 processStrategyRollback(processStrategy, exchange, file);
094 }
095
096 // remove file from the in progress list as its no longer in progress
097 endpoint.getInProgressRepository().remove(file.getFileName());
098 }
099 }
100
101 /**
102 * Strategy when the file was processed and a commit should be executed.
103 *
104 * @param processStrategy the strategy to perform the commit
105 * @param exchange the exchange
106 * @param file the file processed
107 */
108 @SuppressWarnings("unchecked")
109 protected void processStrategyCommit(GenericFileProcessStrategy<T> processStrategy,
110 Exchange exchange, GenericFile<T> file) {
111 if (endpoint.isIdempotent()) {
112 // only add to idempotent repository if we could process the file
113 // only use the filename as the key as the file could be moved into a done folder
114 endpoint.getIdempotentRepository().add(file.getFileName());
115 }
116
117 try {
118 if (log.isTraceEnabled()) {
119 log.trace("Committing remote file strategy: " + processStrategy + " for file: " + file);
120 }
121 processStrategy.commit(operations, endpoint, exchange, file);
122 } catch (Exception e) {
123 handleException(e);
124 }
125 }
126
127 /**
128 * Strategy when the file was not processed and a rollback should be executed.
129 *
130 * @param processStrategy the strategy to perform the commit
131 * @param exchange the exchange
132 * @param file the file processed
133 */
134 protected void processStrategyRollback(GenericFileProcessStrategy<T> processStrategy,
135 Exchange exchange, GenericFile<T> file) {
136 if (log.isWarnEnabled()) {
137 log.warn("Rolling back remote file strategy: " + processStrategy + " for file: " + file);
138 }
139 try {
140 processStrategy.rollback(operations, endpoint, exchange, file);
141 } catch (Exception e) {
142 handleException(e);
143 }
144 }
145
146 protected void handleException(Throwable t) {
147 Throwable newt = (t == null) ? new IllegalArgumentException("Handling [null] exception") : t;
148 getExceptionHandler().handleException(newt);
149 }
150
151 @Override
152 public String toString() {
153 return "GenericFileOnCompletion";
154 }
155 }