import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
/**
*
*/
-public class DistributedDataStore implements DOMStore, SchemaContextListener {
+public class DistributedDataStore implements DOMStore, SchemaContextListener, AutoCloseable {
private static final Logger
LOG = LoggerFactory.getLogger(DistributedDataStore.class);
private final String type;
private final ActorContext actorContext;
- public DistributedDataStore(ActorSystem actorSystem, String type) {
- this(new ActorContext(actorSystem, actorSystem.actorOf(ShardManager.props(type))), type);
+ private SchemaContext schemaContext;
+
+
+
+ /**
+ * Executor used to run FutureTask's
+ *
+ * This is typically used when we need to make a request to an actor and
+ * wait for it's response and the consumer needs to be provided a Future.
+ *
+ * FIXME : Make the thread pool configurable
+ */
+ private final ExecutorService executor =
+ Executors.newFixedThreadPool(10);
+
+ public DistributedDataStore(ActorSystem actorSystem, String type, ClusterWrapper cluster, Configuration configuration) {
+ this(new ActorContext(actorSystem, actorSystem.actorOf(ShardManager.props(type, cluster, configuration), "shardmanager-" + type), configuration), type);
}
public DistributedDataStore(ActorContext actorContext, String type) {
AsyncDataBroker.DataChangeScope scope) {
ActorRef dataChangeListenerActor = actorContext.getActorSystem().actorOf(
- DataChangeListener.props());
+ DataChangeListener.props(listener));
Object result = actorContext.executeShardOperation(Shard.DEFAULT_NAME,
new RegisterChangeListener(path, dataChangeListenerActor.path(),
- AsyncDataBroker.DataChangeScope.BASE),
+ AsyncDataBroker.DataChangeScope.BASE).toSerializable(),
ActorContext.ASK_DURATION
);
- RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
- return new DataChangeListenerRegistrationProxy(actorContext.actorSelection(reply.getListenerRegistrationPath()), listener);
+ RegisterChangeListenerReply reply = RegisterChangeListenerReply.fromSerializable(actorContext.getActorSystem(),result);
+ return new DataChangeListenerRegistrationProxy(actorContext.actorSelection(reply.getListenerRegistrationPath()), listener, dataChangeListenerActor);
}
@Override
public DOMStoreTransactionChain createTransactionChain() {
- return new TransactionChainProxy(actorContext);
+ return new TransactionChainProxy(actorContext, executor, schemaContext);
}
@Override
public DOMStoreReadTransaction newReadOnlyTransaction() {
- return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY);
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY,
+ executor, schemaContext);
}
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
- return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY);
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY,
+ executor, schemaContext);
}
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
- return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE);
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE,
+ executor, schemaContext);
}
@Override public void onGlobalContextUpdated(SchemaContext schemaContext) {
+ this.schemaContext = schemaContext;
actorContext.getShardManager().tell(
new UpdateSchemaContext(schemaContext), null);
}
+
+ @Override public void close() throws Exception {
+ actorContext.shutdown();
+
+ }
}