Skip to content

Commit 2c3a4ca

Browse files
taldekartaldekar
committed
Improve internal API readability
1 parent b915d76 commit 2c3a4ca

File tree

1 file changed

+71
-98
lines changed

1 file changed

+71
-98
lines changed

‎plugin/src/software/aws/toolkits/eclipse/amazonq/broker/EventBroker.java

+71-98
Original file line numberDiff line numberDiff line change
@@ -19,24 +19,22 @@
1919
import java.util.concurrent.TimeUnit;
2020
import java.util.concurrent.atomic.AtomicBoolean;
2121
import java.util.concurrent.atomic.AtomicReference;
22-
import java.util.concurrent.locks.ReentrantLock;
2322

2423
import software.aws.toolkits.eclipse.amazonq.observers.EventObserver;
2524
import software.aws.toolkits.eclipse.amazonq.observers.StreamObserver;
26-
import software.aws.toolkits.eclipse.amazonq.plugin.Activator;
2725

2826
public final class EventBroker {
2927

3028
@FunctionalInterface
3129
private interface TypedCallable<T> {
32-
void call(T event);
30+
void callWith(T event);
3331
}
3432

35-
private final class BlockingCallerRunsPolicy implements RejectedExecutionHandler {
33+
public static final class CallerRunsPolicyBlocking implements RejectedExecutionHandler {
3634

3735
private final BlockingQueue<Runnable> workQueue;
3836

39-
BlockingCallerRunsPolicy(final BlockingQueue<Runnable> workQueue) {
37+
CallerRunsPolicyBlocking(final BlockingQueue<Runnable> workQueue) {
4038
this.workQueue = workQueue;
4139
}
4240

@@ -56,122 +54,98 @@ public void rejectedExecution(final Runnable runnable, final ThreadPoolExecutor
5654

5755
}
5856

59-
private class OrderedThreadPoolExecutor {
57+
public static final class OrderedThreadPoolExecutor {
6058

61-
private final Map<String, BlockingQueue<?>> typedEventQueue;
62-
private final Map<String, AtomicBoolean> typedJobStatus;
63-
private final Map<String, ReentrantLock> typedJobLock;
64-
private final Map<String, TypedCallable<?>> typedCallback;
59+
private final Map<String, BlockingQueue<?>> interestIdToEventQueueMap;
60+
private final Map<String, AtomicBoolean> interestIdToJobStatusMap;
61+
private final Map<String, TypedCallable<?>> interestIdToCallbackMap;
6562

66-
private final BlockingQueue<Runnable> workQueue;
63+
private final BlockingQueue<Runnable> scheduledJobsQueue;
6764
private final ThreadPoolExecutor executor;
6865
private final int eventQueueCapacity;
6966

70-
OrderedThreadPoolExecutor(final int coreThreadCount, final int maxThreadCount, final int queueCapacity,
71-
final int keepAliveTime, final int eventQueueCapacity) {
72-
workQueue = new ArrayBlockingQueue<>(queueCapacity);
73-
typedEventQueue = new ConcurrentHashMap<>();
74-
typedJobStatus = new ConcurrentHashMap<>();
75-
typedJobLock = new ConcurrentHashMap<>();
76-
typedCallback = new ConcurrentHashMap<>();
67+
OrderedThreadPoolExecutor(final int coreThreadCount, final int maxThreadCount, final int jobQueueCapacity,
68+
final int eventQueueCapacity, final int keepAliveTime, final TimeUnit keepAliveTimeUnit) {
69+
scheduledJobsQueue = new ArrayBlockingQueue<>(jobQueueCapacity);
70+
interestIdToEventQueueMap = new ConcurrentHashMap<>();
71+
interestIdToJobStatusMap = new ConcurrentHashMap<>();
72+
interestIdToCallbackMap = new ConcurrentHashMap<>();
7773

7874
this.eventQueueCapacity = eventQueueCapacity;
7975

80-
executor = new ThreadPoolExecutor(coreThreadCount, maxThreadCount, keepAliveTime, TimeUnit.MILLISECONDS,
81-
workQueue, Executors.defaultThreadFactory(), new BlockingCallerRunsPolicy(workQueue));
76+
executor = new ThreadPoolExecutor(coreThreadCount, maxThreadCount, keepAliveTime, keepAliveTimeUnit,
77+
scheduledJobsQueue, Executors.defaultThreadFactory(), new CallerRunsPolicyBlocking(scheduledJobsQueue));
8278
}
8379

84-
public <T, R> void registerCallback(final String interestId, final TypedCallable<R> callback) {
85-
typedCallback.putIfAbsent(interestId, callback);
80+
public <T, R> void registerCallbackForInterest(final String interestId, final TypedCallable<R> callback) {
81+
interestIdToCallbackMap.putIfAbsent(interestId, callback);
8682
}
8783

88-
public <T> boolean hasRegisteredCallback(final String interestType) {
89-
return typedCallback.containsKey(interestType);
84+
public <T> boolean isCallbackRegisteredForInterest(final String interestId) {
85+
return interestIdToCallbackMap.containsKey(interestId);
9086
}
9187

9288
@SuppressWarnings("unchecked")
93-
public <T, R> void submit(final String interestId, final R event) {
94-
BlockingQueue<R> eventQueue = (BlockingQueue<R>) typedEventQueue.computeIfAbsent(interestId,
89+
public <T, R> void submitEventForInterest(final String interestId, final R event) {
90+
BlockingQueue<R> eventQueue = (BlockingQueue<R>) interestIdToEventQueueMap.computeIfAbsent(interestId,
9591
k -> new ArrayBlockingQueue<>(eventQueueCapacity, true));
9692
try {
9793
eventQueue.put(event);
9894
} catch (InterruptedException e) {
9995
e.printStackTrace();
10096
}
10197

102-
handleScheduling(interestId, (Class<R>) event.getClass(), eventQueue);
98+
handleJobScheduling(interestId, (Class<R>) event.getClass(), eventQueue);
10399
}
104100

105-
public <T, R> void handleScheduling(final String interestId, final Class<R> eventType,
101+
private <T, R> void handleJobScheduling(final String interestId, final Class<R> eventType,
106102
final BlockingQueue<R> eventQueue) {
107-
AtomicBoolean jobStatus = typedJobStatus.computeIfAbsent(interestId, k -> new AtomicBoolean(false));
108-
ReentrantLock jobLock = typedJobLock.computeIfAbsent(interestId, k -> new ReentrantLock(true));
103+
AtomicBoolean jobStatus = interestIdToJobStatusMap.computeIfAbsent(interestId, k -> new AtomicBoolean(false));
109104

110-
jobLock.lock();
111-
try {
112-
if (!jobStatus.get() && !eventQueue.isEmpty()) {
113-
if (jobStatus.compareAndSet(false, true)) {
114-
executor.submit(() -> processEventQueue(interestId, eventType,
115-
eventQueue, jobStatus, jobLock));
116-
}
117-
}
118-
} finally {
119-
jobLock.unlock();
105+
if (!jobStatus.get() && !eventQueue.isEmpty() && jobStatus.compareAndSet(false, true)) {
106+
executor.submit(() -> processQueuedEvents(interestId, eventType, eventQueue, jobStatus));
120107
}
121108
}
122109

123110
@SuppressWarnings("unchecked")
124-
public <T, R> void processEventQueue(final String interestId, final Class<R> eventType,
125-
final BlockingQueue<R> eventQueue, final AtomicBoolean jobStatus, final ReentrantLock jobLock) {
126-
if (jobStatus == null || jobLock == null || eventQueue == null) {
127-
throw new NullPointerException("ThreadPoolExecutor in unexpected state");
128-
}
129-
130-
jobLock.lock();
111+
private <T, R> void processQueuedEvents(final String interestId, final Class<R> eventType,
112+
final BlockingQueue<R> eventQueue, final AtomicBoolean jobStatus) {
131113
try {
132-
TypedCallable<R> eventCallback = (TypedCallable<R>) typedCallback.get(interestId);
114+
TypedCallable<R> eventCallback = (TypedCallable<R>) interestIdToCallbackMap.get(interestId);
133115
if (eventCallback == null) {
134116
return;
135117
}
136118

137119
while (!eventQueue.isEmpty()) {
138-
try {
139-
R newEvent = eventQueue.take();
140-
if (newEvent != null) {
141-
try {
142-
eventCallback.call(newEvent);
143-
} catch (Exception e) {
144-
e.printStackTrace();
145-
}
120+
R newEvent = eventQueue.poll();
121+
if (newEvent != null) {
122+
try {
123+
eventCallback.callWith(newEvent);
124+
} catch (Exception e) {
125+
e.printStackTrace();
146126
}
147-
} catch (InterruptedException e) {
148-
e.printStackTrace();
149127
}
150128
}
151129
} finally {
152-
try {
153-
jobStatus.set(false);
154-
} finally {
155-
jobLock.unlock();
156-
}
130+
jobStatus.set(false);
157131
}
158132
}
133+
159134
}
160135

161136
private static final EventBroker INSTANCE;
162-
private final Map<Class<?>, SubmissionPublisher<?>> publishers;
163-
private final OrderedThreadPoolExecutor emissionExecutor;
164-
private final OrderedThreadPoolExecutor consumptionExecutor;
137+
private final Map<Class<?>, SubmissionPublisher<?>> eventTypeToPublisherMap;
138+
private final OrderedThreadPoolExecutor publisherExecutor;
139+
private final OrderedThreadPoolExecutor subscriberExecutor;
165140

166141
static {
167142
INSTANCE = new EventBroker();
168143
}
169144

170145
private EventBroker() {
171-
publishers = new ConcurrentHashMap<>();
172-
173-
emissionExecutor = new OrderedThreadPoolExecutor(5, 30, 30, 10, 100000000);
174-
consumptionExecutor = new OrderedThreadPoolExecutor(5, 30, 30, 10, 100000000);
146+
eventTypeToPublisherMap = new ConcurrentHashMap<>();
147+
publisherExecutor = new OrderedThreadPoolExecutor(3, 10, 10, 100, 10, TimeUnit.MILLISECONDS);
148+
subscriberExecutor = new OrderedThreadPoolExecutor(3, 10, 10, 100, 10, TimeUnit.MILLISECONDS);
175149
}
176150

177151
public static EventBroker getInstance() {
@@ -184,20 +158,20 @@ public <T> void post(final T event) {
184158
return;
185159
}
186160

187-
SubmissionPublisher<T> publisher = getPublisher((Class<T>) event.getClass());
188-
if (!emissionExecutor.hasRegisteredCallback((event.getClass().getName()))) {
189-
registerPublisherCallback(publisher, event.getClass().getName());
161+
SubmissionPublisher<T> publisher = getPublisherForEventType((Class<T>) event.getClass());
162+
if (!publisherExecutor.isCallbackRegisteredForInterest((event.getClass().getName()))) {
163+
registerPublisherCallbackForInterest(event.getClass().getName(), publisher);
190164
}
191165

192-
emissionExecutor.submit(event.getClass().getName(), event);
166+
publisherExecutor.submitEventForInterest(event.getClass().getName(), event);
193167
}
194168

195169
public <T> Subscription subscribe(final EventObserver<T> observer) {
196-
SubmissionPublisher<T> publisher = getPublisher(observer.getEventType());
170+
SubmissionPublisher<T> publisher = getPublisherForEventType(observer.getEventType());
197171
AtomicReference<Subscription> subscriptionReference = new AtomicReference<>();
198172
String subscriberId = UUID.randomUUID().toString();
199173

200-
registerSubscriberCallback(observer, subscriberId);
174+
registerSubscriberCallbackForInterest(subscriberId, observer);
201175

202176
Subscriber<T> subscriber = new Subscriber<>() {
203177

@@ -207,19 +181,18 @@ public <T> Subscription subscribe(final EventObserver<T> observer) {
207181
public void onSubscribe(final Subscription subscription) {
208182
this.subscription = subscription;
209183
subscriptionReference.set(subscription);
210-
211184
this.subscription.request(1);
212185
}
213186

214187
@Override
215188
public void onNext(final T event) {
216-
consumptionExecutor.submit(subscriberId, event);
217-
this.subscription.request(1);
189+
subscriberExecutor.submitEventForInterest(subscriberId, event);
190+
subscription.request(1);
218191
}
219192

220193
@Override
221-
public void onError(final Throwable throwable) {
222-
return;
194+
public void onError(final Throwable error) {
195+
error.printStackTrace();
223196
}
224197

225198
@Override
@@ -234,11 +207,11 @@ public void onComplete() {
234207
}
235208

236209
public <T> Subscription subscribe(final StreamObserver<T> observer) {
237-
SubmissionPublisher<T> publisher = getPublisher(observer.getEventType());
210+
SubmissionPublisher<T> publisher = getPublisherForEventType(observer.getEventType());
238211
AtomicReference<Subscription> subscriptionReference = new AtomicReference<>();
239212
String subscriberId = UUID.randomUUID().toString();
240213

241-
registerSubscriberCallback(observer, subscriberId);
214+
registerSubscriberCallbackForInterest(subscriberId, observer);
242215

243216
Subscriber<T> subscriber = new Subscriber<>() {
244217

@@ -248,19 +221,18 @@ public <T> Subscription subscribe(final StreamObserver<T> observer) {
248221
public void onSubscribe(final Subscription subscription) {
249222
this.subscription = subscription;
250223
subscriptionReference.set(subscription);
251-
252224
this.subscription.request(1);
253225
}
254226

255227
@Override
256228
public void onNext(final T event) {
257-
consumptionExecutor.submit(subscriberId, event);
258-
this.subscription.request(1);
229+
subscriberExecutor.submitEventForInterest(subscriberId, event);
230+
subscription.request(1);
259231
}
260232

261233
@Override
262-
public void onError(final Throwable throwable) {
263-
observer.onError(throwable);
234+
public void onError(final Throwable error) {
235+
observer.onError(error);
264236
}
265237

266238
@Override
@@ -275,30 +247,31 @@ public void onComplete() {
275247
}
276248

277249
@SuppressWarnings("unchecked")
278-
private <T> SubmissionPublisher<T> getPublisher(final Class<T> eventType) {
279-
return (SubmissionPublisher<T>) publishers.computeIfAbsent(eventType,
250+
private <T> SubmissionPublisher<T> getPublisherForEventType(final Class<T> eventType) {
251+
return (SubmissionPublisher<T>) eventTypeToPublisherMap.computeIfAbsent(eventType,
280252
key -> new SubmissionPublisher<>(Runnable::run, Flow.defaultBufferSize()));
281253
}
282254

283-
private <T> void registerSubscriberCallback(final EventObserver<T> subscriber, final String subscriberId) {
284-
Activator.getLogger().info(subscriberId);
255+
private <T> void registerSubscriberCallbackForInterest(final String interestId,
256+
final EventObserver<T> observer) {
285257
TypedCallable<T> eventCallback = new TypedCallable<>() {
286258
@Override
287-
public void call(final T event) {
288-
subscriber.onEvent(event);
259+
public void callWith(final T event) {
260+
observer.onEvent(event);
289261
}
290262
};
291-
consumptionExecutor.registerCallback(subscriberId, eventCallback);
263+
subscriberExecutor.registerCallbackForInterest(interestId, eventCallback);
292264
}
293265

294-
private <T> void registerPublisherCallback(final SubmissionPublisher<T> publisher, final String eventId) {
266+
private <T> void registerPublisherCallbackForInterest(final String interestId,
267+
final SubmissionPublisher<T> publisher) {
295268
TypedCallable<T> eventCallback = new TypedCallable<>() {
296269
@Override
297-
public void call(final T event) {
270+
public void callWith(final T event) {
298271
publisher.submit(event);
299272
}
300273
};
301-
emissionExecutor.registerCallback(eventId, eventCallback);
274+
publisherExecutor.registerCallbackForInterest(interestId, eventCallback);
302275
}
303276

304277
}

0 commit comments

Comments
 (0)