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

   1 /*
   2  * Copyright (c) 2015 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.cluster.datastore;
   9
  10 import akka.actor.ActorSelection;
  11 import akka.dispatch.Mapper;
  12 import com.google.common.base.Preconditions;
  13 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  14 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  15 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  16 import org.slf4j.Logger;
  17 import org.slf4j.LoggerFactory;
  18 import scala.concurrent.Future;
  19
  20 /**
  21  * A {@link Mapper} extracting the {@link ActorSelection} pointing to the actor which
  22  * is backing a particular transaction.
  23  * <p/>
  24  * This class is not for general consumption. It is public only to support the pre-lithium compatibility
  25  * package.
  26  * TODO: once we remove compatibility, make this class package-private and final.
  27  */
  28 public class TransactionReadyReplyMapper extends Mapper<Object, ActorSelection> {
  29     protected static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER = new Mapper<Throwable, Throwable>() {
  30         @Override
  31         public Throwable apply(final Throwable failure) {
  32             return failure;
  33         }
  34     };
  35     private static final Logger LOG = LoggerFactory.getLogger(TransactionReadyReplyMapper.class);
  36     private final TransactionIdentifier identifier;
  37     private final ActorContext actorContext;
  38
  39     protected TransactionReadyReplyMapper(final ActorContext actorContext, final TransactionIdentifier identifier) {
  40         this.actorContext = Preconditions.checkNotNull(actorContext);
  41         this.identifier = Preconditions.checkNotNull(identifier);
  42     }
  43
  44     protected final ActorContext getActorContext() {
  45         return actorContext;
  46     }
  47
  48     protected String extractCohortPathFrom(final ReadyTransactionReply readyTxReply) {
  49         return readyTxReply.getCohortPath();
  50     }
  51
  52     @Override
  53     public final ActorSelection checkedApply(final Object serializedReadyReply) {
  54         LOG.debug("Tx {} readyTransaction", identifier);
  55
  56         // At this point the ready operation succeeded and we need to extract the cohort
  57         // actor path from the reply.
  58         if (ReadyTransactionReply.isSerializedType(serializedReadyReply)) {
  59             ReadyTransactionReply readyTxReply = ReadyTransactionReply.fromSerializable(serializedReadyReply);
  60             return actorContext.actorSelection(extractCohortPathFrom(readyTxReply));
  61         }
  62
  63         // Throwing an exception here will fail the Future.
  64         throw new IllegalArgumentException(String.format("%s: Invalid reply type %s",
  65                 identifier, serializedReadyReply.getClass()));
  66     }
  67
  68     static Future<ActorSelection> transform(final Future<Object> readyReplyFuture, final ActorContext actorContext,
  69             final TransactionIdentifier identifier) {
  70         return readyReplyFuture.transform(new TransactionReadyReplyMapper(actorContext, identifier),
  71             SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher());
  72     }
  73 }