@Test public void testStoreTask() { JobStore store = getJobStore(); Job<BatchJobExt> job = createBatchJobObject(); store.init().await(); store.storeJob(job).await(); Pair<List<Job<?>>, Integer> jobsAndErrors = store.retrieveJobs().toBlocking().first(); assertThat(jobsAndErrors.getLeft().get(0)).isEqualTo(job); Task task = createTaskObject(job); store.storeTask(task).await(); Task retrievedTask = store.retrieveTask(task.getId()).toBlocking().first(); assertThat(task).isEqualTo(retrievedTask); }
@Override public Completable storeTask(Task task) { return Observable.fromCallable((Callable<Statement>) () -> { String jobId = task.getJobId(); String taskId = task.getId(); checkIfJobIsActive(jobId); String taskJsonString = ObjectMappers.writeValueAsString(mapper, task); Statement taskStatement = insertActiveTaskStatement.bind(taskId, taskJsonString); Statement taskIdStatement = insertActiveTaskIdStatement.bind(jobId, taskId); BatchStatement batchStatement = new BatchStatement(); batchStatement.add(taskStatement); batchStatement.add(taskIdStatement); transactionLogger().logBeforeCreate(insertActiveTaskStatement, "storeTask", task); return batchStatement; }).flatMap(statement -> execute(statement).doOnNext(rs -> transactionLogger().logAfterCreate(insertActiveTaskStatement, "storeTask", task)) ).toCompletable(); }
CassandraJobStore implements JobStore { @Override public Completable storeTask(Task task) { return Observable.fromCallable((Callable<Statement>) () -> { String jobId = task.getJobId(); String taskId = task.getId(); checkIfJobIsActive(jobId); String taskJsonString = ObjectMappers.writeValueAsString(mapper, task); Statement taskStatement = insertActiveTaskStatement.bind(taskId, taskJsonString); Statement taskIdStatement = insertActiveTaskIdStatement.bind(jobId, taskId); BatchStatement batchStatement = new BatchStatement(); batchStatement.add(taskStatement); batchStatement.add(taskIdStatement); transactionLogger().logBeforeCreate(insertActiveTaskStatement, "storeTask", task); return batchStatement; }).flatMap(statement -> execute(statement).doOnNext(rs -> transactionLogger().logAfterCreate(insertActiveTaskStatement, "storeTask", task)) ).toCompletable(); } }
CassandraJobStore implements JobStore { @Override public Completable storeTask(Task task) { return Observable.fromCallable((Callable<Statement>) () -> { String jobId = task.getJobId(); String taskId = task.getId(); checkIfJobIsActive(jobId); String taskJsonString = ObjectMappers.writeValueAsString(mapper, task); Statement taskStatement = insertActiveTaskStatement.bind(taskId, taskJsonString); Statement taskIdStatement = insertActiveTaskIdStatement.bind(jobId, taskId); BatchStatement batchStatement = new BatchStatement(); batchStatement.add(taskStatement); batchStatement.add(taskIdStatement); transactionLogger().logBeforeCreate(insertActiveTaskStatement, "storeTask", task); return batchStatement; }).flatMap(statement -> execute(statement).doOnNext(rs -> transactionLogger().logAfterCreate(insertActiveTaskStatement, "storeTask", task)) ).toCompletable(); } @Inject  CassandraJobStore(CassandraStoreConfiguration configuration,
                             Session session,
                             TitusRuntime titusRuntime);   CassandraJobStore(CassandraStoreConfiguration configuration,
                      Session session,
                      TitusRuntime titusRuntime,
                      ObjectMapper mapper,
                      int initialBucketCount,
                      int maxBucketSize); }
