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