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.File; 020import java.io.FileNotFoundException; 021import java.nio.file.Files; 022import java.nio.file.attribute.PosixFilePermission; 023import java.util.Arrays; 024import java.util.HashSet; 025import java.util.Set; 026 027import org.apache.camel.Component; 028import org.apache.camel.Exchange; 029import org.apache.camel.Message; 030import org.apache.camel.PollingConsumer; 031import org.apache.camel.Processor; 032import org.apache.camel.component.file.strategy.FileMoveExistingStrategy; 033import org.apache.camel.processor.idempotent.MemoryIdempotentRepository; 034import org.apache.camel.spi.Metadata; 035import org.apache.camel.spi.UriEndpoint; 036import org.apache.camel.spi.UriParam; 037import org.apache.camel.spi.UriPath; 038import org.apache.camel.util.FileUtil; 039import org.apache.camel.util.ObjectHelper; 040 041/** 042 * The file component is used for reading or writing files. 043 */ 044@UriEndpoint(firstVersion = "1.0.0", scheme = "file", title = "File", syntax = "file:directoryName", consumerClass = FileConsumer.class, label = "core,file") 045public class FileEndpoint extends GenericFileEndpoint<File> { 046 047 private static final Integer CHMOD_WRITE_MASK = 02; 048 private static final Integer CHMOD_READ_MASK = 04; 049 private static final Integer CHMOD_EXECUTE_MASK = 01; 050 051 private final FileOperations operations = new FileOperations(this); 052 053 @UriPath(name = "directoryName") @Metadata(required = "true") 054 private File file; 055 @UriParam(label = "advanced", defaultValue = "true") 056 private boolean copyAndDeleteOnRenameFail = true; 057 @UriParam(label = "advanced") 058 private boolean renameUsingCopy; 059 @UriParam(label = "producer,advanced", defaultValue = "true") 060 private boolean forceWrites = true; 061 @UriParam(label = "consumer,advanced") 062 private boolean probeContentType; 063 @UriParam(label = "consumer,advanced") 064 private String extendedAttributes; 065 @UriParam(label = "producer,advanced") 066 private String chmod; 067 @UriParam(label = "producer,advanced") 068 private String chmodDirectory; 069 070 public FileEndpoint() { 071 } 072 073 public FileEndpoint(String endpointUri, Component component) { 074 super(endpointUri, component); 075 } 076 077 public FileConsumer createConsumer(Processor processor) throws Exception { 078 ObjectHelper.notNull(operations, "operations"); 079 ObjectHelper.notNull(file, "file"); 080 081 // auto create starting directory if needed 082 if (!file.exists() && !file.isDirectory()) { 083 if (isAutoCreate()) { 084 log.debug("Creating non existing starting directory: {}", file); 085 boolean absolute = FileUtil.isAbsolute(file); 086 boolean created = operations.buildDirectory(file.getPath(), absolute); 087 if (!created) { 088 log.warn("Cannot auto create starting directory: {}", file); 089 } 090 } else if (isStartingDirectoryMustExist()) { 091 throw new FileNotFoundException("Starting directory does not exist: " + file); 092 } 093 } 094 095 FileConsumer result = newFileConsumer(processor, operations); 096 097 if (isDelete() && getMove() != null) { 098 throw new IllegalArgumentException("You cannot set both delete=true and move options"); 099 } 100 101 // if noop=true then idempotent should also be configured 102 if (isNoop() && !isIdempotentSet()) { 103 log.info("Endpoint is configured with noop=true so forcing endpoint to be idempotent as well"); 104 setIdempotent(true); 105 } 106 107 // if idempotent and no repository set then create a default one 108 if (isIdempotentSet() && isIdempotent() && idempotentRepository == null) { 109 log.info("Using default memory based idempotent repository with cache max size: {}", DEFAULT_IDEMPOTENT_CACHE_SIZE); 110 idempotentRepository = MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_IDEMPOTENT_CACHE_SIZE); 111 } 112 113 if (ObjectHelper.isNotEmpty(getReadLock())) { 114 // check if its a valid 115 String valid = "none,markerFile,fileLock,rename,changed,idempotent,idempotent-changed,idempotent-rename"; 116 String[] arr = valid.split(","); 117 boolean matched = Arrays.stream(arr).anyMatch(n -> n.equals(getReadLock())); 118 if (!matched) { 119 throw new IllegalArgumentException("ReadLock invalid: " + getReadLock() + ", must be one of: " + valid); 120 } 121 } 122 123 // set max messages per poll 124 result.setMaxMessagesPerPoll(getMaxMessagesPerPoll()); 125 result.setEagerLimitMaxMessagesPerPoll(isEagerMaxMessagesPerPoll()); 126 127 configureConsumer(result); 128 return result; 129 } 130 131 @Override 132 public PollingConsumer createPollingConsumer() throws Exception { 133 ObjectHelper.notNull(operations, "operations"); 134 ObjectHelper.notNull(file, "file"); 135 136 if (log.isDebugEnabled()) { 137 log.debug("Creating GenericFilePollingConsumer with queueSize: {} blockWhenFull: {} blockTimeout: {}", 138 getPollingConsumerQueueSize(), isPollingConsumerBlockWhenFull(), getPollingConsumerBlockTimeout()); 139 } 140 GenericFilePollingConsumer result = new GenericFilePollingConsumer(this); 141 // should not call configurePollingConsumer when its GenericFilePollingConsumer 142 result.setBlockWhenFull(isPollingConsumerBlockWhenFull()); 143 result.setBlockTimeout(getPollingConsumerBlockTimeout()); 144 145 return result; 146 } 147 148 public GenericFileProducer<File> createProducer() throws Exception { 149 ObjectHelper.notNull(operations, "operations"); 150 151 // you cannot use temp file and file exists append 152 if (getFileExist() == GenericFileExist.Append && ((getTempPrefix() != null) || (getTempFileName() != null))) { 153 throw new IllegalArgumentException("You cannot set both fileExist=Append and tempPrefix/tempFileName options"); 154 } 155 156 // ensure fileExist and moveExisting is configured correctly if in use 157 if (getFileExist() == GenericFileExist.Move && getMoveExisting() == null) { 158 throw new IllegalArgumentException("You must configure moveExisting option when fileExist=Move"); 159 } else if (getMoveExisting() != null && getFileExist() != GenericFileExist.Move) { 160 throw new IllegalArgumentException("You must configure fileExist=Move when moveExisting has been set"); 161 } 162 if (this.getMoveExistingFileStrategy() == null) { 163 this.setMoveExistingFileStrategy(createDefaultMoveExistingFileStrategy()); 164 } 165 return new GenericFileProducer<>(this, operations); 166 } 167 168 public Exchange createExchange(GenericFile<File> file) { 169 Exchange exchange = createExchange(); 170 if (file != null) { 171 file.bindToExchange(exchange, probeContentType); 172 } 173 return exchange; 174 } 175 176 /** 177 * Strategy to create a new {@link FileConsumer} 178 * 179 * @param processor the given processor 180 * @param operations file operations 181 * @return the created consumer 182 */ 183 protected FileConsumer newFileConsumer(Processor processor, GenericFileOperations<File> operations) { 184 return new FileConsumer(this, processor, operations, processStrategy != null ? processStrategy : createGenericFileStrategy()); 185 } 186 187 /** 188 * Default Existing File Move Strategy 189 * @return the default implementation for file component 190 */ 191 private FileMoveExistingStrategy createDefaultMoveExistingFileStrategy() { 192 return new GenericFileDefaultMoveExistingFileStrategy(); 193 } 194 195 public File getFile() { 196 return file; 197 } 198 199 /** 200 * The starting directory 201 */ 202 public void setFile(File file) { 203 this.file = file; 204 // update configuration as well 205 getConfiguration().setDirectory(FileUtil.isAbsolute(file) ? file.getAbsolutePath() : file.getPath()); 206 } 207 208 @Override 209 public String getScheme() { 210 return "file"; 211 } 212 213 @Override 214 protected String createEndpointUri() { 215 return getFile().toURI().toString(); 216 } 217 218 @Override 219 public char getFileSeparator() { 220 return File.separatorChar; 221 } 222 223 @Override 224 public boolean isAbsolute(String name) { 225 // relative or absolute path? 226 return FileUtil.isAbsolute(new File(name)); 227 } 228 229 public boolean isCopyAndDeleteOnRenameFail() { 230 return copyAndDeleteOnRenameFail; 231 } 232 233 /** 234 * Whether to fallback and do a copy and delete file, in case the file could not be renamed directly. This option is not available for the FTP component. 235 */ 236 public void setCopyAndDeleteOnRenameFail(boolean copyAndDeleteOnRenameFail) { 237 this.copyAndDeleteOnRenameFail = copyAndDeleteOnRenameFail; 238 } 239 240 public boolean isRenameUsingCopy() { 241 return renameUsingCopy; 242 } 243 244 /** 245 * Perform rename operations using a copy and delete strategy. 246 * This is primarily used in environments where the regular rename operation is unreliable (e.g. across different file systems or networks). 247 * This option takes precedence over the copyAndDeleteOnRenameFail parameter that will automatically fall back to the copy and delete strategy, 248 * but only after additional delays. 249 */ 250 public void setRenameUsingCopy(boolean renameUsingCopy) { 251 this.renameUsingCopy = renameUsingCopy; 252 } 253 254 public boolean isForceWrites() { 255 return forceWrites; 256 } 257 258 /** 259 * Whether to force syncing writes to the file system. 260 * You can turn this off if you do not want this level of guarantee, for example if writing to logs / audit logs etc; this would yield better performance. 261 */ 262 public void setForceWrites(boolean forceWrites) { 263 this.forceWrites = forceWrites; 264 } 265 266 public boolean isProbeContentType() { 267 return probeContentType; 268 } 269 270 /** 271 * Whether to enable probing of the content type. If enable then the consumer uses {@link Files#probeContentType(java.nio.file.Path)} to 272 * determine the content-type of the file, and store that as a header with key {@link Exchange#FILE_CONTENT_TYPE} on the {@link Message}. 273 */ 274 public void setProbeContentType(boolean probeContentType) { 275 this.probeContentType = probeContentType; 276 } 277 278 public String getExtendedAttributes() { 279 return extendedAttributes; 280 } 281 282 /** 283 * To define which file attributes of interest. Like posix:permissions,posix:owner,basic:lastAccessTime, 284 * it supports basic wildcard like posix:*, basic:lastAccessTime 285 */ 286 public void setExtendedAttributes(String extendedAttributes) { 287 this.extendedAttributes = extendedAttributes; 288 } 289 290 /** 291 * Chmod value must be between 000 and 777; If there is a leading digit like in 0755 we will ignore it. 292 */ 293 public boolean chmodPermissionsAreValid(String chmod) { 294 if (chmod == null || chmod.length() < 3 || chmod.length() > 4) { 295 return false; 296 } 297 String permissionsString = chmod.trim().substring(chmod.length() - 3); // if 4 digits chop off leading one 298 for (int i = 0; i < permissionsString.length(); i++) { 299 Character c = permissionsString.charAt(i); 300 if (!Character.isDigit(c) || Integer.parseInt(c.toString()) > 7) { 301 return false; 302 } 303 } 304 return true; 305 } 306 307 public Set<PosixFilePermission> getPermissions() { 308 Set<PosixFilePermission> permissions = new HashSet<>(); 309 if (ObjectHelper.isEmpty(chmod)) { 310 return permissions; 311 } 312 313 String chmodString = chmod.substring(chmod.length() - 3); // if 4 digits chop off leading one 314 315 Integer ownerValue = Integer.parseInt(chmodString.substring(0, 1)); 316 Integer groupValue = Integer.parseInt(chmodString.substring(1, 2)); 317 Integer othersValue = Integer.parseInt(chmodString.substring(2, 3)); 318 319 if ((ownerValue & CHMOD_WRITE_MASK) > 0) { 320 permissions.add(PosixFilePermission.OWNER_WRITE); 321 } 322 if ((ownerValue & CHMOD_READ_MASK) > 0) { 323 permissions.add(PosixFilePermission.OWNER_READ); 324 } 325 if ((ownerValue & CHMOD_EXECUTE_MASK) > 0) { 326 permissions.add(PosixFilePermission.OWNER_EXECUTE); 327 } 328 329 if ((groupValue & CHMOD_WRITE_MASK) > 0) { 330 permissions.add(PosixFilePermission.GROUP_WRITE); 331 } 332 if ((groupValue & CHMOD_READ_MASK) > 0) { 333 permissions.add(PosixFilePermission.GROUP_READ); 334 } 335 if ((groupValue & CHMOD_EXECUTE_MASK) > 0) { 336 permissions.add(PosixFilePermission.GROUP_EXECUTE); 337 } 338 339 if ((othersValue & CHMOD_WRITE_MASK) > 0) { 340 permissions.add(PosixFilePermission.OTHERS_WRITE); 341 } 342 if ((othersValue & CHMOD_READ_MASK) > 0) { 343 permissions.add(PosixFilePermission.OTHERS_READ); 344 } 345 if ((othersValue & CHMOD_EXECUTE_MASK) > 0) { 346 permissions.add(PosixFilePermission.OTHERS_EXECUTE); 347 } 348 349 return permissions; 350 } 351 352 public String getChmod() { 353 return chmod; 354 } 355 356 /** 357 * Specify the file permissions which is sent by the producer, the chmod value must be between 000 and 777; 358 * If there is a leading digit like in 0755 we will ignore it. 359 */ 360 public void setChmod(String chmod) throws Exception { 361 if (ObjectHelper.isNotEmpty(chmod) && chmodPermissionsAreValid(chmod)) { 362 this.chmod = chmod.trim(); 363 } else { 364 throw new IllegalArgumentException("chmod option [" + chmod + "] is not valid"); 365 } 366 } 367 368 public Set<PosixFilePermission> getDirectoryPermissions() { 369 Set<PosixFilePermission> permissions = new HashSet<>(); 370 if (ObjectHelper.isEmpty(chmodDirectory)) { 371 return permissions; 372 } 373 374 String chmodString = chmodDirectory.substring(chmodDirectory.length() - 3); // if 4 digits chop off leading one 375 376 Integer ownerValue = Integer.parseInt(chmodString.substring(0, 1)); 377 Integer groupValue = Integer.parseInt(chmodString.substring(1, 2)); 378 Integer othersValue = Integer.parseInt(chmodString.substring(2, 3)); 379 380 if ((ownerValue & CHMOD_WRITE_MASK) > 0) { 381 permissions.add(PosixFilePermission.OWNER_WRITE); 382 } 383 if ((ownerValue & CHMOD_READ_MASK) > 0) { 384 permissions.add(PosixFilePermission.OWNER_READ); 385 } 386 if ((ownerValue & CHMOD_EXECUTE_MASK) > 0) { 387 permissions.add(PosixFilePermission.OWNER_EXECUTE); 388 } 389 390 if ((groupValue & CHMOD_WRITE_MASK) > 0) { 391 permissions.add(PosixFilePermission.GROUP_WRITE); 392 } 393 if ((groupValue & CHMOD_READ_MASK) > 0) { 394 permissions.add(PosixFilePermission.GROUP_READ); 395 } 396 if ((groupValue & CHMOD_EXECUTE_MASK) > 0) { 397 permissions.add(PosixFilePermission.GROUP_EXECUTE); 398 } 399 400 if ((othersValue & CHMOD_WRITE_MASK) > 0) { 401 permissions.add(PosixFilePermission.OTHERS_WRITE); 402 } 403 if ((othersValue & CHMOD_READ_MASK) > 0) { 404 permissions.add(PosixFilePermission.OTHERS_READ); 405 } 406 if ((othersValue & CHMOD_EXECUTE_MASK) > 0) { 407 permissions.add(PosixFilePermission.OTHERS_EXECUTE); 408 } 409 410 return permissions; 411 } 412 413 public String getChmodDirectory() { 414 return chmodDirectory; 415 } 416 417 /** 418 * Specify the directory permissions used when the producer creates missing directories, the chmod value must be between 000 and 777; 419 * If there is a leading digit like in 0755 we will ignore it. 420 */ 421 public void setChmodDirectory(String chmodDirectory) throws Exception { 422 if (ObjectHelper.isNotEmpty(chmodDirectory) && chmodPermissionsAreValid(chmodDirectory)) { 423 this.chmodDirectory = chmodDirectory.trim(); 424 } else { 425 throw new IllegalArgumentException("chmodDirectory option [" + chmodDirectory + "] is not valid"); 426 } 427 } 428 429}