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;
+
+ /**
+ * 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) {
- this(new ActorContext(actorSystem, actorSystem.actorOf(ShardManager.props(type))), type);
+ this(new ActorContext(actorSystem, actorSystem.actorOf(ShardManager.props(type), "shardmanager-" + type)), 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(),
);
RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
- return new DataChangeListenerRegistrationProxy(actorContext.actorSelection(reply.getListenerRegistrationPath()), listener);
+ return new DataChangeListenerRegistrationProxy(actorContext.actorSelection(reply.getListenerRegistrationPath()), listener, dataChangeListenerActor);
}
@Override
public DOMStoreTransactionChain createTransactionChain() {
- return new TransactionChainProxy(actorContext);
+ return new TransactionChainProxy(actorContext, executor);
}
@Override
public DOMStoreReadTransaction newReadOnlyTransaction() {
- return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY);
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY,
+ executor);
}
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
- return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY);
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY,
+ executor);
}
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
- return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE);
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE,
+ executor);
}
@Override public void onGlobalContextUpdated(SchemaContext schemaContext) {
actorContext.getShardManager().tell(
new UpdateSchemaContext(schemaContext), null);
}
+
+ @Override public void close() throws Exception {
+ actorContext.shutdown();
+
+ }
}