@Test
public void testStaticRegistryBuilder() throws Exception {
  AtomixRegistry registry = SimpleRegistry.builder()
      .addProfileType(ConsensusProfile.TYPE)
      .addDiscoveryProviderType(BootstrapDiscoveryProvider.TYPE)
      .addPrimitiveType(AtomicCounterType.instance())
      .addProtocolType(MultiRaftProtocol.TYPE)
      .addPartitionGroupType(RaftPartitionGroup.TYPE)
      .build();

  assertEquals(ConsensusProfile.TYPE, registry.getType(Profile.Type.class, "consensus"));
  assertEquals(BootstrapDiscoveryProvider.TYPE, registry.getType(NodeDiscoveryProvider.Type.class, "bootstrap"));
  assertEquals(AtomicCounterType.instance(), registry.getType(PrimitiveType.class, "atomic-counter"));
  assertEquals(MultiRaftProtocol.TYPE, registry.getType(PrimitiveProtocol.Type.class, "multi-raft"));
  assertEquals(RaftPartitionGroup.TYPE, registry.getType(PartitionGroup.Type.class, "raft"));
}
 

public AbstractProxyClient(
    String name,
    PrimitiveType type,
    PrimitiveProtocol protocol,
    Collection<ProxySession<S>> partitions) {
  this.name = checkNotNull(name, "name cannot be null");
  this.type = checkNotNull(type, "type cannot be null");
  this.protocol = checkNotNull(protocol, "protocol cannot be null");
  partitions.forEach(partition -> {
    this.partitionIds.add(partition.partitionId());
    this.partitions.put(partition.partitionId(), partition);
    states.put(partition.partitionId(), PrimitiveState.CLOSED);
    partition.addStateChangeListener(state -> onStateChange(partition.partitionId(), state));
  });
  Collections.sort(partitionIds);
}
 

TestProtocolService(
    PartitionId partition,
    String name,
    PrimitiveType primitiveType,
    ServiceConfig config,
    PrimitiveService service,
    TestProtocolServiceRegistry registry,
    ThreadContext context) {
  this.partition = partition;
  this.name = name;
  this.primitiveType = primitiveType;
  this.config = config;
  this.service = service;
  this.registry = registry;
  this.context = context;
  this.clock = context.schedule(Duration.ofMillis(100), Duration.ofMillis(100), this::tick);
  open();
}
 

public PrimaryBackupSessionClient(
    String clientName,
    PartitionId partitionId,
    SessionId sessionId,
    PrimitiveType primitiveType,
    PrimitiveDescriptor descriptor,
    ClusterMembershipService clusterMembershipService,
    PrimaryBackupClientProtocol protocol,
    PrimaryElection primaryElection,
    ThreadContext threadContext) {
  this.partitionId = checkNotNull(partitionId);
  this.sessionId = checkNotNull(sessionId);
  this.primitiveType = primitiveType;
  this.descriptor = descriptor;
  this.clusterMembershipService = clusterMembershipService;
  this.protocol = protocol;
  this.primaryElection = primaryElection;
  this.threadContext = threadContext;
  clusterMembershipService.addListener(membershipEventListener);
  primaryElection.addListener(primaryElectionListener);
  this.log = ContextualLoggerFactory.getLogger(getClass(), LoggerContext.builder(SessionClient.class)
      .addValue(clientName)
      .add("type", primitiveType.name())
      .add("name", descriptor.name())
      .build());
}
 

public DefaultRaftSessionClient(
    String serviceName,
    PrimitiveType primitiveType,
    ServiceConfig serviceConfig,
    PartitionId partitionId,
    RaftClientProtocol protocol,
    MemberSelectorManager selectorManager,
    RaftSessionManager sessionManager,
    ReadConsistency readConsistency,
    CommunicationStrategy communicationStrategy,
    ThreadContext context,
    Duration minTimeout,
    Duration maxTimeout) {
  this.serviceName = checkNotNull(serviceName, "serviceName cannot be null");
  this.primitiveType = checkNotNull(primitiveType, "serviceType cannot be null");
  this.serviceConfig = checkNotNull(serviceConfig, "serviceConfig cannot be null");
  this.partitionId = checkNotNull(partitionId, "partitionId cannot be null");
  this.protocol = checkNotNull(protocol, "protocol cannot be null");
  this.selectorManager = checkNotNull(selectorManager, "selectorManager cannot be null");
  this.readConsistency = checkNotNull(readConsistency, "readConsistency cannot be null");
  this.communicationStrategy = checkNotNull(communicationStrategy, "communicationStrategy cannot be null");
  this.context = checkNotNull(context, "context cannot be null");
  this.minTimeout = checkNotNull(minTimeout, "minTimeout cannot be null");
  this.maxTimeout = checkNotNull(maxTimeout, "maxTimeout cannot be null");
  this.sessionManager = checkNotNull(sessionManager, "sessionManager cannot be null");
}
 

