JdbcProjectImpl implements ProjectLoader { @Override public void uploadFlowFile(final int projectId, final int projectVersion, final File flowFile, final int flowVersion) throws ProjectManagerException { logger.info(String .format( "Uploading flow file %s, version %d for project %d, version %d, file length is [%d bytes]", flowFile.getName(), flowVersion, projectId, projectVersion, flowFile.length())); if (flowFile.length() > MAX_FLOW_FILE_SIZE_IN_BYTES) { throw new ProjectManagerException("Flow file length exceeds 10 MB limit."); } final byte[] buffer = new byte[MAX_FLOW_FILE_SIZE_IN_BYTES]; final String INSERT_FLOW_FILES = "INSERT INTO project_flow_files (project_id, project_version, flow_name, flow_version, " + "modified_time, " + "flow_file) values (?,?,?,?,?,?)"; try (final FileInputStream input = new FileInputStream(flowFile); final BufferedInputStream bufferedStream = new BufferedInputStream(input)) { final int size = bufferedStream.read(buffer); logger.info("Read bytes for " + flowFile.getName() + ", size:" + size); final byte[] buf = Arrays.copyOfRange(buffer, 0, size); try { this.dbOperator .update(INSERT_FLOW_FILES, projectId, projectVersion, flowFile.getName(), flowVersion, System.currentTimeMillis(), buf); } catch (final SQLException e) { throw new ProjectManagerException( "Error uploading flow file " + flowFile.getName() + ", version " + flowVersion + ".", e); } } catch (final IOException e) { throw new ProjectManagerException( String.format( "Error reading flow file %s, version: %d, length: [%d bytes].", flowFile.getName(), flowVersion, flowFile.length())); } } @Inject  JdbcProjectImpl(final Props props, final DatabaseOperator databaseOperator); @Override List<Project> fetchAllActiveProjects(); @Override Project fetchProjectById(final int id); @Override Project fetchProjectByName(final String name); @Override synchronized Project createNewProject(final String name, final String description,
      final User creator); @Override void uploadProjectFile(final int projectId, final int version, final File localFile,
      final String uploader, final String uploaderIPAddr); @Override void addProjectVersion(final int projectId, final int version, final File localFile,
      final File startupDependencies, final String uploader, final byte[] md5,
      final String resourceId, final String uploaderIPAddr); @Override ProjectFileHandler fetchProjectMetaData(final int projectId, final int version); @Override ProjectFileHandler getUploadedFile(final int projectId, final int version); @Override void changeProjectVersion(final Project project, final int version, final String user); @Override void updatePermission(final Project project, final String name, final Permission perm,
      final boolean isGroup); @Override void updateProjectSettings(final Project project); @Override void removePermission(final Project project, final String name, final boolean isGroup); @Override void removeProject(final Project project, final String user); @Override boolean postEvent(final Project project, final EventType type, final String user,
      final String message); @Override List<ProjectLogEvent> getProjectEvents(final Project project, final int num,
      final int skip); @Override void updateDescription(final Project project, final String description, final String user); @Override int getLatestProjectVersion(final Project project); @Override void uploadFlows(final Project project, final int version, final Collection<Flow> flows); @Override void uploadFlow(final Project project, final int version, final Flow flow); @Override void updateFlow(final Project project, final int version, final Flow flow); @Override Flow fetchFlow(final Project project, final String flowId); @Override List<Flow> fetchAllProjectFlows(final Project project); @Override Map<Project, List<Flow>> fetchAllFlowsForProjects(final List<Project> projects); @Override void uploadProjectProperties(final Project project, final List<Props> properties); @Override void uploadProjectProperty(final Project project, final Props props); @Override void updateProjectProperty(final Project project, final Props props); @Override Props fetchProjectProperty(final int projectId, final int projectVer,
      final String propsName); @Override Props fetchProjectProperty(final Project project, final String propsName); @Override Map<String, Props> fetchProjectProperties(final int projectId, final int version); @Override void cleanOlderProjectVersion(final int projectId, final int version,
      final List<Integer> excludedVersions); @Override void uploadFlowFile(final int projectId, final int projectVersion, final File flowFile,
      final int flowVersion); @Override File getUploadedFlowFile(final int projectId, final int projectVersion,
      final String flowFileName, final int flowVersion, final File tempDir); @Override int getLatestFlowVersion(final int projectId, final int projectVersion,
      final String flowName); @Override boolean isFlowFileUploaded(final int projectId, final int projectVersion); @Override List<Project> fetchProjectById(final List<Integer> ids);  }
@Test public void testUploadFlowFile() throws Exception { final File testYamlFile = ExecutionsTestUtil.getFlowFile(BASIC_FLOW_YAML_DIR, BASIC_FLOW_FILE); this.loader.uploadFlowFile(PROJECT_ID, PROJECT_VERSION, testYamlFile, FLOW_VERSION); final File tempDir = Files.createTempDir(); tempDir.deleteOnExit(); final File file = this.loader .getUploadedFlowFile(PROJECT_ID, PROJECT_VERSION, BASIC_FLOW_FILE, FLOW_VERSION, tempDir); assertThat(file.getName()).isEqualTo(BASIC_FLOW_FILE); assertThat(FileUtils.contentEquals(testYamlFile, file)).isTrue(); }
@Test public void testDuplicateUploadFlowFileException() throws Exception { final File testYamlFile = ExecutionsTestUtil.getFlowFile(BASIC_FLOW_YAML_DIR, BASIC_FLOW_FILE); this.loader.uploadFlowFile(PROJECT_ID, PROJECT_VERSION, testYamlFile, FLOW_VERSION); assertThatThrownBy( () -> this.loader.uploadFlowFile(PROJECT_ID, PROJECT_VERSION, testYamlFile, FLOW_VERSION)) .isInstanceOf(ProjectManagerException.class) .hasMessageContaining( "Error uploading flow file " + BASIC_FLOW_FILE + ", version " + FLOW_VERSION + "."); }
@Test public void testUploadLargeFlowFileException() throws Exception { final File testYamlFile = ExecutionsTestUtil.getFlowFile(LARGE_FLOW_YAML_DIR, LARGE_FLOW_FILE); assertThatThrownBy( () -> this.loader.uploadFlowFile(PROJECT_ID, PROJECT_VERSION, testYamlFile, FLOW_VERSION)) .isInstanceOf(ProjectManagerException.class) .hasMessageContaining( "Flow file length exceeds 10 MB limit."); }