8000 fix(NotificationManager): add thread-safety to NotificationManager by jaeopt · Pull Request #460 · optimizely/java-sdk · GitHub
[go: up one dir, main page]
Skip to content

fix(NotificationManager): add thread-safety to NotificationManager #460

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Feb 3, 2022
Merged

fix(NotificationManager): add thread-safety to NotificationManager #460

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d6fb6d8
add thread-safety to NotificationManager
jaeopt Jan 21, 2022
9673ee1
fix latch await return value test
jaeopt Jan 21, 2022
2634397
Merge branch 'master' into jae/notif-concurrent
jaeopt Feb 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 15 additions & 10 deletions core-api/src/main/java/com/optimizely/ab/notification/NotificationManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
Expand All @@ -33,7 +34,7 @@ public class NotificationManager<T> {

private static final Logger logger = LoggerFactory.getLogger(NotificationManager.class);

private final Map<Integer, NotificationHandler<T>> handlers = new LinkedHashMap<>();
private final Map<Integer, NotificationHandler<T>> handlers = Collections.synchronizedMap(new LinkedHashMap<>());
private final AtomicInteger counter;

public NotificationManager() {
Expand All @@ -47,10 +48,12 @@ public NotificationManager(AtomicInteger counter) {
public int addHandler(NotificationHandler<T> newHandler) {

// Prevent registering a duplicate listener.
for (NotificationHandler<T> handler: handlers.values()) {
if (handler.equals(newHandler)) {
logger.warn("Notification listener was already added");
return -1;
synchronized (handlers) {
for (NotificationHandler<T> handler : handlers.values()) {
if (handler.equals(newHandler)) {
logger.warn("Notification listener was already added");
return -1;
}
}
}

Expand All @@ -61,11 +64,13 @@ public int addHandler(NotificationHandler<T> newHandler) {
}

public void send(final T message) {
for (Map.Entry<Integer, NotificationHandler<T>> handler: handlers.entrySet()) {
try {
handler.getValue().handle(message);
} catch (Exception e) {
logger.warn("Catching exception sending notification for class: {}, handler: {}", message.getClass(), handler.getKey());
synchronized (handlers) {
for (Map.Entry<Integer, NotificationHandler<T>> handler: handlers.entrySet()) {
try {
handler.getValue().handle(message);
} catch (Exception e) {
logger.warn("Catching exception sending notification for class: {}, handler: {}", message.getClass(), handler.getKey());
}
}
}
}
Expand Down
33 changes: 33 additions & 0 deletions core-api/src/test/java/com/optimizely/ab/notification/NotificationManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line c B2D5 hange
Expand Up @@ -20,6 +20,11 @@
import org.junit.Test;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.*;
Expand Down Expand Up @@ -70,4 +75,32 @@ public void testSendWithError() {
assertEquals(1, messages.size());
assertEquals("message1", messages.get(0).getMessage());
}

@Test
public void testThreadSafety() throws InterruptedException {
int numThreads = 10;
int numRepeats = 2;
ExecutorService executor = Executors.newFixedThreadPool(numThreads);
CountDownLatch latch = new CountDownLatch(numThreads);
AtomicBoolean failedAlready = new AtomicBoolean(false);

for(int i = 0; i < numThreads; i++) {
executor.execute(() -> {
try {
for (int j = 0; j < numRepeats; j++) {
if(!failedAlready.get()) {
notificationManager.addHandler(new TestNotificationHandler<>());
notificationManager.send(new TestNotification("message1"));
}
}
} catch (Exception e) {
failedAlready.set(true);
} finally {
latch.countDown();
}
});
}
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertEquals(numThreads * numRepeats, notificationManager.size());
}
}
0