public RaftServiceContext(
    PrimitiveId primitiveId,
    String serviceName,
    PrimitiveType primitiveType,
    ServiceConfig config,
    PrimitiveService service,
    RaftContext raft,
    ThreadContextFactory threadContextFactory) {
  this.primitiveId = checkNotNull(primitiveId);
  this.serviceName = checkNotNull(serviceName);
  this.primitiveType = checkNotNull(primitiveType);
  this.config = checkNotNull(config);
  this.service = checkNotNull(service);
  this.raft = checkNotNull(raft);
  this.sessions = raft.getSessions();
  this.threadContextFactory = threadContextFactory;
  this.log = ContextualLoggerFactory.getLogger(getClass(), LoggerContext.builder(PrimitiveService.class)
      .addValue(primitiveId)
      .add("type", primitiveType)
      .add("name", serviceName)
      .build());
  service.init(this);
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .register(PrimitiveType.super.namespace())
      .register(Namespaces.BASIC)
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID)
      .register(CollectionUpdateResult.class)
      .register(CollectionUpdateResult.Status.class)
      .register(CollectionEvent.class)
      .register(CollectionEvent.Type.class)
      .register(IteratorBatch.class)
      .register(TransactionId.class)
      .register(TransactionLog.class)
      .register(SetUpdate.class)
      .register(SetUpdate.Type.class)
      .register(PrepareResult.class)
      .register(CommitResult.class)
      .register(RollbackResult.class)
      .build();
}
 

/**
 * Initializes a new service.
 */
@SuppressWarnings("unchecked")
private RaftServiceContext initializeService(PrimitiveId primitiveId, PrimitiveType primitiveType, String serviceName, byte[] config) {
  RaftServiceContext oldService = raft.getServices().getService(serviceName);
  ServiceConfig serviceConfig = config == null ? new ServiceConfig() : Serializer.using(primitiveType.namespace()).decode(config);
  RaftServiceContext service = new RaftServiceContext(
      primitiveId,
      serviceName,
      primitiveType,
      serviceConfig,
      primitiveType.newService(serviceConfig),
      raft,
      threadContextFactory);
  raft.getServices().registerService(service);

  // If a service with this name was already registered, remove all of its sessions.
  if (oldService != null) {
    raft.getSessions().removeSessions(oldService.serviceId());
  }
  return service;
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .register(PrimitiveType.super.namespace())
      .register(Namespaces.BASIC)
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID)
      .register(CollectionUpdateResult.class)
      .register(CollectionUpdateResult.Status.class)
      .register(CollectionEvent.class)
      .register(CollectionEvent.Type.class)
      .register(IteratorBatch.class)
      .register(TransactionId.class)
      .register(TransactionLog.class)
      .register(SetUpdate.class)
      .register(SetUpdate.Type.class)
      .register(PrepareResult.class)
      .register(CommitResult.class)
      .register(RollbackResult.class)
      .build();
}
 

@SuppressWarnings("unchecked")
protected AbstractSession(
    SessionId sessionId,
    String primitiveName,
    PrimitiveType primitiveType,
    MemberId memberId,
    Serializer serializer) {
  this.sessionId = checkNotNull(sessionId);
  this.primitiveName = checkNotNull(primitiveName);
  this.primitiveType = checkNotNull(primitiveType);
  this.memberId = memberId;
  this.serializer = checkNotNull(serializer);
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .register(PrimitiveType.super.namespace())
      .register(Namespaces.BASIC)
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID)
      .register(CollectionUpdateResult.class)
      .register(CollectionUpdateResult.Status.class)
      .register(CollectionEvent.class)
      .register(CollectionEvent.Type.class)
      .register(IteratorBatch.class)
      .build();
}
 

public DefaultProxyClient(
    String name,
    PrimitiveType type,
    PrimitiveProtocol protocol,
    Class<S> serviceType,
    Collection<SessionClient> partitions,
    Partitioner<String> partitioner) {
  super(name, type, protocol, createSessions(type, serviceType, partitions));
  this.partitioner = checkNotNull(partitioner);
  this.serializer = Serializer.using(type.namespace());
}
 

