class PeerTypeRegistration implements TypeRegistration {

  @Override
  public void initialize() {
    // Relevant during reconnect
    TypeRegistry typeRegistry = cache.getPdxRegistry();
    if (typeRegistry != null) {
      typeRegistry.flushCache();
      logger.debug("Flushing TypeRegistry");
    }

    InternalRegionFactory factory = cache.createInternalRegionFactory();

    factory.setScope(Scope.DISTRIBUTED_ACK);
    if (cache.getPdxPersistent()) {
      if (cache.getCacheConfig().pdxDiskStoreUserSet) {
        factory.setDiskStoreName(cache.getPdxDiskStore());
      } else {
        factory.setDiskStoreName(cache.getOrCreateDefaultDiskStore().getName());
      }
      factory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
    } else {
      factory.setDataPolicy(DataPolicy.REPLICATE);
    }

    // Add a listener that makes sure that if anyone in the DS is using PDX
    // Our PDX configuration is valid for this member. This is important if
    // we are the gateway, we need to validate that we have a distributed system
    // id.
    factory.addCacheListener(new CacheListenerAdapter<Object, Object>() {
      @Override
      public void afterCreate(EntryEvent<Object, Object> event) {
        verifyConfiguration();
        // update the local map and reverse map with the pdxtypes registered
        Object value = event.getNewValue();
        Object key = event.getKey();
        if (value != null) {
          updateLocalAndReverseMaps(key, value);
        }
      }
    });

    factory.setCacheWriter(new CacheWriterAdapter<Object, Object>() {

      @Override
      public void beforeCreate(EntryEvent<Object, Object> event) throws CacheWriterException {
        Object newValue = event.getNewValue();
        if (newValue instanceof PdxType) {
          logger.info("Adding new type: {}", ((PdxType) event.getNewValue()).toFormattedString());
        } else {
          logger.info("Adding new type: {} {}", event.getKey(),
              ((EnumInfo) newValue).toFormattedString());
        }
      }

      @Override
      public void beforeUpdate(EntryEvent<Object, Object> event) throws CacheWriterException {
        if (!event.getRegion().get(event.getKey()).equals(event.getNewValue())) {
          PdxRegistryMismatchException ex = new PdxRegistryMismatchException(
              "Trying to add a PDXType with the same id as an existing PDX type. id="
                  + event.getKey() + ", existing pdx type " + event.getOldValue() + ", new type "
                  + event.getNewValue());
          throw new CacheWriterException(ex);
        }
      }

    });

    factory.setIsUsedForMetaRegion(true);
    factory.setMetaRegionWithTransactions(true);
    try {
      idToType = factory.create(REGION_NAME);
    } catch (TimeoutException | RegionExistsException ex) {
      throw new PdxInitializationException("Could not create pdx registry", ex);
    }

    statistics.initialize();

    // If there is anything in the id to type registry,
    // we should validate our configuration now.
    // And send those types to any existing gateways.
    if (!getIdToType().isEmpty()) {
      verifyConfiguration();
    }
  }

  public  PeerTypeRegistration(final InternalCache cache);

}