@Test @SuppressWarnings("unchecked") void removeNodesAndServicesOnNewConfig() { final ConfigurationProvider configProvider = mock(ConfigurationProvider.class); DirectProcessor<ClusterConfig> configs = DirectProcessor.create(); ClusterConfig clusterConfig = new ClusterConfig(); when(configProvider.configs()).thenReturn(configs); when(configProvider.config()).thenReturn(clusterConfig); Node mock101 = mock(Node.class); when(mock101.identifier()).thenReturn(new NodeIdentifier("10.143.190.101", 8091)); when(mock101.addService(any(ServiceType.class), anyInt(), any(Optional.class))) .thenReturn(Mono.empty()); when(mock101.removeService(any(ServiceType.class), any(Optional.class))) .thenReturn(Mono.empty()); when(mock101.serviceEnabled(any(ServiceType.class))).thenReturn(true); when(mock101.disconnect()).thenReturn(Mono.empty()); Node mock102 = mock(Node.class); when(mock102.identifier()).thenReturn(new NodeIdentifier("10.143.190.102", 8091)); when(mock102.addService(any(ServiceType.class), anyInt(), any(Optional.class))) .thenReturn(Mono.empty()); when(mock102.removeService(any(ServiceType.class), any(Optional.class))) .thenReturn(Mono.empty()); when(mock102.serviceEnabled(any(ServiceType.class))).thenReturn(true); when(mock102.disconnect()).thenReturn(Mono.empty()); final Map<String, Node> mocks = new HashMap<>(); mocks.put("10.143.190.101", mock101); mocks.put("10.143.190.102", mock102); new Core(ENV, AUTHENTICATOR, SeedNode.LOCALHOST) { @Override public ConfigurationProvider createConfigurationProvider() { return configProvider; } @Override protected Node createNode(final NodeIdentifier target, final Optional<String> alternate) { return mocks.get(target.address()); } }; configs.onNext(clusterConfig); BucketConfig twoNodesConfig = BucketConfigParser.parse( readResource("two_nodes_config_more_services.json", CoreTest.class), ENV, LOCALHOST ); clusterConfig.setBucketConfig(twoNodesConfig); configs.onNext(clusterConfig); verify(mock101, times(1)) .addService(ServiceType.VIEWS, 8092, Optional.empty()); verify(mock101, times(1)) .addService(ServiceType.MANAGER, 8091, Optional.empty()); verify(mock101, times(1)) .addService(ServiceType.QUERY, 8093, Optional.empty()); verify(mock101, times(1)) .addService(ServiceType.KV, 11210, Optional.of("travel-sample")); verify(mock102, times(1)) .addService(ServiceType.VIEWS, 8092, Optional.empty()); verify(mock102, times(1)) .addService(ServiceType.MANAGER, 8091, Optional.empty()); verify(mock102, times(1)) .addService(ServiceType.QUERY, 8093, Optional.empty()); verify(mock102, times(1)) .addService(ServiceType.KV, 11210, Optional.of("travel-sample")); verify(mock102, times(1)) .addService(ServiceType.SEARCH, 8094, Optional.empty()); BucketConfig twoNodesLessServices = BucketConfigParser.parse( readResource("two_nodes_config.json", CoreTest.class), ENV, LOCALHOST ); clusterConfig.setBucketConfig(twoNodesLessServices); configs.onNext(clusterConfig); verify(mock102, times(1)) .removeService(ServiceType.SEARCH, Optional.empty()); }
public static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes) { return new Core(environment, authenticator, seedNodes); }
Core { public static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes) { return new Core(environment, authenticator, seedNodes); } }
Core { public static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes) { return new Core(environment, authenticator, seedNodes); } protected  Core(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes); }
Core { public static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes) { return new Core(environment, authenticator, seedNodes); } protected  Core(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes); static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes); @Stability.Internal ConfigurationProvider configurationProvider(); void send(final Request<R> request); @Stability.Internal @SuppressWarnings({"unchecked"}) void send(final Request<R> request, final boolean registerForTimeout); CoreContext context(); @Stability.Internal Stream<EndpointDiagnostics> diagnostics(); @Stability.Internal Optional<Flux<ServiceState>> serviceState(NodeIdentifier nodeIdentifier, ServiceType type, Optional<String> bucket); @Stability.Internal void initGlobalConfig(); @Stability.Internal void openBucket(final String name); @Stability.Internal ClusterConfig clusterConfig(); @Stability.Internal Mono<Void> ensureServiceAt(final NodeIdentifier identifier, final ServiceType serviceType, final int port,
                                    final Optional<String> bucket, final Optional<String> alternateAddress); @Stability.Internal Mono<Void> shutdown(); @Stability.Internal Mono<Void> shutdown(Duration timeout); }
Core { public static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes) { return new Core(environment, authenticator, seedNodes); } protected  Core(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes); static Core create(final CoreEnvironment environment, final Authenticator authenticator, final Set<SeedNode> seedNodes); @Stability.Internal ConfigurationProvider configurationProvider(); void send(final Request<R> request); @Stability.Internal @SuppressWarnings({"unchecked"}) void send(final Request<R> request, final boolean registerForTimeout); CoreContext context(); @Stability.Internal Stream<EndpointDiagnostics> diagnostics(); @Stability.Internal Optional<Flux<ServiceState>> serviceState(NodeIdentifier nodeIdentifier, ServiceType type, Optional<String> bucket); @Stability.Internal void initGlobalConfig(); @Stability.Internal void openBucket(final String name); @Stability.Internal ClusterConfig clusterConfig(); @Stability.Internal Mono<Void> ensureServiceAt(final NodeIdentifier identifier, final ServiceType serviceType, final int port,
                                    final Optional<String> bucket, final Optional<String> alternateAddress); @Stability.Internal Mono<Void> shutdown(); @Stability.Internal Mono<Void> shutdown(Duration timeout);  }