|
38 | 38 | import org.springframework.core.ReactiveAdapter; |
39 | 39 | import org.springframework.core.ReactiveAdapterRegistry; |
40 | 40 | import org.springframework.lang.Nullable; |
| 41 | +import org.springframework.scheduling.SchedulingAwareRunnable; |
41 | 42 | import org.springframework.scheduling.support.DefaultScheduledTaskObservationConvention; |
42 | 43 | import org.springframework.scheduling.support.ScheduledTaskObservationContext; |
43 | 44 | import org.springframework.scheduling.support.ScheduledTaskObservationConvention; |
@@ -120,8 +121,10 @@ public static Runnable createSubscriptionRunnable(Method method, Object targetBe |
120 | 121 |
|
121 | 122 | boolean shouldBlock = (scheduled.fixedDelay() > 0 || StringUtils.hasText(scheduled.fixedDelayString())); |
122 | 123 | Publisher<?> publisher = getPublisherFor(method, targetBean); |
123 | | - Supplier<ScheduledTaskObservationContext> contextSupplier = () -> new ScheduledTaskObservationContext(targetBean, method); |
124 | | - return new SubscribingRunnable(publisher, shouldBlock, subscriptionTrackerRegistry, observationRegistrySupplier, contextSupplier); |
| 124 | + Supplier<ScheduledTaskObservationContext> contextSupplier = |
| 125 | + () -> new ScheduledTaskObservationContext(targetBean, method); |
| 126 | + return new SubscribingRunnable(publisher, shouldBlock, scheduled.scheduler(), |
| 127 | + subscriptionTrackerRegistry, observationRegistrySupplier, contextSupplier); |
125 | 128 | } |
126 | 129 |
|
127 | 130 | /** |
@@ -180,30 +183,43 @@ static Publisher<?> getPublisherFor(Method method, Object bean) { |
180 | 183 | * Utility implementation of {@code Runnable} that subscribes to a {@code Publisher} |
181 | 184 | * or subscribes-then-blocks if {@code shouldBlock} is set to {@code true}. |
182 | 185 | */ |
183 | | - static final class SubscribingRunnable implements Runnable { |
| 186 | + static final class SubscribingRunnable implements SchedulingAwareRunnable { |
184 | 187 |
|
185 | | - private final Publisher<?> publisher; |
| 188 | + private static final ScheduledTaskObservationConvention DEFAULT_CONVENTION = |
| 189 | + new DefaultScheduledTaskObservationConvention(); |
186 | 190 |
|
187 | | - private static final ScheduledTaskObservationConvention DEFAULT_CONVENTION = new DefaultScheduledTaskObservationConvention(); |
| 191 | + private final Publisher<?> publisher; |
188 | 192 |
|
189 | 193 | final boolean shouldBlock; |
190 | 194 |
|
| 195 | + @Nullable |
| 196 | + private final String qualifier; |
| 197 | + |
191 | 198 | private final List<Runnable> subscriptionTrackerRegistry; |
192 | 199 |
|
193 | 200 | final Supplier<ObservationRegistry> observationRegistrySupplier; |
194 | 201 |
|
195 | 202 | final Supplier<ScheduledTaskObservationContext> contextSupplier; |
196 | 203 |
|
197 | | - SubscribingRunnable(Publisher<?> publisher, boolean shouldBlock, List<Runnable> subscriptionTrackerRegistry, |
198 | | - Supplier<ObservationRegistry> observationRegistrySupplier, Supplier<ScheduledTaskObservationContext> contextSupplier) { |
| 204 | + SubscribingRunnable(Publisher<?> publisher, boolean shouldBlock, |
| 205 | + @Nullable String qualifier, List<Runnable> subscriptionTrackerRegistry, |
| 206 | + Supplier<ObservationRegistry> observationRegistrySupplier, |
| 207 | + Supplier<ScheduledTaskObservationContext> contextSupplier) { |
199 | 208 |
|
200 | 209 | this.publisher = publisher; |
201 | 210 | this.shouldBlock = shouldBlock; |
| 211 | + this.qualifier = qualifier; |
202 | 212 | this.subscriptionTrackerRegistry = subscriptionTrackerRegistry; |
203 | 213 | this.observationRegistrySupplier = observationRegistrySupplier; |
204 | 214 | this.contextSupplier = contextSupplier; |
205 | 215 | } |
206 | 216 |
|
| 217 | + @Override |
| 218 | + @Nullable |
| 219 | + public String getQualifier() { |
| 220 | + return this.qualifier; |
| 221 | + } |
| 222 | + |
207 | 223 | @Override |
208 | 224 | public void run() { |
209 | 225 | Observation observation = TASKS_SCHEDULED_EXECUTION.observation(null, DEFAULT_CONVENTION, |
|
0 commit comments