+
+ @Override public String persistenceId() {
+ return "shard-manager-" + type;
+ }
+
+ @VisibleForTesting public Collection<String> getKnownModules() {
+ return knownModules;
+ }
+
+ private class ShardInformation {
+ private final String shardName;
+ private final ActorRef actor;
+ private final ActorPath actorPath;
+ private final Map<ShardIdentifier, String> peerAddresses;
+
+ private ShardInformation(String shardName, ActorRef actor,
+ Map<ShardIdentifier, String> peerAddresses) {
+ this.shardName = shardName;
+ this.actor = actor;
+ this.actorPath = actor.path();
+ this.peerAddresses = peerAddresses;
+ }
+
+ public String getShardName() {
+ return shardName;
+ }
+
+ public ActorRef getActor(){
+ return actor;
+ }
+
+ public ActorPath getActorPath() {
+ return actorPath;
+ }
+
+ public void updatePeerAddress(ShardIdentifier peerId, String peerAddress){
+ LOG.info("updatePeerAddress for peer {} with address {}", peerId,
+ peerAddress);
+ if(peerAddresses.containsKey(peerId)){
+ peerAddresses.put(peerId, peerAddress);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Sending PeerAddressResolved for peer {} with address {} to {}",
+ peerId, peerAddress, actor.path());
+ }
+ actor
+ .tell(new PeerAddressResolved(peerId, peerAddress),
+ getSelf());
+
+ }
+ }
+ }
+
+ private static class ShardManagerCreator implements Creator<ShardManager> {
+ private static final long serialVersionUID = 1L;
+
+ final String type;
+ final ClusterWrapper cluster;
+ final Configuration configuration;
+ final DatastoreContext datastoreContext;
+
+ ShardManagerCreator(String type, ClusterWrapper cluster,
+ Configuration configuration, DatastoreContext datastoreContext) {
+ this.type = type;
+ this.cluster = cluster;
+ this.configuration = configuration;
+ this.datastoreContext = datastoreContext;
+ }
+
+ @Override
+ public ShardManager create() throws Exception {
+ return new ShardManager(type, cluster, configuration, datastoreContext);
+ }
+ }
+
+ static class SchemaContextModules implements Serializable {
+ private final Set<String> modules;
+
+ SchemaContextModules(Set<String> modules){
+ this.modules = modules;
+ }
+
+ public Set<String> getModules() {
+ return modules;
+ }
+ }