+ public synchronized <T extends DOMDataTreeListener> ListenerRegistration<T> registerListener(final T listener,
+ final Collection<DOMDataTreeIdentifier> subtrees, final boolean allowRxMerges,
+ final Collection<DOMDataTreeProducer> producers) throws DOMDataTreeLoopException {
+ requireNonNull(listener, "listener");
+
+ // Cross-check specified trees for exclusivity and eliminate duplicates, noDupSubtrees is effectively a Set
+ final Collection<DOMDataTreeIdentifier> noDupSubtrees;
+ switch (subtrees.size()) {
+ case 0:
+ throw new IllegalArgumentException("Subtrees must not be empty.");
+ case 1:
+ noDupSubtrees = subtrees;
+ break;
+ default:
+ // Check subtrees for mutual inclusion, this is an O(N**2) operation
+ for (DOMDataTreeIdentifier toCheck : subtrees) {
+ for (DOMDataTreeIdentifier against : subtrees) {
+ if (!toCheck.equals(against)) {
+ checkArgument(!toCheck.contains(against), "Subtree %s contains subtree %s", toCheck,
+ against);
+ }
+ }
+ }
+
+ noDupSubtrees = ImmutableSet.copyOf(subtrees);
+ }
+
+ LOG.trace("Requested registration of listener {} to subtrees {}", listener, noDupSubtrees);
+
+ // Lookup shards corresponding to subtrees and construct a map of which subtrees we want from which shard
+ final ListMultimap<DOMDataTreeShardRegistration<?>, DOMDataTreeIdentifier> needed =
+ ArrayListMultimap.create();
+ for (final DOMDataTreeIdentifier subtree : subtrees) {
+ final DOMDataTreeShardRegistration<?> reg = verifyNotNull(shards.lookup(subtree).getValue());
+ needed.put(reg, subtree);
+ }
+
+ LOG.trace("Listener {} is attaching to shards {}", listener, needed);
+
+ // Sanity check: all selected shards have to support one of the listening interfaces
+ needed.asMap().forEach((reg, trees) -> {
+ final DOMDataTreeShard shard = reg.getInstance();
+ checkArgument(shard instanceof ListenableDOMDataTreeShard
+ || shard instanceof DOMStoreTreeChangePublisher, "Subtrees %s do not point to listenable subtree.",
+ trees);
+ });
+
+ // Sanity check: all producers have to come from this implementation and must not form loops
+ for (DOMDataTreeProducer producer : producers) {
+ checkArgument(producer instanceof ShardedDOMDataTreeProducer);
+ simpleLoopCheck(subtrees, ((ShardedDOMDataTreeProducer) producer).getSubtrees());
+ }
+
+ final ListenerRegistration<?> underlyingRegistration = createRegisteredListener(listener, needed.asMap(),
+ allowRxMerges, producers);
+ return new AbstractListenerRegistration<T>(listener) {
+ @Override
+ protected void removeRegistration() {
+ ShardedDOMDataTree.this.removeListener(listener);
+ underlyingRegistration.close();
+ }
+ };
+ }
+
+ private static ListenerRegistration<?> createRegisteredListener(final DOMDataTreeListener userListener,
+ final Map<DOMDataTreeShardRegistration<?>, Collection<DOMDataTreeIdentifier>> needed,
+ final boolean allowRxMerges, final Collection<DOMDataTreeProducer> producers) {
+ // FIXME: Add attachment of producers
+ for (final DOMDataTreeProducer producer : producers) {
+ // FIXME: We should also unbound listeners
+ ((ShardedDOMDataTreeProducer) producer).bindToListener(userListener);
+ }
+
+ return DOMDataTreeListenerAggregator.aggregateIfNeeded(userListener, needed, allowRxMerges,
+ DOMDataTreeShardRegistration::getInstance);
+ }
+
+ private static void simpleLoopCheck(final Collection<DOMDataTreeIdentifier> listen,
+ final Set<DOMDataTreeIdentifier> writes) throws DOMDataTreeLoopException {
+ for (final DOMDataTreeIdentifier listenPath : listen) {
+ for (final DOMDataTreeIdentifier writePath : writes) {
+ if (listenPath.contains(writePath)) {
+ throw new DOMDataTreeLoopException(String.format(
+ "Listener must not listen on parent (%s), and also writes child (%s)", listenPath,
+ writePath));
+ } else if (writePath.contains(listenPath)) {
+ throw new DOMDataTreeLoopException(
+ String.format("Listener must not write parent (%s), and also listen on child (%s)",
+ writePath, listenPath));
+ }
+ }
+ }
+ }
+
+ void removeListener(final DOMDataTreeListener listener) {
+ // FIXME: detach producers