org.springframework.core.task.TaskRejectedException#org.springframework.scheduling.Trigger源码实例Demo

@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	try {
		if (this.enterpriseConcurrentScheduler) {
			return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
		}
		else {
			ErrorHandler errorHandler =
					(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
			return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

public ScheduledFuture<?> schedule(Runnable task, final Trigger trigger) {
	ManagedScheduledExecutorService executor = (ManagedScheduledExecutorService) scheduledExecutor;
	return executor.schedule(task, new javax.enterprise.concurrent.Trigger() {
		@Override
		@Nullable
		public Date getNextRunTime(@Nullable LastExecution le, Date taskScheduledTime) {
			return (trigger.nextExecutionTime(le != null ?
					new SimpleTriggerContext(le.getScheduledStart(), le.getRunStart(), le.getRunEnd()) :
					new SimpleTriggerContext()));
		}
		@Override
		public boolean skipRun(LastExecution lastExecution, Date scheduledRunTime) {
			return false;
		}
	});
}
 

@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	try {
		if (this.enterpriseConcurrentScheduler) {
			return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
		}
		else {
			ErrorHandler errorHandler =
					(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
			return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
		}
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
	}
}
 

public ScheduledFuture<?> schedule(Runnable task, final Trigger trigger) {
	ManagedScheduledExecutorService executor = (ManagedScheduledExecutorService) scheduledExecutor;
	return executor.schedule(task, new javax.enterprise.concurrent.Trigger() {
		@Override
		@Nullable
		public Date getNextRunTime(@Nullable LastExecution le, Date taskScheduledTime) {
			return (trigger.nextExecutionTime(le != null ?
					new SimpleTriggerContext(le.getScheduledStart(), le.getRunStart(), le.getRunEnd()) :
					new SimpleTriggerContext()));
		}
		@Override
		public boolean skipRun(LastExecution lastExecution, Date scheduledRunTime) {
			return false;
		}
	});
}
 

public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
    if (!enabled) {
        logger.debug("skip : {}", task);

        return null;
    }

    ScheduledFuture<?> future = instance.schedule(task, trigger);
    String runnableKey = findRunnableKey(task);

    if (Boolean.FALSE.equals(skipMap.get(runnableKey))) {
        future.cancel(true);
    }

    return future;
}
 

@Test
void shouldPublishRenewalErrors() {

	prepareRenewal();
	when(this.vaultOperations.doWithSession(any(RestOperationsCallback.class)))
			.thenThrow(new HttpClientErrorException(HttpStatus.I_AM_A_TEAPOT));

	this.secretLeaseContainer.start();

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));

	runnableCaptor.getValue().run();

	verify(this.leaseListenerAdapter).onLeaseEvent(any(SecretLeaseCreatedEvent.class));
	verify(this.leaseListenerAdapter).onLeaseError(this.captor.capture(), any(VaultException.class));
	verifyNoMoreInteractions(this.leaseListenerAdapter);

	SecretLeaseEvent leaseEvent = this.captor.getValue();

	assertThat(leaseEvent.getSource()).isEqualTo(this.requestedSecret);
	assertThat(leaseEvent.getLease()).isNotNull();
}
 

@Test
@SuppressWarnings("unchecked")
void shouldRunTokenRenewal() {

	when(this.clientAuthentication.login())
			.thenReturn(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), eq(VaultResponse.class)))
			.thenReturn(fromToken(LoginToken.of("foo".toCharArray(), Duration.ofSeconds(10))));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);

	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));

	runnableCaptor.getValue().run();

	verify(this.restOperations).postForObject(eq("auth/token/renew-self"),
			eq(new HttpEntity<>(
					VaultHttpHeaders.from(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5))))),
			any(Class.class));
	verify(this.clientAuthentication, times(1)).login();
	verify(this.listener).onAuthenticationEvent(any(BeforeLoginTokenRenewedEvent.class));
	verify(this.listener).onAuthenticationEvent(any(AfterLoginTokenRenewedEvent.class));
}
 

@Test
void shouldReScheduleTokenRenewalAfterSuccessfulRenewal() {

	when(this.clientAuthentication.login())
			.thenReturn(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), eq(VaultResponse.class)))
			.thenReturn(fromToken(LoginToken.of("foo".toCharArray(), Duration.ofSeconds(10))));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);

	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));

	runnableCaptor.getValue().run();

	verify(this.taskScheduler, times(2)).schedule(any(Runnable.class), any(Trigger.class));
}
 

