class UploadFilePipe extends FixedForwardPipe {

	@Override
	public PipeRunResult doPipe(Message message, IPipeLineSession session) throws PipeRunException {
		InputStream inputStream = (InputStream) session.get(getSessionKey());
		if (inputStream == null) {
			throw new PipeRunException(this, getLogPrefix(session) + "got null value from session under key [" + getSessionKey() + "]");
		}

		File dir;
		if (StringUtils.isNotEmpty(getDirectory())) {
			dir = new File(getDirectory());
		} else {
			if (StringUtils.isNotEmpty(getDirectorySessionKey())) {
				dir = new File((String) session.get(getDirectorySessionKey()));
			} else {
				String filename;
				try {
					filename = message.asString();
				} catch (IOException e) {
					throw new PipeRunException(this, getLogPrefix(session)+"cannot open stream", e);
				}
				dir = new File(filename);
			}
		}

		if (!dir.exists()) {
			if (dir.mkdirs()) {
				log.debug(getLogPrefix(session) + "created directory [" + dir.getPath() + "]");
			} else {
				log.warn(getLogPrefix(session) + "directory [" + dir.getPath() + "] could not be created");
			}
		}
		
		String fileName;
		try {
			fileName = (String) session.get("fileName");
			if (FileUtils.extensionEqualsIgnoreCase(fileName, "zip")) {
				FileUtils.unzipStream(inputStream, dir);
			} else {
				throw new PipeRunException(this, getLogPrefix(session) + "file extension [" + FileUtils.getFileNameExtension(fileName) + "] should be 'zip'");
			}
		} catch (IOException e) {
			throw new PipeRunException(this, getLogPrefix(session) + " Exception on uploading and unzipping/writing file", e);
		}

		return new PipeRunResult(getForward(), dir.getPath());
	}

}