opendaylight/md-sal/samples/toaster-provider/src/main/java/org/opendaylight/controller/sample/toaster/provider/OpendaylightToaster.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. 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.sample.toaster.provider;
   9
  10 import static java.util.Objects.requireNonNull;
  11 import static org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION;
  12 import static org.opendaylight.mdsal.common.api.LogicalDatastoreType.OPERATIONAL;
  13 import static org.opendaylight.yangtools.yang.common.ErrorType.APPLICATION;
  14
  15 import com.google.common.collect.ImmutableClassToInstanceMap;
  16 import com.google.common.util.concurrent.FluentFuture;
  17 import com.google.common.util.concurrent.FutureCallback;
  18 import com.google.common.util.concurrent.Futures;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import com.google.common.util.concurrent.MoreExecutors;
  21 import com.google.common.util.concurrent.SettableFuture;
  22 import java.util.List;
  23 import java.util.Optional;
  24 import java.util.concurrent.Callable;
  25 import java.util.concurrent.ExecutionException;
  26 import java.util.concurrent.ExecutorService;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.Future;
  29 import java.util.concurrent.atomic.AtomicLong;
  30 import java.util.concurrent.atomic.AtomicReference;
  31 import java.util.function.Function;
  32 import javax.annotation.PreDestroy;
  33 import javax.inject.Inject;
  34 import javax.inject.Singleton;
  35 import org.eclipse.jdt.annotation.NonNull;
  36 import org.opendaylight.controller.md.sal.common.util.jmx.AbstractMXBean;
  37 import org.opendaylight.mdsal.binding.api.DataBroker;
  38 import org.opendaylight.mdsal.binding.api.DataTreeChangeListener;
  39 import org.opendaylight.mdsal.binding.api.DataTreeIdentifier;
  40 import org.opendaylight.mdsal.binding.api.DataTreeModification;
  41 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
  42 import org.opendaylight.mdsal.binding.api.ReadWriteTransaction;
  43 import org.opendaylight.mdsal.binding.api.RpcProviderService;
  44 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  45 import org.opendaylight.mdsal.common.api.CommitInfo;
  46 import org.opendaylight.mdsal.common.api.OptimisticLockFailedException;
  47 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  48 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.CancelToast;
  49 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.CancelToastInput;
  50 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.CancelToastOutput;
  51 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.CancelToastOutputBuilder;
  52 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.DisplayString;
  53 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.MakeToast;
  54 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.MakeToastInput;
  55 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.MakeToastOutput;
  56 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.MakeToastOutputBuilder;
  57 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.RestockToaster;
  58 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.RestockToasterInput;
  59 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.RestockToasterOutput;
  60 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.RestockToasterOutputBuilder;
  61 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.Toaster;
  62 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.Toaster.ToasterStatus;
  63 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.ToasterBuilder;
  64 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.ToasterOutOfBreadBuilder;
  65 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.ToasterRestocked;
  66 import org.opendaylight.yang.gen.v1.http.netconfcentral.org.ns.toaster.rev091120.ToasterRestockedBuilder;
  67 import org.opendaylight.yangtools.concepts.Registration;
  68 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  69 import org.opendaylight.yangtools.yang.binding.Rpc;
  70 import org.opendaylight.yangtools.yang.common.ErrorTag;
  71 import org.opendaylight.yangtools.yang.common.ErrorType;
  72 import org.opendaylight.yangtools.yang.common.RpcError;
  73 import org.opendaylight.yangtools.yang.common.RpcResult;
  74 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  75 import org.osgi.service.component.annotations.Activate;
  76 import org.osgi.service.component.annotations.Component;
  77 import org.osgi.service.component.annotations.Deactivate;
  78 import org.osgi.service.component.annotations.Reference;
  79 import org.osgi.service.metatype.annotations.AttributeDefinition;
  80 import org.osgi.service.metatype.annotations.Designate;
  81 import org.osgi.service.metatype.annotations.ObjectClassDefinition;
  82 import org.slf4j.Logger;
  83 import org.slf4j.LoggerFactory;
  84
  85 @Singleton
  86 @Component(service = MakeToast.class, immediate = true)
  87 @Designate(ocd = OpendaylightToaster.Configuration.class)
  88 public final class OpendaylightToaster extends AbstractMXBean
  89         implements MakeToast, ToasterProviderRuntimeMXBean, DataTreeChangeListener<Toaster>, AutoCloseable {
  90     @ObjectClassDefinition
  91     public @interface Configuration {
  92         @AttributeDefinition(description = "The name of the toaster's manufacturer", max = "255")
  93         String manufacturer() default TOASTER_MANUFACTURER;
  94         @AttributeDefinition(description = "The name of the toaster's model", max = "255")
  95         String modelNumber() default TOASTER_MODEL_NUMBER;
  96         @AttributeDefinition(description = "How many times we attempt to make toast before failing ",
  97             min = "0", max = "65535")
  98         int maxMakeToastTries() default 2;
  99     }
 100
 101     private static final CancelToastOutput EMPTY_CANCEL_OUTPUT = new CancelToastOutputBuilder().build();
 102     private static final MakeToastOutput EMPTY_MAKE_OUTPUT = new MakeToastOutputBuilder().build();
 103     private static final RestockToasterOutput EMPTY_RESTOCK_OUTPUT = new RestockToasterOutputBuilder().build();
 104
 105     private static final Logger LOG = LoggerFactory.getLogger(OpendaylightToaster.class);
 106
 107     private static final InstanceIdentifier<Toaster> TOASTER_IID = InstanceIdentifier.builder(Toaster.class).build();
 108     private static final String TOASTER_MANUFACTURER = "Opendaylight";
 109     private static final String TOASTER_MODEL_NUMBER = "Model 1 - Binding Aware";
 110
 111     private final DataBroker dataBroker;
 112     private final NotificationPublishService notificationProvider;
 113     private final Registration dataTreeChangeListenerRegistration;
 114     private final Registration reg;
 115
 116     private final ExecutorService executor;
 117
 118     // This holds the Future for the current make toast task and is used to cancel the current toast.
 119     private final AtomicReference<Future<?>> currentMakeToastTask = new AtomicReference<>();
 120
 121     // Thread safe holders
 122     private final AtomicLong amountOfBreadInStock = new AtomicLong(100);
 123     private final AtomicLong toastsMade = new AtomicLong(0);
 124     private final AtomicLong darknessFactor = new AtomicLong(1000);
 125
 126     private final @NonNull DisplayString manufacturer;
 127     private final @NonNull DisplayString modelNumber;
 128     private final int maxMakeToastTries;
 129
 130     public OpendaylightToaster(final DataBroker dataProvider,
 131             final NotificationPublishService notificationPublishService, final RpcProviderService rpcProviderService,
 132             final String manufacturer, final String modelNumber, final int maxMakeToastTries) {
 133         super("OpendaylightToaster", "toaster-provider", null);
 134         notificationProvider = requireNonNull(notificationPublishService);
 135         dataBroker = requireNonNull(dataProvider);
 136
 137         this.manufacturer = new DisplayString(manufacturer);
 138         this.modelNumber = new DisplayString(modelNumber);
 139         this.maxMakeToastTries = maxMakeToastTries;
 140
 141         executor = Executors.newFixedThreadPool(1);
 142         reg = rpcProviderService.registerRpcImplementations(ImmutableClassToInstanceMap.<Rpc<?, ?>>builder()
 143             .put(CancelToast.class, this::cancelToast)
 144             .put(MakeToast.class, this)
 145             .put(RestockToaster.class, this::restockToaster)
 146             .build());
 147
 148         LOG.info("Initializing...");
 149
 150         dataTreeChangeListenerRegistration = requireNonNull(dataBroker, "dataBroker must be set")
 151             .registerTreeChangeListener(DataTreeIdentifier.of(CONFIGURATION, TOASTER_IID), this);
 152         try {
 153             setToasterStatusUp(null).get();
 154         } catch (InterruptedException | ExecutionException e) {
 155             throw new IllegalStateException("Failed to commit initial data", e);
 156         }
 157
 158         // Register our MXBean.
 159         register();
 160     }
 161
 162     @Inject
 163     public OpendaylightToaster(final DataBroker dataProvider,
 164             final NotificationPublishService notificationPublishService, final RpcProviderService rpcProviderService) {
 165         this(dataProvider, notificationPublishService, rpcProviderService, TOASTER_MANUFACTURER, TOASTER_MODEL_NUMBER,
 166             2);
 167     }
 168
 169     @Activate
 170     public OpendaylightToaster(@Reference final DataBroker dataProvider,
 171             @Reference final NotificationPublishService notificationPublishService,
 172             @Reference final RpcProviderService rpcProviderService, final @NonNull Configuration configuration) {
 173         this(dataProvider, notificationPublishService, rpcProviderService, configuration.manufacturer(),
 174             configuration.modelNumber(), configuration.maxMakeToastTries());
 175     }
 176
 177     /**
 178      * Implemented from the AutoCloseable interface.
 179      */
 180     @Override
 181     @PreDestroy
 182     @Deactivate
 183     public void close() {
 184         LOG.info("Closing...");
 185
 186         // Unregister our MXBean.
 187         unregister();
 188         reg.close();
 189
 190         // When we close this service we need to shutdown our executor!
 191         executor.shutdown();
 192
 193         if (dataTreeChangeListenerRegistration != null) {
 194             dataTreeChangeListenerRegistration.close();
 195         }
 196
 197         if (dataBroker != null) {
 198             WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
 199             tx.delete(OPERATIONAL,TOASTER_IID);
 200             Futures.addCallback(tx.commit(), new FutureCallback<CommitInfo>() {
 201                 @Override
 202                 public void onSuccess(final CommitInfo result) {
 203                     LOG.debug("Successfully deleted the operational Toaster");
 204                 }
 205
 206                 @Override
 207                 public void onFailure(final Throwable failure) {
 208                     LOG.error("Delete of the operational Toaster failed", failure);
 209                 }
 210             }, MoreExecutors.directExecutor());
 211         }
 212     }
 213
 214     private Toaster buildToaster(final ToasterStatus status) {
 215         // note - we are simulating a device whose manufacture and model are
 216         // fixed (embedded) into the hardware.
 217         // This is why the manufacture and model number are hardcoded.
 218         return new ToasterBuilder()
 219             .setToasterManufacturer(manufacturer)
 220             .setToasterModelNumber(modelNumber)
 221             .setToasterStatus(status)
 222             .build();
 223     }
 224
 225     /**
 226      * Implemented from the DataTreeChangeListener interface.
 227      */
 228     @Override
 229     public void onDataTreeChanged(final List<DataTreeModification<Toaster>> changes) {
 230         for (var change: changes) {
 231             final var rootNode = change.getRootNode();
 232             switch (rootNode.modificationType()) {
 233                 case WRITE -> {
 234                     final var oldToaster = rootNode.dataBefore();
 235                     final var newToaster = rootNode.dataAfter();
 236                     LOG.info("onDataTreeChanged - Toaster config with path {} was added or replaced: old Toaster: {}, "
 237                         + "new Toaster: {}", change.getRootPath().path(), oldToaster, newToaster);
 238
 239                     final var darkness = newToaster.getDarknessFactor();
 240                     if (darkness != null) {
 241                         darknessFactor.set(darkness.toJava());
 242                     }
 243                 }
 244                 case DELETE -> LOG.info("onDataTreeChanged - Toaster config with path {} was deleted: old Toaster: {}",
 245                         change.getRootPath().path(), rootNode.dataBefore());
 246                 default -> {
 247                     // No-op
 248                 }
 249             }
 250         }
 251     }
 252
 253     /**
 254      * RPC call implemented from the ToasterService interface that cancels the current toast, if any.
 255      */
 256     private ListenableFuture<RpcResult<CancelToastOutput>> cancelToast(final CancelToastInput input) {
 257         final var current = currentMakeToastTask.getAndSet(null);
 258         if (current != null) {
 259             current.cancel(true);
 260         }
 261
 262         // Always return success from the cancel toast call
 263         return Futures.immediateFuture(RpcResultBuilder.success(EMPTY_CANCEL_OUTPUT).build());
 264     }
 265
 266     /**
 267      * RPC call implemented from the ToasterService interface that attempts to make toast.
 268      */
 269     @Override
 270     public ListenableFuture<RpcResult<MakeToastOutput>> invoke(final MakeToastInput input) {
 271         LOG.info("makeToast: {}", input);
 272         final var futureResult = SettableFuture.<RpcResult<MakeToastOutput>>create();
 273         checkStatusAndMakeToast(input, futureResult, maxMakeToastTries);
 274         return futureResult;
 275     }
 276
 277     private static RpcError makeToasterOutOfBreadError() {
 278         return RpcResultBuilder.newError(APPLICATION, ErrorTag.RESOURCE_DENIED, "Toaster is out of bread",
 279             "out-of-stock", null, null);
 280     }
 281
 282     private static RpcError makeToasterInUseError() {
 283         return RpcResultBuilder.newWarning(APPLICATION, ErrorTag.IN_USE, "Toaster is busy", null, null, null);
 284     }
 285
 286     private void checkStatusAndMakeToast(final MakeToastInput input,
 287             final SettableFuture<RpcResult<MakeToastOutput>> futureResult, final int tries) {
 288         // Read the ToasterStatus and, if currently Up, try to write the status to Down.
 289         // If that succeeds, then we essentially have an exclusive lock and can proceed
 290         // to make toast.
 291         final ReadWriteTransaction tx = dataBroker.newReadWriteTransaction();
 292         FluentFuture<Optional<Toaster>> readFuture = tx.read(OPERATIONAL, TOASTER_IID);
 293
 294         final ListenableFuture<? extends CommitInfo> commitFuture =
 295             Futures.transformAsync(readFuture, toasterData -> {
 296                 ToasterStatus toasterStatus = ToasterStatus.Up;
 297                 if (toasterData.isPresent()) {
 298                     toasterStatus = toasterData.orElseThrow().getToasterStatus();
 299                 }
 300
 301                 LOG.debug("Read toaster status: {}", toasterStatus);
 302
 303                 if (toasterStatus == ToasterStatus.Up) {
 304
 305                     if (outOfBread()) {
 306                         LOG.debug("Toaster is out of bread");
 307                         tx.cancel();
 308                         return Futures.immediateFailedFuture(
 309                                 new TransactionCommitFailedException("", makeToasterOutOfBreadError()));
 310                     }
 311
 312                     LOG.debug("Setting Toaster status to Down");
 313
 314                     // We're not currently making toast - try to update the status to Down
 315                     // to indicate we're going to make toast. This acts as a lock to prevent
 316                     // concurrent toasting.
 317                     tx.put(OPERATIONAL, TOASTER_IID, buildToaster(ToasterStatus.Down));
 318                     return tx.commit();
 319                 }
 320
 321                 LOG.debug("Oops - already making toast!");
 322
 323                 // Return an error since we are already making toast. This will get
 324                 // propagated to the commitFuture below which will interpret the null
 325                 // TransactionStatus in the RpcResult as an error condition.
 326                 tx.cancel();
 327                 return Futures.immediateFailedFuture(
 328                         new TransactionCommitFailedException("", makeToasterInUseError()));
 329             }, MoreExecutors.directExecutor());
 330
 331         Futures.addCallback(commitFuture, new FutureCallback<CommitInfo>() {
 332             @Override
 333             public void onSuccess(final CommitInfo result) {
 334                 // OK to make toast
 335                 currentMakeToastTask.set(executor.submit(new MakeToastTask(input, futureResult)));
 336             }
 337
 338             @Override
 339             public void onFailure(final Throwable ex) {
 340                 if (ex instanceof OptimisticLockFailedException) {
 341
 342                     // Another thread is likely trying to make toast simultaneously and updated the
 343                     // status before us. Try reading the status again - if another make toast is
 344                     // now in progress, we should get ToasterStatus.Down and fail.
 345
 346                     if (tries - 1 > 0) {
 347                         LOG.debug("Got OptimisticLockFailedException - trying again");
 348                         checkStatusAndMakeToast(input, futureResult, tries - 1);
 349                     } else {
 350                         futureResult.set(RpcResultBuilder.<MakeToastOutput>failed()
 351                                 .withError(ErrorType.APPLICATION, ex.getMessage()).build());
 352                     }
 353                 } else if (ex instanceof TransactionCommitFailedException) {
 354                     LOG.debug("Failed to commit Toaster status", ex);
 355
 356                     // Probably already making toast.
 357                     futureResult.set(RpcResultBuilder.<MakeToastOutput>failed()
 358                             .withRpcErrors(((TransactionCommitFailedException)ex).getErrorList()).build());
 359                 } else {
 360                     LOG.debug("Unexpected error committing Toaster status", ex);
 361                     futureResult.set(RpcResultBuilder.<MakeToastOutput>failed().withError(ErrorType.APPLICATION,
 362                             "Unexpected error committing Toaster status", ex).build());
 363                 }
 364             }
 365         }, MoreExecutors.directExecutor());
 366     }
 367
 368     /**
 369      * RestConf RPC call implemented from the ToasterService interface.
 370      * Restocks the bread for the toaster, resets the toastsMade counter to 0, and sends a
 371      * ToasterRestocked notification.
 372      */
 373     private ListenableFuture<RpcResult<RestockToasterOutput>> restockToaster(final RestockToasterInput input) {
 374         LOG.info("restockToaster: {}", input);
 375
 376         amountOfBreadInStock.set(input.getAmountOfBreadToStock().toJava());
 377
 378         if (amountOfBreadInStock.get() > 0) {
 379             ToasterRestocked reStockedNotification = new ToasterRestockedBuilder()
 380                     .setAmountOfBread(input.getAmountOfBreadToStock()).build();
 381             notificationProvider.offerNotification(reStockedNotification);
 382         }
 383
 384         return Futures.immediateFuture(RpcResultBuilder.success(EMPTY_RESTOCK_OUTPUT).build());
 385     }
 386
 387     /**
 388      * JMX RPC call implemented from the ToasterProviderRuntimeMXBean interface.
 389      */
 390     @Override
 391     public void clearToastsMade() {
 392         LOG.info("clearToastsMade");
 393         toastsMade.set(0);
 394     }
 395
 396     /**
 397      * Accesssor method implemented from the ToasterProviderRuntimeMXBean interface.
 398      */
 399     @Override
 400     public Long getToastsMade() {
 401         return toastsMade.get();
 402     }
 403
 404     private ListenableFuture<?> setToasterStatusUp(final Function<Boolean, MakeToastOutput> resultCallback) {
 405         WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
 406         tx.put(OPERATIONAL,TOASTER_IID, buildToaster(ToasterStatus.Up));
 407
 408         final var future = tx.commit();
 409         Futures.addCallback(future, new FutureCallback<CommitInfo>() {
 410             @Override
 411             public void onSuccess(final CommitInfo result) {
 412                 LOG.info("Successfully set ToasterStatus to Up");
 413                 notifyCallback(true);
 414             }
 415
 416             @Override
 417             public void onFailure(final Throwable failure) {
 418                 // We shouldn't get an OptimisticLockFailedException (or any ex) as no
 419                 // other component should be updating the operational state.
 420                 LOG.error("Failed to update toaster status", failure);
 421
 422                 notifyCallback(false);
 423             }
 424
 425             void notifyCallback(final boolean result) {
 426                 if (resultCallback != null) {
 427                     resultCallback.apply(result);
 428                 }
 429             }
 430         }, MoreExecutors.directExecutor());
 431
 432         return future;
 433     }
 434
 435     private boolean outOfBread() {
 436         return amountOfBreadInStock.get() == 0;
 437     }
 438
 439     private class MakeToastTask implements Callable<Void> {
 440
 441         final MakeToastInput toastRequest;
 442         final SettableFuture<RpcResult<MakeToastOutput>> futureResult;
 443
 444         MakeToastTask(final MakeToastInput toastRequest,
 445             final SettableFuture<RpcResult<MakeToastOutput>> futureResult) {
 446             this.toastRequest = toastRequest;
 447             this.futureResult = futureResult;
 448         }
 449
 450         @Override
 451         public Void call() {
 452             try {
 453                 // make toast just sleeps for n seconds per doneness level.
 454                 Thread.sleep(darknessFactor.get()
 455                         * toastRequest.getToasterDoneness().toJava());
 456
 457             } catch (InterruptedException e) {
 458                 LOG.info("Interrupted while making the toast");
 459             }
 460
 461             toastsMade.incrementAndGet();
 462
 463             amountOfBreadInStock.getAndDecrement();
 464             if (outOfBread()) {
 465                 LOG.info("Toaster is out of bread!");
 466
 467                 notificationProvider.offerNotification(new ToasterOutOfBreadBuilder().build());
 468             }
 469
 470             // Set the Toaster status back to up - this essentially releases the toasting lock.
 471             // We can't clear the current toast task nor set the Future result until the
 472             // update has been committed so we pass a callback to be notified on completion.
 473
 474             setToasterStatusUp(result -> {
 475                 currentMakeToastTask.set(null);
 476                 LOG.debug("Toast done");
 477                 futureResult.set(RpcResultBuilder.success(EMPTY_MAKE_OUTPUT).build());
 478                 return null;
 479             });
 480
 481             return null;
 482         }
 483     }
 484 }