CassandraJobStore implements JobStore { @Override public Completable storeTask(Task task) { return Observable.fromCallable((Callable<Statement>) () -> { String jobId = task.getJobId(); String taskId = task.getId(); checkIfJobIsActive(jobId); String taskJsonString = ObjectMappers.writeValueAsString(mapper, task); Statement taskStatement = insertActiveTaskStatement.bind(taskId, taskJsonString); Statement taskIdStatement = insertActiveTaskIdStatement.bind(jobId, taskId); BatchStatement batchStatement = new BatchStatement(); batchStatement.add(taskStatement); batchStatement.add(taskIdStatement); transactionLogger().logBeforeCreate(insertActiveTaskStatement, "storeTask", task); return batchStatement; }).flatMap(statement -> execute(statement).doOnNext(rs -> transactionLogger().logAfterCreate(insertActiveTaskStatement, "storeTask", task)) ).toCompletable(); } @Inject  CassandraJobStore(CassandraStoreConfiguration configuration,
                             Session session,
                             TitusRuntime titusRuntime);   CassandraJobStore(CassandraStoreConfiguration configuration,
                      Session session,
                      TitusRuntime titusRuntime,
                      ObjectMapper mapper,
                      int initialBucketCount,
                      int maxBucketSize); @Override Completable init(); @Override Observable<Pair<List<Job<?>>, Integer>> retrieveJobs(); @Override Observable<Job<?>> retrieveJob(String jobId); @Override Completable storeJob(Job job); @Override Completable updateJob(Job job); @Override Completable deleteJob(Job job); @Override Observable<Pair<List<Task>, Integer>> retrieveTasksForJob(String jobId); @Override Observable<Task> retrieveTask(String taskId); @Override Completable storeTask(Task task); @Override Completable updateTask(Task task); @Override Completable replaceTask(Task oldTask, Task newTask); @Override Completable moveTask(Job jobFrom, Job jobTo, Task taskAfter); @Override Completable deleteTask(Task task); @Override Observable<Job<?>> retrieveArchivedJob(String jobId); @Override Observable<Task> retrieveArchivedTasksForJob(String jobId); @Override Observable<Task> retrieveArchivedTask(String taskId); }
CassandraJobStore implements JobStore { @Override public Completable storeTask(Task task) { return Observable.fromCallable((Callable<Statement>) () -> { String jobId = task.getJobId(); String taskId = task.getId(); checkIfJobIsActive(jobId); String taskJsonString = ObjectMappers.writeValueAsString(mapper, task); Statement taskStatement = insertActiveTaskStatement.bind(taskId, taskJsonString); Statement taskIdStatement = insertActiveTaskIdStatement.bind(jobId, taskId); BatchStatement batchStatement = new BatchStatement(); batchStatement.add(taskStatement); batchStatement.add(taskIdStatement); transactionLogger().logBeforeCreate(insertActiveTaskStatement, "storeTask", task); return batchStatement; }).flatMap(statement -> execute(statement).doOnNext(rs -> transactionLogger().logAfterCreate(insertActiveTaskStatement, "storeTask", task)) ).toCompletable(); } @Inject  CassandraJobStore(CassandraStoreConfiguration configuration,
                             Session session,
                             TitusRuntime titusRuntime);   CassandraJobStore(CassandraStoreConfiguration configuration,
                      Session session,
                      TitusRuntime titusRuntime,
                      ObjectMapper mapper,
                      int initialBucketCount,
                      int maxBucketSize); @Override Completable init(); @Override Observable<Pair<List<Job<?>>, Integer>> retrieveJobs(); @Override Observable<Job<?>> retrieveJob(String jobId); @Override Completable storeJob(Job job); @Override Completable updateJob(Job job); @Override Completable deleteJob(Job job); @Override Observable<Pair<List<Task>, Integer>> retrieveTasksForJob(String jobId); @Override Observable<Task> retrieveTask(String taskId); @Override Completable storeTask(Task task); @Override Completable updateTask(Task task); @Override Completable replaceTask(Task oldTask, Task newTask); @Override Completable moveTask(Job jobFrom, Job jobTo, Task taskAfter); @Override Completable deleteTask(Task task); @Override Observable<Job<?>> retrieveArchivedJob(String jobId); @Override Observable<Task> retrieveArchivedTasksForJob(String jobId); @Override Observable<Task> retrieveArchivedTask(String taskId);  }