class CacheServerImpl extends AbstractCacheServer implements DistributionAdvisee {

  @Override
  public synchronized void start() throws IOException {
    Assert.assertTrue(this.cache != null);

    this.serialNumber = createSerialNumber();
    if (DynamicRegionFactory.get().isOpen()) {
      // force notifyBySubscription to be true so that meta info is pushed
      // from servers to clients instead of invalidates.
      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());

    // TODO : Need to provide facility to enable/disable client health monitoring.
    // Creating ClientHealthMonitoring region.
    // Force initialization on current cache
    ClientHealthMonitoringRegion.getInstance(this.cache);
    logger.info(String.format("CacheServer Configuration:  %s", getConfig()));

    /*
     * If the stopped cache server is restarted, we'll need to re-register the client membership
     * listener. If the listener is already registered it won't be registered as would the case when
     * start() is invoked for the first time.
     */
    ClientMembershipListener[] membershipListeners =
        ClientMembership.getClientMembershipListeners();

    boolean membershipListenerRegistered = false;
    for (ClientMembershipListener membershipListener : membershipListeners) {
      // just checking by reference as the listener instance is final
      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);

}