@Test
void shouldNotScheduleRenewalIfRenewalTtlExceedsThreshold() {

	when(this.clientAuthentication.login())
			.thenReturn(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), eq(VaultResponse.class)))
			.thenReturn(fromToken(LoginToken.of("foo".toCharArray(), Duration.ofSeconds(2))));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);

	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));

	runnableCaptor.getValue().run();

	verify(this.taskScheduler, times(1)).schedule(any(Runnable.class), any(Trigger.class));
}
 

@Test
void shouldReLoginIfRenewalTtlExceedsThreshold() {

	when(this.clientAuthentication.login()).thenReturn(
			LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)),
			LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), eq(VaultResponse.class)))
			.thenReturn(fromToken(LoginToken.of("foo".toCharArray(), Duration.ofSeconds(2))));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));
	runnableCaptor.getValue().run();

	assertThat(this.sessionManager.getSessionToken())
			.isEqualTo(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5)));

	verify(this.clientAuthentication, times(2)).login();
	verify(this.listener, times(2)).onAuthenticationEvent(any(AfterLoginEvent.class));
	verify(this.listener).onAuthenticationEvent(any(LoginTokenExpiredEvent.class));
}
 

@Test
void shouldReLoginIfRenewalFails() {

	when(this.clientAuthentication.login()).thenReturn(
			LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)),
			LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), eq(VaultResponse.class)))
			.thenThrow(new ResourceAccessException("Connection refused"));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));
	runnableCaptor.getValue().run();

	assertThat(this.sessionManager.getSessionToken())
			.isEqualTo(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5)));

	verify(this.clientAuthentication, times(2)).login();
}
 

@Test
void shouldRetainTokenAfterRenewalFailure() {

	when(this.clientAuthentication.login()).thenReturn(
			LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)),
			LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), eq(VaultResponse.class)))
			.thenThrow(new ResourceAccessException("Connection refused"));
	this.sessionManager.setLeaseStrategy(LeaseStrategy.retainOnError());

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));
	runnableCaptor.getValue().run();

	assertThat(this.sessionManager.getSessionToken())
			.isEqualTo(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));

	verify(this.clientAuthentication).login();
}
 

@Test
void shouldUseTaskScheduler() {

	this.sessionManager = new LifecycleAwareSessionManager(this.clientAuthentication, this.taskScheduler,
			this.restOperations);

	when(this.clientAuthentication.login())
			.thenReturn(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));

	ArgumentCaptor<Trigger> triggerCaptor = ArgumentCaptor.forClass(Trigger.class);

	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(any(Runnable.class), triggerCaptor.capture());

	assertThat(triggerCaptor.getValue().nextExecutionTime(null)).isNotNull();
	assertThat(triggerCaptor.getValue().nextExecutionTime(null)).isNull();
}
 

@Test
@SuppressWarnings("unchecked")
void shouldNotReScheduleTokenRenewalAfterFailedRenewal() {

	when(this.clientAuthentication.login())
			.thenReturn(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));
	when(this.restOperations.postForObject(anyString(), any(), ArgumentMatchers.<Class>any()))
			.thenThrow(new HttpServerErrorException(HttpStatus.INTERNAL_SERVER_ERROR));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);

	this.sessionManager.getSessionToken();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));

	runnableCaptor.getValue().run();

	verify(this.taskScheduler, times(1)).schedule(any(Runnable.class), any(Trigger.class));
}
 

@Test
void shouldReScheduleTokenRenewalAfterSuccessfulRenewal() {

	mockToken(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5)));

	when(this.responseSpec.bodyToMono(VaultResponse.class))
			.thenReturn(Mono.just(fromToken(LoginToken.of("foo".toCharArray(), Duration.ofSeconds(10)))));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);

	this.sessionManager.getSessionToken() //
			.as(StepVerifier::create) //
			.expectNextCount(1) //
			.verifyComplete();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));

	runnableCaptor.getValue().run();

	verify(this.taskScheduler, times(2)).schedule(any(Runnable.class), any(Trigger.class));
}
 

@Test
void shouldReLoginIfRenewalTtlExceedsThreshold() {

	when(this.tokenSupplier.getVaultToken()).thenReturn(
			Mono.just(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5))),
			Mono.just(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5))));
	when(this.responseSpec.bodyToMono(VaultResponse.class))
			.thenReturn(Mono.just(fromToken(LoginToken.of("foo".toCharArray(), Duration.ofSeconds(2)))));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	this.sessionManager.getSessionToken() //
			.as(StepVerifier::create) //
			.expectNextCount(1) //
			.verifyComplete();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));
	runnableCaptor.getValue().run();

	this.sessionManager.getSessionToken().as(StepVerifier::create)
			.expectNext(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5))).verifyComplete();

	verify(this.tokenSupplier, times(2)).getVaultToken();
	verify(this.listener, times(2)).onAuthenticationEvent(any(AfterLoginEvent.class));
	verify(this.listener).onAuthenticationEvent(any(LoginTokenExpiredEvent.class));
}
 

