1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
|
/*
* ============LICENSE_START=======================================================
* ONAP
* ================================================================================
* Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* ============LICENSE_END=========================================================
*/
package org.onap.policy.common.utils.time;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
/**
* Scheduled executor service that uses {@link TestTimeMulti} to execute its tasks. Note:
* the invokeXxx() methods are not currently supported.
*/
public class PseudoScheduledExecutorService implements ScheduledExecutorService {
private static final String NOT_IMPLEMENTED_YET = "not implemented yet";
/**
* Object to be used to execute timer tasks.
*/
private final TestTimeMulti currentTime;
/**
* {@code True} if {@link #shutdown()} or {@link #shutdownNow()} has been called,
* {@code false} otherwise.
*/
private boolean shutdown = false;
/**
* Constructs the object.
*
* @param currentTime object to be used to execute timer tasks
*/
public PseudoScheduledExecutorService(TestTimeMulti currentTime) {
this.currentTime = currentTime;
}
/**
* Cancels <i>all</i> tasks that have not yet been executed.
*/
@Override
public void shutdown() {
shutdown = true;
currentTime.cancelItems(this);
}
/**
* Cancels <i>all</i> tasks that have not yet been executed. Does <i>not</i> interrupt
* any currently executing task.
*/
@Override
public List<Runnable> shutdownNow() {
shutdown = true;
return currentTime.cancelItems(this).stream().map(item -> ((RunnableItem) item).getAction())
.collect(Collectors.toList());
}
@Override
public boolean isShutdown() {
return shutdown;
}
@Override
public boolean isTerminated() {
return shutdown;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
return shutdown;
}
@Override
public <T> Future<T> submit(Callable<T> task) {
return enqueueRunOnce(0, new FutureTask<>(task));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
return enqueueRunOnce(0, new FutureTask<>(task, result));
}
@Override
public Future<?> submit(Runnable task) {
return enqueueRunOnce(0, new FutureTask<>(task, null));
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
throw new UnsupportedOperationException(NOT_IMPLEMENTED_YET);
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
throw new UnsupportedOperationException(NOT_IMPLEMENTED_YET);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
throw new UnsupportedOperationException(NOT_IMPLEMENTED_YET);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
throw new UnsupportedOperationException(NOT_IMPLEMENTED_YET);
}
@Override
public void execute(Runnable command) {
currentTime.enqueue(new RunnableItem(currentTime, this, 0, command));
}
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
return enqueueRunOnce(unit.toMillis(delay), new PseudoScheduledFuture<>(command, null, false));
}
@Override
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
return enqueueRunOnce(unit.toMillis(delay), new PseudoScheduledFuture<>(callable, false));
}
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
return enqueuePeriodic(unit.toMillis(initialDelay), unit.toMillis(period),
new PseudoScheduledFuture<>(command, null, true));
}
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
return enqueuePeriodic(unit.toMillis(initialDelay), unit.toMillis(delay),
new PseudoScheduledFuture<>(command, null, true));
}
/**
* Enqueues a future to be executed one time.
*
* @param delay delay until the future should be executed
* @param future future to be enqueued
* @return the future
*/
private <F extends FutureTask<T>, T> F enqueueRunOnce(long delay, F future) {
currentTime.enqueue(new RunnableItem(currentTime, this, delay, future));
return future;
}
/**
* Enqueues a future to be executed periodically.
*
* @param initialDelayMs delay until the future should be executed the first time
* @param periodMs delay between executions of the future
* @param future future to be enqueued
* @return the future
*/
private <T> ScheduledFuture<T> enqueuePeriodic(long initialDelayMs, long periodMs,
PseudoScheduledFuture<T> future) {
currentTime.enqueue(new PeriodicItem(currentTime, this, initialDelayMs, periodMs, future));
return future;
}
}
|
