CacheServerImpl extends AbstractCacheServer implements DistributionAdvisee { @Override public synchronized void start() throws IOException { Assert.assertTrue(this.cache != null); this.serialNumber = createSerialNumber(); if (DynamicRegionFactory.get().isOpen()) { if (!this.notifyBySubscription) { logger.info("Forcing notifyBySubscription to support dynamic regions"); this.notifyBySubscription = true; } } this.advisor = cacheServerAdvisorProvider.apply(this); this.loadMonitor = new LoadMonitor(loadProbe, maxConnections, loadPollInterval, FORCE_LOAD_UPDATE_FREQUENCY, advisor); ClientSubscriptionConfig clientSubscriptionConfig = getClientSubscriptionConfig(); String diskStoreName = clientSubscriptionConfig.getDiskStoreName(); OverflowAttributes overflowAttributes = new OverflowAttributes() { @Override public String getEvictionPolicy() { return clientSubscriptionConfig.getEvictionPolicy(); } @Override public int getQueueCapacity() { return clientSubscriptionConfig.getCapacity(); } @Override public int getPort() { return port; } @Override public boolean isDiskStore() { return diskStoreName != null; } @Override public String getOverflowDirectory() { return clientSubscriptionConfig.getOverflowDirectory(); } @Override public String getDiskStoreName() { return diskStoreName; } }; acceptor = createAcceptor(overflowAttributes); this.acceptor.start(); this.advisor.handshake(); this.loadMonitor.start(new ServerLocation(getExternalAddress(), getPort()), acceptor.getStats()); ClientHealthMonitoringRegion.getInstance(this.cache); logger.info(String.format("CacheServer Configuration: %s", getConfig())); ClientMembershipListener[] membershipListeners = ClientMembership.getClientMembershipListeners(); boolean membershipListenerRegistered = false; for (ClientMembershipListener membershipListener : membershipListeners) { if (listener == membershipListener) { membershipListenerRegistered = true; break; } } if (!membershipListenerRegistered) { ClientMembership.registerClientMembershipListener(listener); } if (sendResourceEvents) { InternalDistributedSystem system = cache.getInternalDistributedSystem(); system.handleResourceEvent(ResourceEvent.CACHE_SERVER_START, this); } } CacheServerImpl(final InternalCache cache,
      final SecurityService securityService,
      final StatisticsClock statisticsClock,
      final AcceptorBuilder acceptorBuilder,
      final boolean sendResourceEvents,
      final boolean includeMembershipGroups,
      final Supplier<SocketCreator> socketCreatorSupplier,
      final CacheClientNotifierProvider cacheClientNotifierProvider,
      final ClientHealthMonitorProvider clientHealthMonitorProvider,
      final Function<DistributionAdvisee, CacheServerAdvisor> cacheServerAdvisorProvider); @Override CancelCriterion getCancelCriterion(); @Override StatisticsClock getStatisticsClock(); @Override int getPort(); @Override void setPort(int port); @Override void setBindAddress(String address); @Override void setHostnameForClients(String name); @Override void setMaxConnections(int maxCon); @Override void setMaxThreads(int maxThreads); @Override void setNotifyBySubscription(boolean b); @Override void setMaximumMessageCount(int maximumMessageCount); @Override void setSocketBufferSize(int socketBufferSize); @Override int getSocketBufferSize(); @Override void setMaximumTimeBetweenPings(int maximumTimeBetweenPings); @Override int getMaximumTimeBetweenPings(); @Override void setLoadPollInterval(long loadPollInterval); @Override int getMaximumMessageCount(); @Override void setLoadProbe(ServerLoadProbe loadProbe); @Override int getMessageTimeToLive(); @Override ClientSubscriptionConfig getClientSubscriptionConfig(); boolean isDefaultServer(); void setIsDefaultServer(); void configureFrom(CacheServer other); @Override synchronized void start(); @Override Acceptor createAcceptor(OverflowAttributes overflowAttributes); @Override String getExternalAddress(); String getExternalAddress(boolean checkServerRunning); @Override boolean isRunning(); @Override synchronized void stop(); @Override String toString(); @Override Acceptor getAcceptor(); @Override DistributionManager getDistributionManager(); @Override ClientSession getClientSession(String durableClientId); @Override ClientSession getClientSession(DistributedMember member); @Override Set getAllClientSessions(); static String clientMessagesRegion(InternalCache cache, String ePolicy, int capacity,
      int port, String overFlowDir, boolean isDiskStore); static String generateNameForClientMsgsRegion(int id); @Override DistributionAdvisor getDistributionAdvisor(); CacheServerAdvisor getCacheServerAdvisor(); @Override Profile getProfile(); @Override DistributionAdvisee getParentAdvisee(); @Override InternalDistributedSystem getSystem(); @Override String getName(); @Override String getFullPath(); @Override String[] getCombinedGroups(); @Override void fillInProfile(Profile profile); @Override int getSerialNumber(); @Override void registerInterestRegistrationListener(InterestRegistrationListener listener); @Override void unregisterInterestRegistrationListener(InterestRegistrationListener listener); @Override Set getInterestRegistrationListeners(); @Override ConnectionListener getConnectionListener(); @Override ServerConnectionFactory getServerConnectionFactory(); @Override SecurityService getSecurityService(); @Override long getTimeLimitMillis(); @Override Supplier<SocketCreator> getSocketCreatorSupplier(); @Override CacheClientNotifierProvider getCacheClientNotifierProvider(); @Override ClientHealthMonitorProvider getClientHealthMonitorProvider(); static final boolean ENABLE_NOTIFY_BY_SUBSCRIPTION_FALSE; }
@Test public void startNotifiesResourceEventCacheServerStart() throws IOException { InternalCacheServer server = new CacheServerImpl(cache, securityService, StatisticsClockFactory.disabledClock(), new AcceptorBuilder(), true, true, () -> socketCreator, (a, b, c, d, e, f, g, h, i) -> cacheClientNotifier, (a, b, c) -> clientHealthMonitor, a -> advisor); server.start(); verify(system).handleResourceEvent(same(ResourceEvent.CACHE_SERVER_START), same(server)); }