Periodic { @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.countdown('name',statement,repeat-rate-in-seconds) submit a repeatedly-called background statement until it returns 0") public Stream<JobInfo> countdown(@Name("name") String name, @Name("statement") String statement, @Name("rate") long rate) { validateQuery(statement); JobInfo info = submit(name, new Countdown(name, statement, rate, log), log); info.rate = rate; return Stream.of(info); }  @Procedure @Description("apoc.periodic.list - list all jobs") Stream<JobInfo> list(); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.commit(statement,params) - runs the given statement in separate transactions until it returns 0") Stream<RundownResult> commit(@Name("statement") String statement, @Name(value = "params", defaultValue = "{}") Map<String,Object> parameters); @Procedure @Description("apoc.periodic.cancel(name) - cancel job with the given name") Stream<JobInfo> cancel(@Name("name") String name); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.submit('name',statement) - submit a one-off background statement") Stream<JobInfo> submit(@Name("name") String name, @Name("statement") String statement); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.repeat('name',statement,repeat-rate-in-seconds, config) submit a repeatedly-called background statement. Fourth parameter 'config' is optional and can contain 'params' entry for nested statement.") Stream<JobInfo> repeat(@Name("name") String name, @Name("statement") String statement, @Name("rate") long rate, @Name(value = "config", defaultValue = "{}") Map<String,Object> config ); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.countdown('name',statement,repeat-rate-in-seconds) submit a repeatedly-called background statement until it returns 0") Stream<JobInfo> countdown(@Name("name") String name, @Name("statement") String statement, @Name("rate") long rate); JobInfo submit(String name, Runnable task, Log log); JobInfo schedule(String name, Runnable task, long delay, long repeat); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.iterate('statement returning items', 'statement per item', {batchSize:1000,iterateList:true,parallel:false,params:{},concurrency:50,retries:0}) YIELD batches, total - run the second statement for each item returned by the first statement. Returns number of batches and total processed rows") Stream<BatchAndTotalResult> iterate(
            @Name("cypherIterate") String cypherIterate,
            @Name("cypherAction") String cypherAction,
            @Name("config") Map<String,Object> config); static final Pattern RUNTIME_PATTERN; static final Pattern CYPHER_PREFIX_PATTERN; static final String CYPHER_RUNTIME_SLOTTED; @Context public GraphDatabaseService db; @Context public TerminationGuard terminationGuard; @Context public Log log; @Context public Pools pools; @Context public Transaction tx; }
@Test public void testCountdown() { int startValue = 3; int rate = 1; db.executeTransactionally("CREATE (counter:Counter {c: $startValue})", Collections.singletonMap("startValue", startValue)); String statementToRepeat = "MATCH (counter:Counter) SET counter.c = counter.c - 1 RETURN counter.c as count"; Map<String, Object> params = map("statement", statementToRepeat, "rate", rate); testResult(db, "CALL apoc.periodic.countdown('decrement', $statement, $rate)", params, r -> { try { Thread.sleep(startValue * rate * 1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } long count = TestUtil.singleResultFirstColumn(db, "MATCH (counter:Counter) RETURN counter.c as c"); assertEquals(0L, count); }); }