@Test public void insertJobs_memoryOnlyJob_doesNotWriteToDatabase() { JobDatabase database = noopDatabase(); FastJobStorage subject = new FastJobStorage(database, new DirectExecutor()); subject.insertJobs(DataSetMemory.FULL_SPECS); verify(database, times(0)).insertJobs(DataSet1.FULL_SPECS); }
@Override public synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs) { List<FullSpec> durable = Stream.of(fullSpecs).filterNot(FullSpec::isMemoryOnly).toList(); if (durable.size() > 0) { serialExecutor.execute(() -> { jobDatabase.insertJobs(durable); }); } for (FullSpec fullSpec : fullSpecs) { jobs.add(fullSpec.getJobSpec()); constraintsByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getConstraintSpecs()); dependenciesByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getDependencySpecs()); } }
FastJobStorage implements JobStorage { @Override public synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs) { List<FullSpec> durable = Stream.of(fullSpecs).filterNot(FullSpec::isMemoryOnly).toList(); if (durable.size() > 0) { serialExecutor.execute(() -> { jobDatabase.insertJobs(durable); }); } for (FullSpec fullSpec : fullSpecs) { jobs.add(fullSpec.getJobSpec()); constraintsByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getConstraintSpecs()); dependenciesByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getDependencySpecs()); } } }
FastJobStorage implements JobStorage { @Override public synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs) { List<FullSpec> durable = Stream.of(fullSpecs).filterNot(FullSpec::isMemoryOnly).toList(); if (durable.size() > 0) { serialExecutor.execute(() -> { jobDatabase.insertJobs(durable); }); } for (FullSpec fullSpec : fullSpecs) { jobs.add(fullSpec.getJobSpec()); constraintsByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getConstraintSpecs()); dependenciesByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getDependencySpecs()); } } FastJobStorage(@NonNull JobDatabase jobDatabase, @NonNull Executor serialExecutor); }
FastJobStorage implements JobStorage { @Override public synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs) { List<FullSpec> durable = Stream.of(fullSpecs).filterNot(FullSpec::isMemoryOnly).toList(); if (durable.size() > 0) { serialExecutor.execute(() -> { jobDatabase.insertJobs(durable); }); } for (FullSpec fullSpec : fullSpecs) { jobs.add(fullSpec.getJobSpec()); constraintsByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getConstraintSpecs()); dependenciesByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getDependencySpecs()); } } FastJobStorage(@NonNull JobDatabase jobDatabase, @NonNull Executor serialExecutor); @Override synchronized void init(); @Override synchronized void flush(); @Override synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs); @Override synchronized @Nullable JobSpec getJobSpec(@NonNull String id); @Override synchronized @NonNull List<JobSpec> getAllJobSpecs(); @Override synchronized @NonNull List<JobSpec> getPendingJobsWithNoDependenciesInCreatedOrder(long currentTime); @Override synchronized @NonNull List<JobSpec> getJobsInQueue(@NonNull String queue); @Override synchronized int getJobInstanceCount(@NonNull String factoryKey); @Override synchronized void updateJobRunningState(@NonNull String id, boolean isRunning); @Override synchronized void updateJobAfterRetry(@NonNull String id, boolean isRunning, int runAttempt, long nextRunAttemptTime, @NonNull String serializedData); @Override synchronized void updateAllJobsToBePending(); @Override void updateJobs(@NonNull List<JobSpec> jobSpecs); @Override synchronized void deleteJob(@NonNull String jobId); @Override synchronized void deleteJobs(@NonNull List<String> jobIds); @Override synchronized @NonNull List<ConstraintSpec> getConstraintSpecs(@NonNull String jobId); @Override synchronized @NonNull List<ConstraintSpec> getAllConstraintSpecs(); @Override synchronized @NonNull List<DependencySpec> getDependencySpecsThatDependOnJob(@NonNull String jobSpecId); @Override @NonNull List<DependencySpec> getAllDependencySpecs(); }
FastJobStorage implements JobStorage { @Override public synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs) { List<FullSpec> durable = Stream.of(fullSpecs).filterNot(FullSpec::isMemoryOnly).toList(); if (durable.size() > 0) { serialExecutor.execute(() -> { jobDatabase.insertJobs(durable); }); } for (FullSpec fullSpec : fullSpecs) { jobs.add(fullSpec.getJobSpec()); constraintsByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getConstraintSpecs()); dependenciesByJobId.put(fullSpec.getJobSpec().getId(), fullSpec.getDependencySpecs()); } } FastJobStorage(@NonNull JobDatabase jobDatabase, @NonNull Executor serialExecutor); @Override synchronized void init(); @Override synchronized void flush(); @Override synchronized void insertJobs(@NonNull List<FullSpec> fullSpecs); @Override synchronized @Nullable JobSpec getJobSpec(@NonNull String id); @Override synchronized @NonNull List<JobSpec> getAllJobSpecs(); @Override synchronized @NonNull List<JobSpec> getPendingJobsWithNoDependenciesInCreatedOrder(long currentTime); @Override synchronized @NonNull List<JobSpec> getJobsInQueue(@NonNull String queue); @Override synchronized int getJobInstanceCount(@NonNull String factoryKey); @Override synchronized void updateJobRunningState(@NonNull String id, boolean isRunning); @Override synchronized void updateJobAfterRetry(@NonNull String id, boolean isRunning, int runAttempt, long nextRunAttemptTime, @NonNull String serializedData); @Override synchronized void updateAllJobsToBePending(); @Override void updateJobs(@NonNull List<JobSpec> jobSpecs); @Override synchronized void deleteJob(@NonNull String jobId); @Override synchronized void deleteJobs(@NonNull List<String> jobIds); @Override synchronized @NonNull List<ConstraintSpec> getConstraintSpecs(@NonNull String jobId); @Override synchronized @NonNull List<ConstraintSpec> getAllConstraintSpecs(); @Override synchronized @NonNull List<DependencySpec> getDependencySpecsThatDependOnJob(@NonNull String jobSpecId); @Override @NonNull List<DependencySpec> getAllDependencySpecs();  }