Addressed code review comments, added unit-tests.
Change-Id: I2a74e9969540cb636658aa012f82a81e0fd2eac2
Issue-ID: SDC-1667
Signed-off-by: vempo <vitaliy.emporopulo@amdocs.com>
@Validated
public class ItemsImpl implements Items {
+ private static final String ONBOARDING_METHOD = "onboardingMethod";
+
private ItemManager itemManager = ItemManagerFactory.getInstance().createInterface();
private static ActivityLogManager activityLogManager = ActivityLogManagerFactory.getInstance().createInterface();
.put(ItemAction.RESTORE, new ActionSideAffects(ActivityType.Restore, NotificationEventTypes.RESTORE));
}
- private static final String ONBOARDING_METHOD = "onboardingMethod";
-
@Override
public Response actOn(ItemActionRequestDto request, String itemId, String user) {
actionSideAffectsMap.get(request.getAction()).execute(item, user);
try {
- Notifier notifier = NotifierFactory.getInstance();
- notifier.execute(Collections.singleton(itemId), request.getAction());
+ Notifier catalogNotifier = NotifierFactory.getInstance();
+ catalogNotifier.execute(Collections.singleton(itemId), request.getAction());
} catch (Exception e) {
- LOGGER.error("Failed to send catalog notification on item " + itemId + " Error: " + e.getMessage());
+ LOGGER.error("Failed to send catalog notification on item {}", itemId, e);
}
return Response.ok().build();
try {
notifier.notifySubscribers(syncEvent, userName);
} catch (Exception e) {
- LOGGER.error("Failed to send sync notification to users subscribed to item '" + itemId);
+ LOGGER.error("Failed to send sync notification to users subscribed to item '{}'", itemId, e);
}
}
}
package org.openecomp.sdcrests.item.rest.services.catalog.notification;
import java.util.Collection;
+import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import org.openecomp.sdc.logging.api.Logger;
import org.openecomp.sdc.logging.api.LoggerFactory;
private static final long DEFAULT_INTERVAL = 5000;
private final BiFunction<Collection<String>, ItemAction, Callable<NextAction>> taskProducer;
+ private final int numberOfRetries;
+ private final long retryInterval;
+
AsyncNotifier(BiFunction<Collection<String>, ItemAction, Callable<NextAction>> taskProducer) {
+ this(taskProducer, DEFAULT_NUM_OF_RETRIES, DEFAULT_INTERVAL);
+ }
+
+ AsyncNotifier(BiFunction<Collection<String>, ItemAction, Callable<NextAction>> taskProducer, int numOfRetries,
+ long retryInterval) {
this.taskProducer = taskProducer;
+ this.numberOfRetries = numOfRetries;
+ this.retryInterval = retryInterval;
}
@Override
Callable<AsyncNotifier.NextAction> worker = taskProducer.apply(itemIds, action);
- RetryingTask retryingTask =
- new RetryingTask(worker, DEFAULT_NUM_OF_RETRIES, DEFAULT_INTERVAL, EXECUTOR_SERVICE);
+ RetryingTask retryingTask = new RetryingTask(worker, numberOfRetries, retryInterval, EXECUTOR_SERVICE);
EXECUTOR_SERVICE.submit(LoggingContext.copyToCallable(retryingTask));
}
private final Callable<AsyncNotifier.NextAction> worker;
private final long delay;
private final ScheduledExecutorService scheduler;
- private volatile int retries;
+ private final AtomicInteger retries;
RetryingTask(Callable<AsyncNotifier.NextAction> worker, int numOfRetries, long delay,
ScheduledExecutorService scheduler) {
- this.worker = worker;
- this.retries = numOfRetries;
- this.delay = delay;
- this.scheduler = scheduler;
+ this.worker = Objects.requireNonNull(worker);
+ this.retries = new AtomicInteger(requirePositiveRetries(numOfRetries));
+ this.delay = requirePositiveDelay(delay);
+ this.scheduler = Objects.requireNonNull(scheduler);
+ }
+
+ private int requirePositiveRetries(int number) {
+
+ if (number < 1) {
+ throw new IllegalArgumentException("Number of retries must be positive");
+ }
+
+ return number;
+ }
+
+ private long requirePositiveDelay(long number) {
+
+ if (number < 1) {
+ throw new IllegalArgumentException("Delay must be positive");
+ }
+
+ return number;
}
@Override
- public synchronized Void call() throws Exception {
+ public Void call() throws Exception {
NextAction next = worker.call();
if (next == NextAction.DONE) {
return null;
}
- retries--;
- if (retries == 0) {
+ int attempts = retries.decrementAndGet();
+ if (attempts < 1) {
LOGGER.warn("Exhausted number of retries for task {}, exiting", worker);
return null;
}
private static final Notifier INSTANCE;
+ private NotifierFactory() {
+ // prevent instantiation
+ }
+
static {
INSTANCE = createInstance();
}
package org.openecomp.sdcrests.item.rest.services.catalog.notification.http;
import java.util.Collection;
+import java.util.EnumMap;
+import java.util.Map;
import java.util.concurrent.Callable;
import java.util.function.BiFunction;
import org.openecomp.sdc.common.session.SessionContextProviderFactory;
private static final String CATALOG_HTTP_PROTOCOL = "HTTP";
private static final String CATALOG_HTTPS_PROTOCOL = "HTTPS";
+ private static final Map<ItemAction, String> ACTION_PATHS;
+
+ static {
+ ACTION_PATHS = new EnumMap<>(ItemAction.class);
+ ACTION_PATHS.put(ItemAction.ARCHIVE, "archived");
+ ACTION_PATHS.put(ItemAction.RESTORE, "restored");
+ }
private final String notifyCatalogUrl;
return createNotificationTask(itemIds, action);
}
- private static String getEndpoint(ItemAction action) {
+ private HttpNotificationTask createNotificationTask(Collection<String> itemIds, ItemAction action) {
+ String userId = SessionContextProviderFactory.getInstance().createInterface().get().getUser().getUserId();
+ String notificationEndpoint = notifyCatalogUrl + getApiPath(action);
+ LOGGER.debug("Catalog notification URL: {}", notificationEndpoint);
+ return new HttpNotificationTask(notificationEndpoint, userId, itemIds);
+ }
- if (action == ItemAction.ARCHIVE) {
- return "archived";
- } else if (action == ItemAction.RESTORE) {
- return "restored";
- } else {
+ static String getApiPath(ItemAction action) {
+
+ String path = ACTION_PATHS.get(action);
+ if (path == null) {
throw new IllegalArgumentException("Unsupported action: " + action.name());
}
+
+ return path;
}
@Override
task.call();
}
- private HttpNotificationTask createNotificationTask(Collection<String> itemIds, ItemAction action) {
- String userId = SessionContextProviderFactory.getInstance().createInterface().get().getUser().getUserId();
- String notificationEndpoint = notifyCatalogUrl + getEndpoint(action);
- LOGGER.debug("Catalog notification URL: " + notificationEndpoint);
- return new HttpNotificationTask(notificationEndpoint, userId, itemIds);
- }
}
package org.openecomp.sdcrests.item.rest.services.catalog.notification;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.openecomp.sdcrests.item.rest.services.catalog.notification.AsyncNotifier.NextAction.DONE;
import static org.openecomp.sdcrests.item.rest.services.catalog.notification.AsyncNotifier.NextAction.RETRY;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.UUID;
import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiFunction;
import org.apache.commons.lang.mutable.MutableInt;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.ExpectedException;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
+import org.openecomp.sdcrests.item.types.ItemAction;
/**
* @author evitaliy
*/
public class AsyncNotifierTest {
+ private static final String NUMBER_OF_RETRIES_MESSAGE = "Number of retries must be positive";
+ private static final String DELAY_MESSAGE = "Delay must be positive";
+
+ @Rule
+ public ExpectedException exception = ExpectedException.none();
+
+ @Test(expected = NullPointerException.class)
+ public void errorWhenWorkerNull() {
+ new AsyncNotifier.RetryingTask(null, 10, 10, Mockito.mock(ScheduledExecutorService.class));
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void errorWhenSchedulerServiceNull() {
+ new AsyncNotifier.RetryingTask(() -> DONE, 10, 10, null);
+ }
+
+ @Test
+ public void errorWhenNumberOfRetriesNegative() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage(NUMBER_OF_RETRIES_MESSAGE);
+ new AsyncNotifier.RetryingTask(() -> DONE, -12, 10, Mockito.mock(ScheduledExecutorService.class));
+ }
+
+ @Test
+ public void errorWhenNumberOfRetriesZero() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage(NUMBER_OF_RETRIES_MESSAGE);
+ new AsyncNotifier.RetryingTask(() -> DONE, 0, 10, Mockito.mock(ScheduledExecutorService.class));
+ }
+
+ @Test
+ public void errorWhenDelayNegative() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage(DELAY_MESSAGE);
+ new AsyncNotifier.RetryingTask(() -> DONE, 1, -77, Mockito.mock(ScheduledExecutorService.class));
+ }
+
+ @Test
+ public void errorWhenDelayZero() {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage(DELAY_MESSAGE);
+ new AsyncNotifier.RetryingTask(() -> DONE, 1, 0, Mockito.mock(ScheduledExecutorService.class));
+ }
+
@Test
public void taskRunsOnceWhenSuccessful() throws Exception {
assertEquals(1, counter.intValue());
}
+ private ScheduledExecutorService createMockScheduledExecutor() {
+
+ ScheduledExecutorService executorServiceMock = Mockito.mock(ScheduledExecutorService.class);
+ Answer passThrough = invocation -> {
+ ((Callable<?>) invocation.getArgument(0)).call();
+ return null;
+ };
+
+ Mockito.doAnswer(passThrough).when(executorServiceMock).submit(any(Callable.class));
+ Mockito.doAnswer(passThrough).when(executorServiceMock)
+ .schedule(any(Callable.class), anyLong(), any(TimeUnit.class));
+ return executorServiceMock;
+ }
+
@Test
public void taskRunsTwiceWhenFailedFirstTime() throws Exception {
assertEquals(numOfRetries, counter.intValue());
}
- private ScheduledExecutorService createMockScheduledExecutor() {
+ @Test
+ public void workerExecutedWithGivenItemIdsAndAction()
+ throws InterruptedException, ExecutionException, TimeoutException {
- ScheduledExecutorService executorServiceMock = Mockito.mock(ScheduledExecutorService.class);
- Answer passThrough = invocation -> {
- ((Callable<?>) invocation.getArgument(0)).call();
- return null;
+ CompletableFuture<Boolean> completed = new CompletableFuture<>();
+ Callable<AsyncNotifier.NextAction> mockTask = () -> {
+ completed.complete(true);
+ return DONE;
};
- Mockito.doAnswer(passThrough).when(executorServiceMock).submit(any(Callable.class));
- Mockito.doAnswer(passThrough).when(executorServiceMock)
- .schedule(any(Callable.class), anyLong(), any(TimeUnit.class));
- return executorServiceMock;
+ final Collection<String> itemIds = Collections.singleton(UUID.randomUUID().toString());
+ final ItemAction action = ItemAction.RESTORE;
+
+ BiFunction<Collection<String>, ItemAction, Callable<AsyncNotifier.NextAction>> mockProducer = (i, a) -> {
+ assertEquals(itemIds, i);
+ assertEquals(action, a);
+ return mockTask;
+ };
+
+ new AsyncNotifier(mockProducer, 1, 1).execute(itemIds, action);
+ assertTrue(completed.get(5, TimeUnit.SECONDS));
}
}
\ No newline at end of file
package org.openecomp.sdcrests.item.rest.services.catalog.notification.http;
import static org.hamcrest.CoreMatchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import java.util.HashSet;
+import java.util.Set;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.openecomp.sdcrests.item.rest.services.catalog.notification.EntryNotConfiguredException;
+import org.openecomp.sdcrests.item.types.ItemAction;
/**
* @author evitaliy
@Rule
public ExpectedException exception = ExpectedException.none();
+ @Test
+ public void uniquePathExistsForEveryAction() {
+
+ ItemAction[] availableActions = ItemAction.values();
+ Set<String> collectedPaths = new HashSet<>(availableActions.length);
+ for (ItemAction action : availableActions) {
+ String path = HttpTaskProducer.getApiPath(action);
+ assertFalse("Path empty for action '" + action.name() + "'", path == null || path.isEmpty());
+ collectedPaths.add(path);
+ }
+
+ assertEquals("Paths not unique for some actions", availableActions.length, collectedPaths.size());
+ }
+
+ @Test
+ public void restorePathEqualsRestored() {
+ assertEquals("restored", HttpTaskProducer.getApiPath(ItemAction.RESTORE));
+ }
+
+ @Test
+ public void archivePathEqualsArchived() {
+ assertEquals("archived", HttpTaskProducer.getApiPath(ItemAction.ARCHIVE));
+ }
+
@Test
public void errorWhenProtocolNotDefined() {
HttpConfiguration config = mockConfiguration();
public void testGetInstanceSystemProperty() throws Throwable {
expectedException.expect(IOException.class);
- expectedException.expectMessage((NON_EXISTENT));
+ expectedException.expectMessage(containsString(NON_EXISTENT));
try (ConfigurationSystemPropertyUpdater updater = new ConfigurationSystemPropertyUpdater(NON_EXISTENT)) {
ConfigurationManager.getInstance();
Code Review / sdc.git / commitdiff