opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataChangePublisherActor.java

   1 /*
   2  * Copyright (c) 2017 Inocybe Technologies and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import akka.actor.Props;
  11 import com.google.common.base.Optional;
  12 import com.google.common.base.Preconditions;
  13 import java.util.function.Consumer;
  14 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  15 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  16 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  17 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  18 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  19 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  20
  21 /**
  22  * Actor used to generate and publish DataChange notifications.
  23  *
  24  * @author Thomas Pantelis
  25  */
  26 public class ShardDataChangePublisherActor
  27         extends ShardDataTreeNotificationPublisherActor<ShardDataChangeListenerPublisher> {
  28
  29     private ShardDataChangePublisherActor(final String name, final String logContext) {
  30         super(new DefaultShardDataChangeListenerPublisher(logContext), name, logContext);
  31     }
  32
  33     @Override
  34     protected void handleReceive(Object message) {
  35         if (message instanceof RegisterListener) {
  36             RegisterListener reg = (RegisterListener)message;
  37             if (reg.initialState.isPresent()) {
  38                 DefaultShardDataChangeListenerPublisher.notifySingleListener(reg.path, reg.listener, reg.scope,
  39                         reg.initialState.get(), logContext());
  40             }
  41
  42             publisher().registerDataChangeListener(reg.path, reg.listener, reg.scope, Optional.absent(),
  43                     reg.onRegistration);
  44         } else {
  45             super.handleReceive(message);
  46         }
  47     }
  48
  49     static Props props(final String name, final String logContext) {
  50         return Props.create(ShardDataChangePublisherActor.class, name, logContext);
  51     }
  52
  53     static class RegisterListener {
  54         private final YangInstanceIdentifier path;
  55         private final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener;
  56         private final DataChangeScope scope;
  57         private final Optional<DataTreeCandidate> initialState;
  58         private final Consumer<ListenerRegistration<
  59             AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>> onRegistration;
  60
  61         RegisterListener(final YangInstanceIdentifier path,
  62                 final AsyncDataChangeListener<YangInstanceIdentifier,NormalizedNode<?, ?>> listener,
  63                 final DataChangeScope scope, final Optional<DataTreeCandidate> initialState,
  64                 final Consumer<ListenerRegistration<
  65                     AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>> onRegistration) {
  66             this.path = Preconditions.checkNotNull(path);
  67             this.listener = Preconditions.checkNotNull(listener);
  68             this.scope = Preconditions.checkNotNull(scope);
  69             this.initialState = Preconditions.checkNotNull(initialState);
  70             this.onRegistration = Preconditions.checkNotNull(onRegistration);
  71         }
  72     }
  73 }