class CleanupOldFilesPipe extends FixedForwardPipe {

	@Override
	public PipeRunResult doPipe(Message message, IPipeLineSession session) throws PipeRunException {
		try {
			String filename;
			if (StringUtils.isNotEmpty(getFilePattern())) {
				filename = FileUtils.getFilename(getParameterList(), session, "", getFilePattern());
			} else {
				if (StringUtils.isNotEmpty(getFilePatternSessionKey())) {
					filename = FileUtils.getFilename(getParameterList(), session, "", (String)session.get(getFilePatternSessionKey()));
				} else {
					if (StringUtils.isEmpty(message.asString())) {
						throw new PipeRunException(this, "input empty, but should contain filename to delete");
					} else {
						File in = new File(message.asString());
						filename = in.getName();
					}
				}
			}
			
			List delFiles = getFilesForDeletion(filename);
			if (delFiles != null && delFiles.size() > 0) {
				for (Iterator fileIt = delFiles.iterator(); fileIt.hasNext();) {
					File file = (File)fileIt.next();
					String curfilename=file.getName();
					if (file.delete()) {
						log.info(getLogPrefix(session)+"deleted file ["+file.getAbsolutePath()+"]");
					} else {
						log.warn(getLogPrefix(session)+"could not delete file ["+file.getAbsolutePath()+"]");
					}
				}
			} else {
				log.info(getLogPrefix(session)+"no files match pattern ["+filename+"]");
			}

			if (isDeleteEmptySubdirectories()) {
				File file = new File(filename);
				if (file.exists()) {
					deleteEmptySubdirectories(getLogPrefix(session), file, 0);
				}
			}
			
			return new PipeRunResult(getForward(), message);
		}
		catch(Exception e) {
			throw new PipeRunException(this, "Error while deleting file(s)", e); 
		}
	}

}