@Test
void shouldReLoginIfRenewFails() {

	when(this.tokenSupplier.getVaultToken()).thenReturn(
			Mono.just(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5))),
			Mono.just(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5))));
	when(this.responseSpec.bodyToMono(VaultResponse.class)).thenReturn(Mono.error(new RuntimeException("foo")));

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	this.sessionManager.getSessionToken() //
			.as(StepVerifier::create) //
			.expectNextCount(1) //
			.verifyComplete();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));
	runnableCaptor.getValue().run();

	this.sessionManager.getSessionToken().as(StepVerifier::create)
			.expectNext(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5))).verifyComplete();

	verify(this.tokenSupplier, times(2)).getVaultToken();
}
 

@Test
void shouldRetainTokenAfterRenewalFailure() {

	when(this.tokenSupplier.getVaultToken()).thenReturn(
			Mono.just(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5))),
			Mono.just(LoginToken.renewable("bar".toCharArray(), Duration.ofSeconds(5))));
	when(this.responseSpec.bodyToMono(VaultResponse.class)).thenReturn(Mono.error(new RuntimeException("foo")));
	this.sessionManager.setLeaseStrategy(LeaseStrategy.retainOnError());

	ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
	this.sessionManager.getSessionToken() //
			.as(StepVerifier::create) //
			.expectNextCount(1) //
			.verifyComplete();
	verify(this.taskScheduler).schedule(runnableCaptor.capture(), any(Trigger.class));
	runnableCaptor.getValue().run();

	this.sessionManager.getSessionToken().as(StepVerifier::create)
			.expectNext(LoginToken.renewable("login".toCharArray(), Duration.ofSeconds(5))).verifyComplete();

	verify(this.tokenSupplier).getVaultToken();
}
 

@Test
@SuppressWarnings("unchecked")
void shouldRenewLease() {

	prepareRenewal();

	when(this.vaultOperations.doWithSession(any(RestOperationsCallback.class)))
			.thenReturn(Lease.of("new_lease", Duration.ofSeconds(70), true));

	this.secretLeaseContainer.start();

	ArgumentCaptor<Runnable> captor = ArgumentCaptor.forClass(Runnable.class);
	verify(this.taskScheduler).schedule(captor.capture(), any(Trigger.class));

	captor.getValue().run();
	verifyZeroInteractions(this.scheduledFuture);
	verify(this.taskScheduler, times(2)).schedule(captor.capture(), any(Trigger.class));
}
 

@Test
void wontScheduleOnNonUnixWithSudo() throws IOException {
    Assumptions.assumeTrue(!SystemUtils.IS_OS_UNIX);
    final TaskScheduler scheduler = Mockito.mock(TaskScheduler.class);
    final Resource jobsDir = Mockito.mock(Resource.class);
    Mockito.when(jobsDir.exists()).thenReturn(true);
    final DataServices dataServices = Mockito.mock(DataServices.class);
    Mockito.when(dataServices.getPersistenceService()).thenReturn(Mockito.mock(PersistenceService.class));
    Assertions.assertThat(
        new DiskCleanupTask(
            new DiskCleanupProperties(),
            scheduler,
            jobsDir,
            dataServices,
            JobsProperties.getJobsPropertiesDefaults(),
            Mockito.mock(Executor.class),
            new SimpleMeterRegistry()
        )
    ).isNotNull();
    Mockito.verify(scheduler, Mockito.never()).schedule(Mockito.any(Runnable.class), Mockito.any(Trigger.class));
}
 

@Test
@SuppressWarnings("unchecked")
void shouldRetainLeaseAfterRenewalFailure() {

	prepareRenewal();
	when(this.vaultOperations.doWithSession(any(RestOperationsCallback.class)))
			.thenThrow(new VaultException("Renewal failure"));

	this.secretLeaseContainer.setLeaseStrategy(LeaseStrategy.retainOnError());
	this.secretLeaseContainer.start();

	ArgumentCaptor<Runnable> captor = ArgumentCaptor.forClass(Runnable.class);
	verify(this.taskScheduler).schedule(captor.capture(), any(Trigger.class));
	captor.getValue().run();

	verify(this.taskScheduler, times(2)).schedule(captor.capture(), any(Trigger.class));
	captor.getValue().run();

	verify(this.vaultOperations, times(2)).doWithSession(any(RestOperationsCallback.class));
}
 

