NotificationHookConsumer implements Service, ActiveStateChangeHandler { void startInternal(Configuration configuration, ExecutorService executorService) { if (consumers == null) { consumers = new ArrayList<>(); } if (executorService != null) { executors = executorService; } if (!HAConfiguration.isHAEnabled(configuration)) { LOG.info("HA is disabled, starting consumers inline."); startConsumers(executorService); } } @Inject  NotificationHookConsumer(NotificationInterface notificationInterface, AtlasEntityStore atlasEntityStore,
                                    ServiceState serviceState, AtlasInstanceConverter instanceConverter,
                                    AtlasTypeRegistry typeRegistry); @Override void start(); @Override void stop(); @Override void instanceIsActive(); @Override void instanceIsPassive(); static final String CONSUMER_THREADS_PROPERTY; static final String CONSUMER_RETRIES_PROPERTY; static final String CONSUMER_FAILEDCACHESIZE_PROPERTY; static final String CONSUMER_RETRY_INTERVAL; static final int SERVER_READY_WAIT_TIME_MS; }
@Test public void testConsumersStartedIfHAIsDisabled() throws Exception { when(configuration.getBoolean(HAConfiguration.ATLAS_SERVER_HA_ENABLED_KEY, false)).thenReturn(false); when(configuration.getInt(NotificationHookConsumer.CONSUMER_THREADS_PROPERTY, 1)).thenReturn(1); List<NotificationConsumer<Object>> consumers = new ArrayList(); consumers.add(mock(NotificationConsumer.class)); when(notificationInterface.createConsumers(NotificationInterface.NotificationType.HOOK, 1)). thenReturn(consumers); NotificationHookConsumer notificationHookConsumer = new NotificationHookConsumer(notificationInterface, atlasEntityStore, serviceState, instanceConverter, typeRegistry); notificationHookConsumer.startInternal(configuration, executorService); verify(notificationInterface).createConsumers(NotificationInterface.NotificationType.HOOK, 1); verify(executorService).submit(any(NotificationHookConsumer.HookConsumer.class)); }
@Test public void testConsumersAreNotStartedIfHAIsEnabled() throws Exception { when(configuration.containsKey(HAConfiguration.ATLAS_SERVER_HA_ENABLED_KEY)).thenReturn(true); when(configuration.getBoolean(HAConfiguration.ATLAS_SERVER_HA_ENABLED_KEY)).thenReturn(true); when(configuration.getInt(NotificationHookConsumer.CONSUMER_THREADS_PROPERTY, 1)).thenReturn(1); List<NotificationConsumer<Object>> consumers = new ArrayList(); consumers.add(mock(NotificationConsumer.class)); when(notificationInterface.createConsumers(NotificationInterface.NotificationType.HOOK, 1)). thenReturn(consumers); NotificationHookConsumer notificationHookConsumer = new NotificationHookConsumer(notificationInterface, atlasEntityStore, serviceState, instanceConverter, typeRegistry); notificationHookConsumer.startInternal(configuration, executorService); verifyZeroInteractions(notificationInterface); }