@SuppressWarnings("unchecked")
public LogProxySession(String name, PrimitiveType type, Class<S> serviceType, ServiceConfig serviceConfig, Serializer serializer, LogSession session) {
  this.name = checkNotNull(name, "name cannot be null");
  this.type = checkNotNull(type, "type cannot be null");
  this.service = type.newService(serviceConfig);
  this.serviceConfig = serviceConfig;
  this.userSerializer = checkNotNull(serializer, "serializer cannot be null");
  this.session = checkNotNull(session, "session cannot be null");
  ServiceProxyHandler serviceProxyHandler = new ServiceProxyHandler(serviceType);
  S serviceProxy = (S) java.lang.reflect.Proxy.newProxyInstance(serviceType.getClassLoader(), new Class[]{serviceType}, serviceProxyHandler);
  proxy = new ServiceProxy<>(serviceProxy, serviceProxyHandler);
}
 

@Override
public PrimitiveType getPrimitiveType(String typeName) {
  PrimitiveType type = primitiveTypes.get(typeName);
  if (type == null) {
    throw new ServiceException("Unknown primitive type " + typeName);
  }
  return type;
}
 

@SuppressWarnings("unchecked")
public PrimaryBackupServiceContext(
    String serverName,
    PrimitiveId primitiveId,
    PrimitiveType primitiveType,
    PrimitiveDescriptor descriptor,
    ThreadContext threadContext,
    ClusterMembershipService clusterMembershipService,
    MemberGroupService memberGroupService,
    PrimaryBackupServerProtocol protocol,
    PrimaryElection primaryElection) {
  this.localMemberId = clusterMembershipService.getLocalMember().id();
  this.serverName = checkNotNull(serverName);
  this.primitiveId = checkNotNull(primitiveId);
  this.primitiveType = checkNotNull(primitiveType);
  this.serviceConfig = Serializer.using(primitiveType.namespace()).decode(descriptor.config());
  this.descriptor = checkNotNull(descriptor);
  this.service = primitiveType.newService(serviceConfig);
  this.threadContext = checkNotNull(threadContext);
  this.clusterMembershipService = checkNotNull(clusterMembershipService);
  this.memberGroupService = checkNotNull(memberGroupService);
  this.protocol = checkNotNull(protocol);
  this.primaryElection = checkNotNull(primaryElection);
  this.log = ContextualLoggerFactory.getLogger(getClass(), LoggerContext.builder(PrimitiveService.class)
      .addValue(serverName)
      .add("type", descriptor.type())
      .add("name", descriptor.name())
      .build());
  clusterMembershipService.addListener(membershipEventListener);
  primaryElection.addListener(primaryElectionListener);
}
 

public RaftSession(
    SessionId sessionId,
    MemberId member,
    String name,
    PrimitiveType primitiveType,
    ReadConsistency readConsistency,
    long minTimeout,
    long maxTimeout,
    long lastUpdated,
    Serializer serializer,
    RaftServiceContext context,
    RaftContext server,
    ThreadContextFactory threadContextFactory) {
  super(sessionId, name, primitiveType, member, serializer);
  this.readConsistency = readConsistency;
  this.minTimeout = minTimeout;
  this.maxTimeout = maxTimeout;
  this.lastUpdated = lastUpdated;
  this.eventIndex = sessionId.id();
  this.completeIndex = sessionId.id();
  this.lastApplied = sessionId.id();
  this.protocol = server.getProtocol();
  this.context = context;
  this.server = server;
  this.eventExecutor = threadContextFactory.createContext();
  this.log = ContextualLoggerFactory.getLogger(getClass(), LoggerContext.builder(Session.class)
      .addValue(sessionId)
      .add("type", context.serviceType())
      .add("name", context.serviceName())
      .build());
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .register(PrimitiveType.super.namespace())
      .register(Leadership.class)
      .register(Leader.class)
      .build();
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .register(PrimitiveType.super.namespace())
      .register(Namespaces.BASIC)
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID)
      .register(CollectionUpdateResult.class)
      .register(CollectionUpdateResult.Status.class)
      .register(CollectionEvent.class)
      .register(CollectionEvent.Type.class)
      .register(IteratorBatch.class)
      .build();
}
 

protected DefaultDistributedCollectionService(PrimitiveType primitiveType, T collection) {
  super(primitiveType, DistributedCollectionClient.class);
  this.collection = collection;
  this.serializer = Serializer.using(Namespace.builder()
      .register(primitiveType.namespace())
      .register(SessionId.class)
      .register(IteratorContext.class)
      .build());
}
 