@Test
void shouldRotateGenericSecretNow() {

	when(this.taskScheduler.schedule(any(Runnable.class), any(Trigger.class))).thenReturn(this.scheduledFuture);

	when(this.vaultOperations.read(this.rotatingGenericSecret.getPath())).thenReturn(
			createGenericSecrets(Collections.singletonMap("key", "value")),
			createGenericSecrets(Collections.singletonMap("foo", "bar")));

	this.secretLeaseContainer.addRequestedSecret(this.rotatingGenericSecret);
	this.secretLeaseContainer.start();

	this.secretLeaseContainer.rotate(this.rotatingGenericSecret);

	ArgumentCaptor<Runnable> captor = ArgumentCaptor.forClass(Runnable.class);
	verify(this.taskScheduler, times(2)).schedule(captor.capture(), any(Trigger.class));
	verify(this.scheduledFuture).cancel(false);
	verify(this.taskScheduler, times(2)).schedule(captor.capture(), any(Trigger.class));

	ArgumentCaptor<SecretLeaseEvent> createdEvents = ArgumentCaptor.forClass(SecretLeaseEvent.class);
	verify(this.leaseListenerAdapter, times(3)).onLeaseEvent(createdEvents.capture());
}
 

@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	ScheduledExecutorService executor = getScheduledExecutor();
	try {
		ErrorHandler errorHandler = this.errorHandler;
		if (errorHandler == null) {
			errorHandler = TaskUtils.getDefaultErrorHandler(true);
		}
		return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule();
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

public void setApplicationContext(ApplicationContext applicationContext)
		throws BeansException {
	trigger = applicationContext.getBean(getTriggerBeanName(),Trigger.class);
	if (trigger == null) {
		throw new RuntimeException();
	}
}
 

@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	ScheduledExecutorService executor = getScheduledExecutor();
	try {
		ErrorHandler errorHandler = this.errorHandler;
		if (errorHandler == null) {
			errorHandler = TaskUtils.getDefaultErrorHandler(true);
		}
		return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule();
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}
 

@Bean(name = TriggerConstants.TRIGGER_BEAN_NAME)
@Conditional(PeriodicTriggerCondition.class)
public Trigger periodicTrigger() {
	PeriodicTrigger trigger = new PeriodicTrigger(triggerProperties.getFixedDelay(),
			triggerProperties.getTimeUnit());
	trigger.setInitialDelay(triggerProperties.getInitialDelay());
	return trigger;
}
 

@Test
void scheduleRenewalShouldApplyExpiryThreshold() {

	prepareRenewal();

	this.secretLeaseContainer.start();

	ArgumentCaptor<Trigger> captor = ArgumentCaptor.forClass(Trigger.class);
	verify(this.taskScheduler).schedule(any(Runnable.class), captor.capture());

	Date nextExecutionTime = captor.getValue().nextExecutionTime(null);
	assertThat(nextExecutionTime).isBetween(new Date(System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(35)),
			new Date(System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(41)));
}
 

private Trigger getTrigger(Schedule schedule) {
    return triggerContext -> {
        // 将Cron 0/1 * * * * ? 输入取得下一次执行的时间
        final CronTrigger cronTrigger = new CronTrigger(schedule.getCron());
        final Date date = cronTrigger.nextExecutionTime(triggerContext);
        return date;
    };
}
 

public ScheduledFuture<?> schedule(Runnable task, final Trigger trigger) {
	ManagedScheduledExecutorService executor = (ManagedScheduledExecutorService) scheduledExecutor;
	return executor.schedule(task, new javax.enterprise.concurrent.Trigger() {
		@Override
		public Date getNextRunTime(LastExecution le, Date taskScheduledTime) {
			return (trigger.nextExecutionTime(le != null ?
					new SimpleTriggerContext(le.getScheduledStart(), le.getRunStart(), le.getRunEnd()) :
					new SimpleTriggerContext()));
		}
		@Override
		public boolean skipRun(LastExecution lastExecution, Date scheduledRunTime) {
			return false;
		}
	});
}
 

@Override
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
	ScheduledExecutorService executor = getScheduledExecutor();
	try {
		ErrorHandler errorHandler =
				(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
		return new ReschedulingRunnable(task, trigger, executor, errorHandler).schedule();
	}
	catch (RejectedExecutionException ex) {
		throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
	}
}