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()); }  @Override void configure(); @Override PipeRunResult doPipe(Message message, IPipeLineSession session); @IbisDoc({"base directory where files are unzipped to", ""}) void setDirectory(String string); String getDirectory(); @IbisDoc({"the session key that contains the base directory where files are unzipped to", "destination"}) void setDirectorySessionKey(String string); String getDirectorySessionKey(); String getSessionKey(); @IbisDoc({"name of the key in the <code>pipelinesession</code> which contains the inputstream", "file"}) void setSessionKey(String string);  }
@Test public void testNullSessionKey() throws Exception { exception.expect(PipeRunException.class); configureAndStartPipe(); doPipe(pipe, "das", session); fail("this is expected to fail"); }