backport reactor#1746 Add VirtualTimeScheduler#getScheduledTaskCount()

Since VirtualTimeScheduler already has "counter" that keeps track of the
number of the scheduled tasks, it can be used to check whether tasks
have executed on the scheduler.

This change exposes that counter as "scheduled task count".

(cherry picked from commit e532799)
(backported for the benefit of reactor#1916 backport)

reactor-test/src/main/java/reactor/test/scheduler/VirtualTimeScheduler.java

@@ -230,6 +230,20 @@ public void advanceTimeTo(Instant instant) {
 		advanceTime(targetTime - nanoTime);
 	}
 
+	/**
+	 * Get the number of scheduled tasks.
+	 * <p>
+	 * This count includes tasks that have already performed as well as ones scheduled in future.
+	 * For periodical task, initial task is first scheduled and counted as one. Whenever
+	 * subsequent repeat happens this count gets incremented for the one that is scheduled
+	 * for the next run.
+	 *
+	 * @return number of tasks that have scheduled on this scheduler.
+	 */
+	public long getScheduledTaskCount() {
+		return this.counter;
+	}
+
 	@Override
 	public VirtualTimeWorker createWorker() {
 		if (shutdown) {

reactor-test/src/test/java/reactor/test/scheduler/VirtualTimeSchedulerTests.java

@@ -250,6 +250,71 @@ public void racingAdvanceTimeOnVaryingQueue() {
 		}
 	}
 
+	@Test
+	public void scheduledTaskCount() {
+		VirtualTimeScheduler vts = VirtualTimeScheduler.create();
+		assertThat(vts.getScheduledTaskCount()).as("initial value").isEqualTo(0);
+
+		vts.schedule(() -> {
+		});
+		assertThat(vts.getScheduledTaskCount()).as("a task scheduled").isEqualTo(1);
+	}
+
+	@Test
+	public void scheduledTaskCountWithInitialDelay() {
+		// schedule with delay
+		VirtualTimeScheduler vts = VirtualTimeScheduler.create();
+		vts.schedule(() -> {
+		}, 10, TimeUnit.DAYS);
+		assertThat(vts.getScheduledTaskCount()).as("scheduled in future").isEqualTo(1);
+
+		vts.advanceTimeBy(Duration.ofDays(11));
+		assertThat(vts.getScheduledTaskCount()).as("time advanced").isEqualTo(1);
+	}
+
+	@Test
+	public void scheduledTaskCountWithNoInitialDelay() {
+		// schedulePeriodically with no initial delay
+		VirtualTimeScheduler vts = VirtualTimeScheduler.create();
+		vts.schedulePeriodically(() -> {
+		}, 0, 5, TimeUnit.DAYS);
+
+		assertThat(vts.getScheduledTaskCount())
+			.as("initial delay task performed and scheduled for the first periodical task")
+			.isEqualTo(2);
+
+		vts.advanceTimeBy(Duration.ofDays(5));
+		assertThat(vts.getScheduledTaskCount())
+			.as("scheduled for the second periodical task")
+			.isEqualTo(3);
+	}
+
+	@Test
+	public void scheduledTaskCountBySchedulePeriodically() {
+		// schedulePeriodically with initial delay
+		VirtualTimeScheduler vts = VirtualTimeScheduler.create();
+		vts.schedulePeriodically(() -> {
+		}, 10, 5, TimeUnit.DAYS);
+		assertThat(vts.getScheduledTaskCount())
+			.as("scheduled for initial delay task")
+			.isEqualTo(1);
+
+		vts.advanceTimeBy(Duration.ofDays(1));
+		assertThat(vts.getScheduledTaskCount())
+			.as("Still on initial delay")
+			.isEqualTo(1);
+
+		vts.advanceTimeBy(Duration.ofDays(10));
+		assertThat(vts.getScheduledTaskCount())
+			.as("first periodical task scheduled after initial one")
+			.isEqualTo(2);
+
+		vts.advanceTimeBy(Duration.ofDays(5));
+		assertThat(vts.getScheduledTaskCount())
+			.as("second periodical task scheduled")
+			.isEqualTo(3);
+	}
+
 	@SuppressWarnings("unchecked")
 	private static Scheduler uncache(Scheduler potentialCached) {
 		if (potentialCached instanceof Supplier) {
@@ -263,4 +328,4 @@ public void cleanup() {
 		VirtualTimeScheduler.reset();
 	}
 
-}
+}