public TestProtocolSession(
    SessionId sessionId,
    String primitiveName,
    PrimitiveType primitiveType,
    MemberId memberId,
    Serializer serializer,
    TestSessionClient client,
    ThreadContext context) {
  super(sessionId, primitiveName, primitiveType, memberId, serializer);
  this.client = client;
  this.context = context;
}
 

@Override
public <B extends PrimitiveBuilder<B, C, P>, C extends PrimitiveConfig<C>, P extends SyncPrimitive> B primitiveBuilder(
    String name,
    PrimitiveType<B, C, P> primitiveType) {
  checkRunning();
  return primitives.primitiveBuilder(name, primitiveType);
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .register(PrimitiveType.super.namespace())
      .register(Namespaces.BASIC)
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID)
      .register(CollectionUpdateResult.class)
      .register(CollectionUpdateResult.Status.class)
      .register(CollectionEvent.class)
      .register(CollectionEvent.Type.class)
      .register(IteratorBatch.class)
      .build();
}
 

@Override
public CompletableFuture<Set<SessionMetadata>> getSessions(PrimitiveType primitiveType) {
  return getMetadata().thenApply(response -> response.sessions()
      .stream()
      .filter(s -> s.primitiveType().equals(primitiveType.name()))
      .collect(Collectors.toSet()));
}
 

/**
 * Gets or initializes a service context.
 */
private RaftServiceContext getOrInitializeService(PrimitiveId primitiveId, PrimitiveType primitiveType, String serviceName, byte[] config) {
  // Get the state machine executor or create one if it doesn't already exist.
  RaftServiceContext service = raft.getServices().getService(serviceName);
  if (service == null) {
    service = initializeService(primitiveId, primitiveType, serviceName, config);
  }
  return service;
}
 

/**
 * Applies an open session entry to the state machine.
 */
private long applyOpenSession(Indexed<OpenSessionEntry> entry) {
  PrimitiveType primitiveType = raft.getPrimitiveTypes().getPrimitiveType(entry.entry().serviceType());

  // Get the state machine executor or create one if it doesn't already exist.
  RaftServiceContext service = getOrInitializeService(
      PrimitiveId.from(entry.index()),
      primitiveType,
      entry.entry().serviceName(),
      entry.entry().serviceConfig());

  if (service == null) {
    throw new RaftException.UnknownService("Unknown service type " + entry.entry().serviceType());
  }

  SessionId sessionId = SessionId.from(entry.index());
  RaftSession session = raft.getSessions().addSession(new RaftSession(
      sessionId,
      MemberId.from(entry.entry().memberId()),
      entry.entry().serviceName(),
      primitiveType,
      entry.entry().readConsistency(),
      entry.entry().minTimeout(),
      entry.entry().maxTimeout(),
      entry.entry().timestamp(),
      service.serializer(),
      service,
      raft,
      threadContextFactory));
  return service.openSession(entry.index(), entry.entry().timestamp(), session);
}
 

TestSessionClient(
    String name,
    PrimitiveType type,
    SessionId sessionId,
    PartitionId partitionId,
    ThreadContext context,
    TestProtocolService service) {
  this.name = name;
  this.type = type;
  this.sessionId = sessionId;
  this.partitionId = partitionId;
  this.context = context;
  this.service = service;
}
 

@Override
public CompletableFuture<PrimitiveInfo> createPrimitive(String name, PrimitiveType type) {
  PrimitiveInfo info = new PrimitiveInfo(name, type);
  CompletableFuture<PrimitiveInfo> future = new CompletableFuture<>();
  primitives.putIfAbsent(name, type.name()).whenComplete((result, error) -> {
    if (error != null) {
      future.completeExceptionally(error);
    } else if (result == null || result.value().equals(type.name())) {
      future.complete(info);
    } else {
      future.completeExceptionally(new PrimitiveException("A different primitive with the same name already exists"));
    }
  });
  return future;
}
 

@Override
public <S> ProxyClient<S> newProxy(
    String primitiveName,
    PrimitiveType primitiveType,
    Class<S> serviceType,
    ServiceConfig serviceConfig,
    PartitionService partitionService) {
  return new LogProxyClient<S>(primitiveName, primitiveType, this, serviceType, serviceConfig, newClient(partitionService));
}
 

@Override
public PrimitiveType type() {
  return DistributedSetType.instance();
}
 

@Override
public PrimitiveType type() {
  return type;
}