@Test(expected = RuntimeException.class) public void testPersistJobOutputThrowsIfJobDoesNotExist() throws IOException { final Path jobsDir = createTmpDir(FilesystemJobsDAOTest.class); final FilesystemJobsDAO dao = createStandardFilesystemDAO(jobsDir); dao.persistOutput(generateJobId(), generateRandomJobOutput()); }
@Override public void persistOutput(JobId jobId, JobOutput jobOutput) { final Optional<Path> maybeJobDir = resolveJobDir(jobId); if (!maybeJobDir.isPresent()) throw new RuntimeException(jobOutput.getId() + ": cannot be persisted to job " + jobId + ": job dir does not exist"); final Path outputsDir = maybeJobDir.get().resolve(JOB_DIR_OUTPUTS_DIRNAME); createIfDoesNotExist(outputsDir); final Path outputPath = outputsDir.resolve(jobOutput.getId().toString()); writeJobOutputToDisk(jobOutput, outputPath); final Optional<Path> maybeJobOutputsFile = resolveJobFile(jobId, JOB_DIR_OUTPUTS_FILENAME); try { final List<JobOutputDetails> existingJobOutputMetadata = maybeJobOutputsFile.isPresent() ? loadJSON(maybeJobOutputsFile.get(), new TypeReference<List<JobOutputDetails>>(){}) : new ArrayList<>(); final JobOutputDetails jobOutputDetails = new JobOutputDetails( jobOutput.getId(), jobOutput.getData().getSizeOf(), Optional.of(jobOutput.getData().getMimeType()), jobOutput.getName(), jobOutput.getDescription(), jobOutput.getMetadata()); existingJobOutputMetadata.add(jobOutputDetails); writeJSON(resolveJobDir(jobId).get().resolve(JOB_DIR_OUTPUTS_FILENAME), existingJobOutputMetadata); } catch (IOException ex) { throw new RuntimeException(ex); } }
FilesystemJobsDAO implements JobDAO { @Override public void persistOutput(JobId jobId, JobOutput jobOutput) { final Optional<Path> maybeJobDir = resolveJobDir(jobId); if (!maybeJobDir.isPresent()) throw new RuntimeException(jobOutput.getId() + ": cannot be persisted to job " + jobId + ": job dir does not exist"); final Path outputsDir = maybeJobDir.get().resolve(JOB_DIR_OUTPUTS_DIRNAME); createIfDoesNotExist(outputsDir); final Path outputPath = outputsDir.resolve(jobOutput.getId().toString()); writeJobOutputToDisk(jobOutput, outputPath); final Optional<Path> maybeJobOutputsFile = resolveJobFile(jobId, JOB_DIR_OUTPUTS_FILENAME); try { final List<JobOutputDetails> existingJobOutputMetadata = maybeJobOutputsFile.isPresent() ? loadJSON(maybeJobOutputsFile.get(), new TypeReference<List<JobOutputDetails>>(){}) : new ArrayList<>(); final JobOutputDetails jobOutputDetails = new JobOutputDetails( jobOutput.getId(), jobOutput.getData().getSizeOf(), Optional.of(jobOutput.getData().getMimeType()), jobOutput.getName(), jobOutput.getDescription(), jobOutput.getMetadata()); existingJobOutputMetadata.add(jobOutputDetails); writeJSON(resolveJobDir(jobId).get().resolve(JOB_DIR_OUTPUTS_FILENAME), existingJobOutputMetadata); } catch (IOException ex) { throw new RuntimeException(ex); } } }
FilesystemJobsDAO implements JobDAO { @Override public void persistOutput(JobId jobId, JobOutput jobOutput) { final Optional<Path> maybeJobDir = resolveJobDir(jobId); if (!maybeJobDir.isPresent()) throw new RuntimeException(jobOutput.getId() + ": cannot be persisted to job " + jobId + ": job dir does not exist"); final Path outputsDir = maybeJobDir.get().resolve(JOB_DIR_OUTPUTS_DIRNAME); createIfDoesNotExist(outputsDir); final Path outputPath = outputsDir.resolve(jobOutput.getId().toString()); writeJobOutputToDisk(jobOutput, outputPath); final Optional<Path> maybeJobOutputsFile = resolveJobFile(jobId, JOB_DIR_OUTPUTS_FILENAME); try { final List<JobOutputDetails> existingJobOutputMetadata = maybeJobOutputsFile.isPresent() ? loadJSON(maybeJobOutputsFile.get(), new TypeReference<List<JobOutputDetails>>(){}) : new ArrayList<>(); final JobOutputDetails jobOutputDetails = new JobOutputDetails( jobOutput.getId(), jobOutput.getData().getSizeOf(), Optional.of(jobOutput.getData().getMimeType()), jobOutput.getName(), jobOutput.getDescription(), jobOutput.getMetadata()); existingJobOutputMetadata.add(jobOutputDetails); writeJSON(resolveJobDir(jobId).get().resolve(JOB_DIR_OUTPUTS_FILENAME), existingJobOutputMetadata); } catch (IOException ex) { throw new RuntimeException(ex); } } FilesystemJobsDAO(Path jobsDirectory, IdGenerator idGenerator); }
FilesystemJobsDAO implements JobDAO { @Override public void persistOutput(JobId jobId, JobOutput jobOutput) { final Optional<Path> maybeJobDir = resolveJobDir(jobId); if (!maybeJobDir.isPresent()) throw new RuntimeException(jobOutput.getId() + ": cannot be persisted to job " + jobId + ": job dir does not exist"); final Path outputsDir = maybeJobDir.get().resolve(JOB_DIR_OUTPUTS_DIRNAME); createIfDoesNotExist(outputsDir); final Path outputPath = outputsDir.resolve(jobOutput.getId().toString()); writeJobOutputToDisk(jobOutput, outputPath); final Optional<Path> maybeJobOutputsFile = resolveJobFile(jobId, JOB_DIR_OUTPUTS_FILENAME); try { final List<JobOutputDetails> existingJobOutputMetadata = maybeJobOutputsFile.isPresent() ? loadJSON(maybeJobOutputsFile.get(), new TypeReference<List<JobOutputDetails>>(){}) : new ArrayList<>(); final JobOutputDetails jobOutputDetails = new JobOutputDetails( jobOutput.getId(), jobOutput.getData().getSizeOf(), Optional.of(jobOutput.getData().getMimeType()), jobOutput.getName(), jobOutput.getDescription(), jobOutput.getMetadata()); existingJobOutputMetadata.add(jobOutputDetails); writeJSON(resolveJobDir(jobId).get().resolve(JOB_DIR_OUTPUTS_FILENAME), existingJobOutputMetadata); } catch (IOException ex) { throw new RuntimeException(ex); } } FilesystemJobsDAO(Path jobsDirectory, IdGenerator idGenerator); @Override boolean jobExists(JobId jobId); @Override List<JobDetails> getJobs(int pageSize, int page); @Override List<JobDetails> getJobs(int pageSize, int pageNumber, String query); @Override Disposable appendStdout(JobId jobId, Observable<byte[]> stdout); @Override Disposable appendStderr(JobId jobId, Observable<byte[]> stderr); @Override PersistedJob persist(ValidJobRequest validJobRequest); @Override void addNewJobStatus(JobId jobId, JobStatus newStatus, String statusMessage); @Override void persistOutput(JobId jobId, JobOutput jobOutput); @Override void remove(JobId jobId); @Override Optional<JobDetails> getJobDetailsById(JobId jobId); @Override Optional<JobSpec> getSpecJobWasSubmittedAgainst(JobId jobId); @Override boolean hasStdout(JobId jobId); @Override Optional<BinaryData> getStdout(JobId jobId); @Override boolean hasStderr(JobId jobId); @Override Optional<BinaryData> getStderr(JobId jobId); @Override Set<JobId> getJobsWithStatus(JobStatus status); @Override boolean hasOutput(JobId jobId, JobOutputId outputId); @Override Optional<BinaryData> getOutput(JobId jobId, JobOutputId outputId); @Override List<JobOutputDetails> getJobOutputs(JobId jobId); @Override boolean hasJobInputs(JobId jobId); @Override Optional<Map<JobExpectedInputId, JsonNode>> getJobInputs(JobId jobId); @Override Map<String, HealthCheck> getHealthChecks(); }
FilesystemJobsDAO implements JobDAO { @Override public void persistOutput(JobId jobId, JobOutput jobOutput) { final Optional<Path> maybeJobDir = resolveJobDir(jobId); if (!maybeJobDir.isPresent()) throw new RuntimeException(jobOutput.getId() + ": cannot be persisted to job " + jobId + ": job dir does not exist"); final Path outputsDir = maybeJobDir.get().resolve(JOB_DIR_OUTPUTS_DIRNAME); createIfDoesNotExist(outputsDir); final Path outputPath = outputsDir.resolve(jobOutput.getId().toString()); writeJobOutputToDisk(jobOutput, outputPath); final Optional<Path> maybeJobOutputsFile = resolveJobFile(jobId, JOB_DIR_OUTPUTS_FILENAME); try { final List<JobOutputDetails> existingJobOutputMetadata = maybeJobOutputsFile.isPresent() ? loadJSON(maybeJobOutputsFile.get(), new TypeReference<List<JobOutputDetails>>(){}) : new ArrayList<>(); final JobOutputDetails jobOutputDetails = new JobOutputDetails( jobOutput.getId(), jobOutput.getData().getSizeOf(), Optional.of(jobOutput.getData().getMimeType()), jobOutput.getName(), jobOutput.getDescription(), jobOutput.getMetadata()); existingJobOutputMetadata.add(jobOutputDetails); writeJSON(resolveJobDir(jobId).get().resolve(JOB_DIR_OUTPUTS_FILENAME), existingJobOutputMetadata); } catch (IOException ex) { throw new RuntimeException(ex); } } FilesystemJobsDAO(Path jobsDirectory, IdGenerator idGenerator); @Override boolean jobExists(JobId jobId); @Override List<JobDetails> getJobs(int pageSize, int page); @Override List<JobDetails> getJobs(int pageSize, int pageNumber, String query); @Override Disposable appendStdout(JobId jobId, Observable<byte[]> stdout); @Override Disposable appendStderr(JobId jobId, Observable<byte[]> stderr); @Override PersistedJob persist(ValidJobRequest validJobRequest); @Override void addNewJobStatus(JobId jobId, JobStatus newStatus, String statusMessage); @Override void persistOutput(JobId jobId, JobOutput jobOutput); @Override void remove(JobId jobId); @Override Optional<JobDetails> getJobDetailsById(JobId jobId); @Override Optional<JobSpec> getSpecJobWasSubmittedAgainst(JobId jobId); @Override boolean hasStdout(JobId jobId); @Override Optional<BinaryData> getStdout(JobId jobId); @Override boolean hasStderr(JobId jobId); @Override Optional<BinaryData> getStderr(JobId jobId); @Override Set<JobId> getJobsWithStatus(JobStatus status); @Override boolean hasOutput(JobId jobId, JobOutputId outputId); @Override Optional<BinaryData> getOutput(JobId jobId, JobOutputId outputId); @Override List<JobOutputDetails> getJobOutputs(JobId jobId); @Override boolean hasJobInputs(JobId jobId); @Override Optional<Map<JobExpectedInputId, JsonNode>> getJobInputs(JobId jobId); @Override Map<String, HealthCheck> getHealthChecks();  }