CustomProcessorUploadHandler implements FileEventHandler { @Override public void created (Path path) { File createdFile = path.toFile(); LOG.info("Created called with " + createdFile); boolean succeeded = false; try { if (createdFile.getName().endsWith(".tar")) { LOG.info("Processing file at " + path); CustomProcessorInfo customProcessorInfo = this.getCustomProcessorInfo(createdFile); if (customProcessorInfo == null) { LOG.warn("No information found for CustomProcessorRuntime in " + createdFile); return; } InputStream jarFile = this.getJarFile(createdFile); if (jarFile == null) { LOG.warn("No jar file found for CustomProcessorRuntime in " + createdFile); return; } this.catalogService.addCustomProcessorInfoAsBundle(customProcessorInfo, jarFile); succeeded = true; } else { LOG.info("Failing unsupported file that was received: " + path); } } catch (IOException e) { LOG.warn("Exception occured while processing tar file: " + createdFile, e); } catch (ComponentConfigException e) { LOG.warn("UI specification for custom processor incorrect for custom processor file: " + createdFile, e); } catch (NoSuchAlgorithmException e) { LOG.warn("Got NoSuchAlgorithmException while calculating digest for jar: " + createdFile, e); } finally { try { if (succeeded) { LOG.info("CustomProcessorRuntime uploaded successfully from " + createdFile + " Moving file to " + uploadSuccessPath); moveFileToSuccessDirectory(createdFile); } else { LOG.warn("CustomProcessorRuntime failed to upload from " + createdFile + " Moving file to " + uploadFailPath); moveFileToFailDirectory(createdFile); } } catch (IOException e1) { LOG.warn("Error moving " + createdFile.getAbsolutePath() + " to " + (succeeded ? uploadSuccessPath : uploadFailPath), e1); } } } CustomProcessorUploadHandler(String watchPath, String uploadFailPath, String uploadSuccessPath, StreamCatalogService catalogService); @Override String getDirectoryToWatch(); @Override void created(Path path); @Override void deleted(Path path); @Override void modified(Path path);  }
@Test public void testNotTarFile () throws IOException { File f = new File(notTarFilePath); f.createNewFile(); Path path = Paths.get(notTarFilePath); this.customProcessorUploadHandler.created(path); f = new File(failedUploadMoveDirectory + File.separator + notTarFileName); Assert.assertTrue(f.exists()); }
@Test public void testFailures () throws IOException { final String[] fileNames = {"nocustomprocessorinfo.tar", "nojarfile.tar", "nocustomprocessorimpl.tar"}; for (String fileName: fileNames) { URL url = classLoader.getResource(resourceDirectoryPrefix + fileName); String consoleCustomProcessorTarString = url.getFile(); File consoleCustomProcessorTar = new File(consoleCustomProcessorTarString); FileUtils.copyFileToDirectory(consoleCustomProcessorTar, new File(uploadWatchDirectory), false); this.customProcessorUploadHandler.created(Paths.get(uploadWatchDirectory).resolve(fileName)); File f = new File(failedUploadMoveDirectory + File.separator + fileName); Assert.assertTrue(f.exists()); } }
@Test public void testSuccessfulUpload () throws IOException, ComponentConfigException, NoSuchAlgorithmException { String fileName = "consolecustomprocessor.tar"; URL url = classLoader.getResource(resourceDirectoryPrefix + fileName); String consoleCustomProcessorTarString = url.getFile(); File consoleCustomProcessorTar = new File(consoleCustomProcessorTarString); FileUtils.copyFileToDirectory(consoleCustomProcessorTar, new File(uploadWatchDirectory), false); this.customProcessorUploadHandler.created(Paths.get(uploadWatchDirectory).resolve(fileName)); new VerificationsInOrder() {{ InputStream jarFileActual; CustomProcessorInfo actual; catalogService.addCustomProcessorInfoAsBundle(actual = withCapture(), jarFileActual = withCapture()); times = 1; Assert.assertTrue(actual.getName().equals(customProcessorInfo.getName())); Assert.assertTrue(IOUtils.contentEquals(jarFileActual, jarFile)); }}; }