Java 类com.google.common.util.concurrent.CheckedFuture 实例源码

private CheckedFuture<Void, TransactionCommitFailedException> listenForFailure(
        final WriteTransaction tx, final CheckedFuture<Void, TransactionCommitFailedException> future) {
    Futures.addCallback(future, new FutureCallback<Void>() {
        @Override
        public void onFailure(final Throwable t) {
            failTransactionChain(tx,t);
        }

        @Override
        public void onSuccess(final Void result) {
            // Intentionally NOOP
        }
    });

    return future;
}

@Override
public CheckedFuture<Void, TransactionCommitFailedException> submit(
        final DOMDataWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
    checkNotFailed();
    checkNotClosed();

    final CheckedFuture<Void, TransactionCommitFailedException> ret = broker.submit(transaction, cohorts);

    COUNTER_UPDATER.incrementAndGet(this);
    Futures.addCallback(ret, new FutureCallback<Void>() {
        @Override
        public void onSuccess(final Void result) {
            transactionCompleted();
        }

        @Override
        public void onFailure(final Throwable t) {
            transactionFailed(transaction, t);
        }
    });

    return ret;
}

private String readFromGreetingRegistry(HelloWorldInput input) {
    String result = "Hello " + input.getName();
    ReadOnlyTransaction transaction = db.newReadOnlyTransaction();
    InstanceIdentifier<GreetingRegistryEntry> iid = toInstanceIdentifier(input);
    CheckedFuture<Optional<GreetingRegistryEntry>, ReadFailedException> future =
            transaction.read(LogicalDatastoreType.CONFIGURATION, iid);
    Optional<GreetingRegistryEntry> optional = Optional.absent();
    try {
        optional = future.checkedGet();
    } catch (ReadFailedException e) {
        LOG.warn("Reading greeting failed:",e);
    }
    if(optional.isPresent()) {
        result = optional.get().getGreeting();
    }
    return result;
}

/**
 * Tests that the {@link CheckedFuture#checkedGet()} method throws the correct
 * type of cancellation exception when it is cancelled.
 */
public void testCheckedGetThrowsApplicationExceptionOnCancellation() {

  final CheckedFuture<Boolean, ?> future =
      createCheckedFuture(Boolean.TRUE, null, latch);

  assertFalse(future.isDone());
  assertFalse(future.isCancelled());

  new Thread(new Runnable() {
    @Override
    public void run() {
      future.cancel(true);
    }
  }).start();

  try {
    future.checkedGet();
    fail("RPC Should have been cancelled.");
  } catch (Exception e) {
    checkCancelledException(e);
  }

  assertTrue(future.isDone());
  assertTrue(future.isCancelled());
}

@Override
public void executeList() {
    final LogicalDatastoreType dsType = getDataStoreType();
    final org.opendaylight.yangtools.yang.common.QName olId = QName.create(OuterList.QNAME, "id");
    final YangInstanceIdentifier pid =
            YangInstanceIdentifier.builder().node(TestExec.QNAME).node(OuterList.QNAME).build();

    try (DOMDataReadOnlyTransaction tx = domDataBroker.newReadOnlyTransaction()) {
        for (int l = 0; l < outerListElem; l++) {
            YangInstanceIdentifier yid = pid.node(new NodeIdentifierWithPredicates(OuterList.QNAME, olId, l));
            Optional<NormalizedNode<?,?>> optionalDataObject;
            CheckedFuture<Optional<NormalizedNode<?,?>>, ReadFailedException> submitFuture = tx.read(dsType, yid);
            try {
                optionalDataObject = submitFuture.checkedGet();
                if (optionalDataObject != null && optionalDataObject.isPresent()) {
                    txOk++;
                }
            } catch (final ReadFailedException e) {
                LOG.warn("failed to ....", e);
                txError++;
            }
        }
    }
}

@Override
public synchronized CheckedFuture<Void, TransactionCommitFailedException> submit() {
    Preconditions.checkState(!closed, "Transaction %s is already closed", identifier);

    final Set<DOMStoreWriteTransaction> txns = ImmutableSet.copyOf(idToTransaction.values());
    final List<DOMStoreThreePhaseCommitCohort> cohorts = new ArrayList<>(txns.size());
    for (DOMStoreWriteTransaction tx : txns) {
        cohorts.add(tx.ready());
    }

    try {
        return Futures.immediateCheckedFuture(new CommitCoordinationTask(this, cohorts, null).call());
    } catch (TransactionCommitFailedException e) {
        return Futures.immediateFailedCheckedFuture(e);
    }
}

/**
 * This test method invokes and executes the remote rpc.
 */
@Test
public void testInvokeRpcWithNullInput() throws Exception {
    final ContainerNode rpcOutput = makeRPCOutput("bar");
    final DOMRpcResult rpcResult = new DefaultDOMRpcResult(rpcOutput);

    @SuppressWarnings({"unchecked", "rawtypes"})
    final ArgumentCaptor<NormalizedNode<?, ?>> inputCaptor =
            (ArgumentCaptor) ArgumentCaptor.forClass(NormalizedNode.class);

    when(domRpcService2.invokeRpc(eq(TEST_RPC_TYPE), inputCaptor.capture())).thenReturn(
            Futures.<DOMRpcResult, DOMRpcException>immediateCheckedFuture(rpcResult));

    final CheckedFuture<DOMRpcResult, DOMRpcException> frontEndFuture =
            remoteRpcImpl1.invokeRpc(TEST_RPC_ID, null);
    assertTrue(frontEndFuture instanceof RemoteDOMRpcFuture);

    final DOMRpcResult result = frontEndFuture.checkedGet(5, TimeUnit.SECONDS);
    assertEquals(rpcOutput, result.getResult());
}

CheckedFuture<DOMRpcResult, DOMRpcException> invokeRpc(final SchemaPath type, final NormalizedNode<?, ?> input) {
    final AbstractDOMRpcRoutingTableEntry entry = rpcs.get(type);
    if (entry == null) {
        return Futures.<DOMRpcResult, DOMRpcException>immediateFailedCheckedFuture(
            new DOMRpcImplementationNotAvailableException("No implementation of RPC %s available", type));
    }

    return entry.invokeRpc(input);
}

@Nonnull
@Override
public CheckedFuture<DOMRpcResult, DOMRpcException> invokeRpc(@Nonnull DOMRpcIdentifier rpc, @Nullable NormalizedNode<?, ?> input) {
    LOG.debug("get-singleton-constant invoked, current value: {}", constant);

    final LeafNode<Object> value = ImmutableLeafNodeBuilder.create()
            .withNodeIdentifier(new NodeIdentifier(CONSTANT))
            .withValue(constant)
            .build();

    final ContainerNode result = ImmutableContainerNodeBuilder.create()
            .withNodeIdentifier(new NodeIdentifier(OUTPUT))
            .withChild(value)
            .build();

    return Futures.immediateCheckedFuture(new DefaultDOMRpcResult(result));
}

@Override
public CheckedFuture<Optional<NormalizedNode<?,?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
    LOG.debug("Tx: {} Read: {}", getIdentifier(), path);
    checkNotNull(path, "Path must not be null.");

    final DataTreeSnapshot snapshot = stableSnapshot;
    if (snapshot == null) {
        return Futures.immediateFailedCheckedFuture(new ReadFailedException("Transaction is closed"));
    }

    try {
        return Futures.immediateCheckedFuture(snapshot.readNode(path));
    } catch (Exception e) {
        LOG.error("Tx: {} Failed Read of {}", getIdentifier(), path, e);
        return Futures.immediateFailedCheckedFuture(new ReadFailedException("Read failed",e));
    }
}

private <T> CheckedFuture<T, ReadFailedException> executeRead(final String shardName,
        final AbstractRead<T> readCmd) {
    Preconditions.checkState(type != TransactionType.WRITE_ONLY,
            "Reads from write-only transactions are not allowed");

    LOG.debug("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath());

    final SettableFuture<T> proxyFuture = SettableFuture.create();
    TransactionContextWrapper contextWrapper = getContextWrapper(shardName);
    contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
        @Override
        public void invoke(final TransactionContext transactionContext) {
            transactionContext.executeRead(readCmd, proxyFuture);
        }
    });

    return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
}

@Override
public CheckedFuture<YangTextSchemaSource, SchemaSourceException> getSource(SourceIdentifier sourceIdentifier) {
    LOG.trace("Getting yang schema source for {}", sourceIdentifier.getName());

    Future<YangTextSchemaSourceSerializationProxy> result = remoteRepo.getYangTextSchemaSource(sourceIdentifier);

    final SettableFuture<YangTextSchemaSource> res = SettableFuture.create();
    result.onComplete(new OnComplete<YangTextSchemaSourceSerializationProxy>() {
        @Override
        public void onComplete(Throwable throwable,
                YangTextSchemaSourceSerializationProxy yangTextSchemaSourceSerializationProxy) {
            if (yangTextSchemaSourceSerializationProxy != null) {
                res.set(yangTextSchemaSourceSerializationProxy.getRepresentation());
            }
            if (throwable != null) {
                res.setException(throwable);
            }
        }

    }, executionContext);

    return Futures.makeChecked(res, MAPPER);
}

/**
 * This test method invokes and executes the remote rpc.
 */
@Test(expected = DOMRpcException.class)
public void testInvokeRpcWithRemoteFailedFuture() throws Exception {
    final NormalizedNode<?, ?> invokeRpcInput = makeRPCInput("foo");
    @SuppressWarnings({"unchecked", "rawtypes"})
    final ArgumentCaptor<NormalizedNode<?, ?>> inputCaptor =
            (ArgumentCaptor) ArgumentCaptor.forClass(NormalizedNode.class);

    when(domRpcService2.invokeRpc(eq(TEST_RPC_TYPE), inputCaptor.capture())).thenReturn(
            Futures.<DOMRpcResult, DOMRpcException>immediateFailedCheckedFuture(new RemoteDOMRpcException(
                    "Test Exception", null)));

    final CheckedFuture<DOMRpcResult, DOMRpcException> frontEndFuture =
            remoteRpcImpl1.invokeRpc(TEST_RPC_ID, invokeRpcInput);
    assertTrue(frontEndFuture instanceof RemoteDOMRpcFuture);
    frontEndFuture.checkedGet(5, TimeUnit.SECONDS);
}

/**
 * A simple unbound producer. It write some basic things into the data store based on the
 * test model.
 * @throws DOMDataTreeProducerException
 * @throws TransactionCommitFailedException
 */
@Test
public final void testBasicProducer() throws DOMDataTreeProducerException, TransactionCommitFailedException {
    // Create a producer. It is an AutoCloseable resource, hence the try-with pattern
    try (final DOMDataTreeProducer prod = service().createProducer(Collections.singleton(UNORDERED_CONTAINER_TREE))) {
        assertNotNull(prod);

        final DOMDataWriteTransaction tx = prod.createTransaction(true);
        assertNotNull(tx);

        tx.put(LogicalDatastoreType.OPERATIONAL, UNORDERED_CONTAINER_IID, ImmutableContainerNodeBuilder.create().build());

        final CheckedFuture<Void, TransactionCommitFailedException> f = tx.submit();
        assertNotNull(f);

        f.checkedGet();
    }
}

@Override
public CheckedFuture<Void, TransactionCommitFailedException> submit() {
    final AbstractDOMTransactionFactory<?> impl = IMPL_UPDATER.getAndSet(this, null);
    checkRunning(impl);

    final Collection<T> txns = getSubtransactions();
    final Collection<DOMStoreThreePhaseCommitCohort> cohorts = new ArrayList<>(txns.size());

    // FIXME: deal with errors thrown by backed (ready and submit can fail in theory)
    for (final T txn : txns) {
        cohorts.add(txn.ready());
    }

    final CheckedFuture<Void, TransactionCommitFailedException> ret = impl.submit(this, cohorts);
    FUTURE_UPDATER.lazySet(this, ret);
    return ret;
}

@Override
public CheckedFuture<Void, TransactionCommitFailedException> submit(
        final DOMDataWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
    checkNotFailed();
    checkNotClosed();

    final CheckedFuture<Void, TransactionCommitFailedException> ret = broker.submit(transaction, cohorts);

    COUNTER_UPDATER.incrementAndGet(this);
    Futures.addCallback(ret, new FutureCallback<Void>() {
        @Override
        public void onSuccess(final Void result) {
            transactionCompleted();
        }

        @Override
        public void onFailure(final Throwable failure) {
            transactionFailed(transaction, failure);
        }
    }, MoreExecutors.directExecutor());

    return ret;
}

@Override
protected CheckedFuture<Void, TransactionCommitFailedException> submit(final DOMDataWriteTransaction transaction,
        final Collection<DOMStoreThreePhaseCommitCohort> cohorts) {

    Preconditions.checkArgument(transaction != null, "Transaction must not be null.");
    Preconditions.checkArgument(cohorts != null, "Cohorts must not be null.");
    LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier());

    if (cohorts.isEmpty()) {
        return Futures.immediateCheckedFuture(null);
    }

    final AsyncNotifyingSettableFuture clientSubmitFuture =
            new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor);

    doCanCommit(clientSubmitFuture, transaction, cohorts);

    return MappingCheckedFuture.create(clientSubmitFuture, COMMIT_ERROR_MAPPER);
}

/**
 * This test method invokes and executes the remote rpc.
 */
@Test
public void testInvokeRpc() throws Exception {
    final ContainerNode rpcOutput = makeRPCOutput("bar");
    final DOMRpcResult rpcResult = new DefaultDOMRpcResult(rpcOutput);

    final NormalizedNode<?, ?> invokeRpcInput = makeRPCInput("foo");
    @SuppressWarnings({"unchecked", "rawtypes"})
    final ArgumentCaptor<NormalizedNode<?, ?>> inputCaptor =
            (ArgumentCaptor) ArgumentCaptor.forClass(NormalizedNode.class);

    when(domRpcService2.invokeRpc(eq(TEST_RPC_TYPE), inputCaptor.capture())).thenReturn(
            Futures.<DOMRpcResult, DOMRpcException>immediateCheckedFuture(rpcResult));

    final CheckedFuture<DOMRpcResult, DOMRpcException> frontEndFuture =
            remoteRpcImpl1.invokeRpc(TEST_RPC_ID, invokeRpcInput);
    assertTrue(frontEndFuture instanceof RemoteDOMRpcFuture);

    final DOMRpcResult result = frontEndFuture.checkedGet(5, TimeUnit.SECONDS);
    assertEquals(rpcOutput, result.getResult());
}

@Test
public void testSubmitWithNegativeCanCommitResponse() throws Exception {
    doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();

    doReturn(Futures.immediateFuture(false)).when(mockCohort2).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();

    DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
    doReturn(Futures.immediateFuture(false)).when(mockCohort3).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();

    CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
            transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));

    assertFailure(future, null, mockCohort1, mockCohort2, mockCohort3);
}

@Override
public CheckedFuture<PostCanCommitStep, DataValidationFailedException> canCommit(final Object txId,
        final DOMDataTreeCandidate candidate, final SchemaContext ctx) {

    // Simple data validation - verify the year, if present, is >= 1990

    final DataTreeCandidateNode rootNode = candidate.getRootNode();
    final Optional<NormalizedNode<?, ?>> dataAfter = rootNode.getDataAfter();

    LOG.info("In canCommit: modificationType: {}, dataBefore: {}, dataAfter: {}", rootNode.getModificationType(),
            rootNode.getDataBefore(), dataAfter);

    // Note: we don't want to process DELETE modifications but we don't need to explicitly check the
    // ModificationType because dataAfter will not be present. Also dataAfter *should* always contain a
    // MapEntryNode but we verify anyway.
    if (dataAfter.isPresent()) {
        final NormalizedNode<?, ?> normalizedNode = dataAfter.get();
        Verify.verify(normalizedNode instanceof DataContainerNode, "Expected type DataContainerNode, actual was %s",
                normalizedNode.getClass());
        DataContainerNode<?> entryNode = (DataContainerNode<?>) normalizedNode;
        final Optional<DataContainerChild<? extends PathArgument, ?>> possibleYear =
                entryNode.getChild(YEAR_NODE_ID);
        if (possibleYear.isPresent()) {
            final Number year = (Number) possibleYear.get().getValue();

            LOG.info("year is {}", year);

            if (!(year.longValue() >= 1990)) {
                return Futures.immediateFailedCheckedFuture(new DataValidationFailedException(
                        DOMDataTreeIdentifier.class, candidate.getRootPath(),
                            String.format("Invalid year %d - year must be >= 1990", year)));
            }
        }
    }

    // Return the noop PostCanCommitStep as we're only validating input data and not participating in the
    // remaining 3PC stages (pre-commit and commit).
    return PostCanCommitStep.NOOP_SUCCESS_FUTURE;
}

public void testCheckedGetThrowsApplicationExceptionOnInterruption()
    throws InterruptedException {

  final CheckedFuture<Boolean, ?> future =
      createCheckedFuture(Boolean.TRUE, null, latch);

  final CountDownLatch startingGate = new CountDownLatch(1);
  final CountDownLatch successLatch = new CountDownLatch(1);

  assertFalse(future.isDone());
  assertFalse(future.isCancelled());

  Thread getThread = new Thread(new Runnable() {
    @Override
    public void run() {
      startingGate.countDown();

      try {
        future.checkedGet();
      } catch (Exception e) {
        checkInterruptedException(e);

        // This only gets hit if the original call throws an exception and
        // the check call above passes.
        successLatch.countDown();
      }
    }
  });
  getThread.start();

  assertTrue(startingGate.await(500, TimeUnit.MILLISECONDS));
  getThread.interrupt();

  assertTrue(successLatch.await(500, TimeUnit.MILLISECONDS));

  assertFalse(future.isDone());
  assertFalse(future.isCancelled());
}

private void initializeDataTree(DataBroker db) {
    LOG.info("Preparing to initialize the greeting registry");
    WriteTransaction transaction = db.newWriteOnlyTransaction();
    InstanceIdentifier<GreetingRegistry> iid = InstanceIdentifier.create(GreetingRegistry.class);
    GreetingRegistry greetingRegistry = new GreetingRegistryBuilder()
            .build();
    transaction.put(LogicalDatastoreType.OPERATIONAL, iid, greetingRegistry);
    CheckedFuture<Void, TransactionCommitFailedException> future = transaction.submit();
    Futures.addCallback(future, new LoggingFuturesCallBack<>("Failed to create greeting registry", LOG));
}

private void writeToGreetingRegistry(HelloWorldInput input, HelloWorldOutput output) {
    WriteTransaction transaction = db.newWriteOnlyTransaction();
    InstanceIdentifier<GreetingRegistryEntry> iid = toInstanceIdentifier(input);
    GreetingRegistryEntry greeting = new GreetingRegistryEntryBuilder()
            .setGreeting(output.getGreeting())
            .setName(input.getName())
            .build();
    transaction.put(LogicalDatastoreType.OPERATIONAL, iid, greeting);
    CheckedFuture<Void, TransactionCommitFailedException> future = transaction.submit();
    Futures.addCallback(future, new LoggingFuturesCallBack<Void>("Failed to write greeting to greeting registry", LOG));
}

private void validateGreetingRegistry(String name) {
    InstanceIdentifier<GreetingRegistryEntry> iid = InstanceIdentifier.create(GreetingRegistry.class)
            .child(GreetingRegistryEntry.class, new GreetingRegistryEntryKey(name));
    DataBroker db = getSession().getSALService(DataBroker.class);
    ReadOnlyTransaction transaction = db.newReadOnlyTransaction();
    CheckedFuture<Optional<GreetingRegistryEntry>, ReadFailedException> future =
            transaction.read(LogicalDatastoreType.OPERATIONAL, iid);
    Optional<GreetingRegistryEntry> optional = Optional.absent();
    try {
        optional = future.checkedGet();
    } catch (ReadFailedException e) {
        LOG.warn("Reading greeting failed:",e);
    }
    Assert.assertTrue(name + " not recorded in greeting registry",optional.isPresent());
}

public CheckedFuture<ChannelStatusResponse, BeamException> findRelationship(BTBBeamChannel channel, BTBBeamUser user) {
    return new Channels.StatusChecker(this.beam.gson).check(this.get(
            String.format("%d/relationship", channel.id),
            ChannelStatusResponse.class,
            BeamHttpClient.getArgumentsBuilder()
                    .put("user", String.valueOf(user.id))
                .build()
    ));
}

public void testCheckedGetThrowsApplicationExceptionOnInterruption()
    throws InterruptedException {

  final CheckedFuture<Boolean, ?> future =
      createCheckedFuture(Boolean.TRUE, null, latch);

  final CountDownLatch startingGate = new CountDownLatch(1);
  final CountDownLatch successLatch = new CountDownLatch(1);

  assertFalse(future.isDone());
  assertFalse(future.isCancelled());

  Thread getThread = new Thread(new Runnable() {
    @Override
    public void run() {
      startingGate.countDown();

      try {
        future.checkedGet();
      } catch (Exception e) {
        checkInterruptedException(e);

        // This only gets hit if the original call throws an exception and
        // the check call above passes.
        successLatch.countDown();
      }
    }
  });
  getThread.start();

  assertTrue(startingGate.await(500, TimeUnit.MILLISECONDS));
  getThread.interrupt();

  assertTrue(successLatch.await(500, TimeUnit.MILLISECONDS));

  assertFalse(future.isDone());
  assertFalse(future.isCancelled());
}

@Override
public ReadWriteTransaction newReadWriteTransaction() {
    final DOMDataReadWriteTransaction delegateTx = delegate.newReadWriteTransaction();
    return new BindingDOMReadWriteTransactionAdapter(delegateTx, codec) {

        @Override
        public CheckedFuture<Void, TransactionCommitFailedException> submit() {
            return listenForFailure(this,super.submit());
        }

    };
}

public <T> CheckedFuture<T, UserException> executeAsync(final ElasticAction2<T> action){
  final ContextListenerImpl listener = new ContextListenerImpl();
  // need to cast to jersey since the core javax.ws.rs Invocation doesn't support a typed submission.
  final JerseyInvocation invocation = (JerseyInvocation) action.buildRequest(target, listener);
  final SettableFuture<T> future = SettableFuture.create();
  invocation.submit(new GenericType<T>(action.getResponseClass()), new AsyncCallback<T>(future));
  return Futures.makeChecked(future, new Function<Exception, UserException>(){
    @Override
    public UserException apply(Exception input) {
      if(input instanceof ExecutionException){
        input = (Exception) input.getCause();
      }
      return handleException(input, action, listener);
    }});
}

@Override
public void executeList() {
    final LogicalDatastoreType dsType = getDataStoreType();
    final org.opendaylight.yangtools.yang.common.QName olId = QName.create(OuterList.QNAME, "id");
    final YangInstanceIdentifier pid =
            YangInstanceIdentifier.builder().node(TestExec.QNAME).node(OuterList.QNAME).build();

    try (DOMDataReadOnlyTransaction tx = domDataBroker.newReadOnlyTransaction()) {
        for (int l = 0; l < outerListElem; l++) {
            YangInstanceIdentifier yid = pid.node(new NodeIdentifierWithPredicates(OuterList.QNAME, olId, l));
            CheckedFuture<Optional<NormalizedNode<?,?>>, ReadFailedException> submitFuture = tx.read(dsType, yid);
            try {
                Optional<NormalizedNode<?,?>> optionalDataObject = submitFuture.checkedGet();
                if (optionalDataObject != null && optionalDataObject.isPresent()) {
                    NormalizedNode<?, ?> ret = optionalDataObject.get();
                    LOG.trace("optionalDataObject is {}", ret);
                    txOk++;
                } else {
                    txError++;
                    LOG.warn("optionalDataObject is either null or .isPresent is false");
                }
            } catch (final ReadFailedException e) {
                LOG.warn("failed to ....", e);
                txError++;
            }
        }
    }
}

@Override
public CheckedFuture<Boolean, ReadFailedException> exists(final LogicalDatastoreType store,
                                                          final YangInstanceIdentifier path) {
    checkState(root != null, "A modify operation (put, merge or delete) must be performed prior to an exists operation");
    return Futures.makeChecked(Futures.transform(read(store, path),
            (Function<Optional<NormalizedNode<?, ?>>, Boolean>) Optional::isPresent),
            ReadFailedException.MAPPER);
}

@Test
public void testSubmitWithCanCommitException() throws Exception {
    doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();

    IllegalStateException cause = new IllegalStateException("mock");
    doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();

    CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
            transaction, Arrays.asList(mockCohort1, mockCohort2));

    assertFailure(future, cause, mockCohort1, mockCohort2);
}

/**
 * This test method invokes and tests exceptions when akka timeout occured
 * Currently ignored since this test with current config takes around 15 seconds to complete.
 */
@Ignore
@Test(expected = RemoteDOMRpcException.class)
public void testInvokeRpcWithAkkaTimeoutException() throws Exception {
    final NormalizedNode<?, ?> invokeRpcInput = makeRPCInput("foo");
    final CheckedFuture<DOMRpcResult, DOMRpcException> frontEndFuture =
            remoteRpcImpl1.invokeRpc(TEST_RPC_ID, invokeRpcInput);
    assertTrue(frontEndFuture instanceof RemoteDOMRpcFuture);

    frontEndFuture.checkedGet(20, TimeUnit.SECONDS);
}

private ListenableFuture<RpcResult<?>> invoke0(final SchemaPath schemaPath, final NormalizedNode<?, ?> input) {
    final CheckedFuture<DOMRpcResult, DOMRpcException> result = delegate.invokeRpc(schemaPath, input);
    if(result instanceof LazyDOMRpcResultFuture) {
        return ((LazyDOMRpcResultFuture) result).getBindingFuture();
    }

    return transformFuture(schemaPath, result, codec.getCodecFactory());
}

@Override
protected CheckedFuture<DOMRpcResult, DOMRpcException> invokeRpc(final NormalizedNode<?, ?> input) {
    final Optional<NormalizedNode<?, ?>> maybeKey = NormalizedNodes.findNode(input, keyId);

    // Routing key is present, attempt to deliver as a routed RPC
    if (maybeKey.isPresent()) {
        final NormalizedNode<?, ?> key = maybeKey.get();
        final Object value = key.getValue();
        if (value instanceof YangInstanceIdentifier) {
            final YangInstanceIdentifier iid = (YangInstanceIdentifier) value;

            // Find a DOMRpcImplementation for a specific iid
            final List<DOMRpcImplementation> specificImpls = getImplementations(iid);
            if (specificImpls != null) {
                return specificImpls.get(0).invokeRpc(DOMRpcIdentifier.create(getSchemaPath(), iid), input);
            }

            LOG.debug("No implementation for context {} found will now look for wildcard id", iid);

            // Find a DOMRpcImplementation for a wild card. Usually remote-rpc-connector would register an
            // implementation this way
            final List<DOMRpcImplementation> mayBeRemoteImpls = getImplementations(YangInstanceIdentifier.EMPTY);

            if(mayBeRemoteImpls != null){
                return mayBeRemoteImpls.get(0).invokeRpc(DOMRpcIdentifier.create(getSchemaPath(), iid), input);
            }

        } else {
            LOG.warn("Ignoring wrong context value {}", value);
        }
    }

    final List<DOMRpcImplementation> impls = getImplementations(null);
    if (impls != null) {
        return impls.get(0).invokeRpc(globalRpcId, input);
    } else {
        return Futures.<DOMRpcResult, DOMRpcException>immediateFailedCheckedFuture(new DOMRpcImplementationNotAvailableException("No implementation of RPC %s available", getSchemaPath()));
    }
}

@Test
public void testRead() throws Exception {
    final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> result = object().read(
            YangInstanceIdentifier.EMPTY);
    final Optional<NormalizedNode<?, ?>> resultData = result.get();
    Assert.assertTrue(resultData.isPresent());
    Assert.assertEquals(data, resultData.get());
}

@Test
public void testSubmitWithCanCommitDataStoreUnavailableException() throws Exception {
    doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
    NoShardLeaderException rootCause = new NoShardLeaderException("mock");
    DataStoreUnavailableException cause = new DataStoreUnavailableException(rootCause.getMessage(), rootCause);
    doReturn(Futures.immediateFailedFuture(rootCause)).when(mockCohort2).canCommit();
    doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();

    CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
        transaction, Arrays.asList(mockCohort1, mockCohort2));

    assertFailure(future, cause, mockCohort1, mockCohort2);
}

@SuppressWarnings("unchecked")
@Override
public CheckedFuture<YangTextSchemaSource, SchemaSourceException> getSource(
        final SourceIdentifier sourceIdentifier) {
    if (yangProvider == null) {
        return Futures.immediateFailedCheckedFuture(new MissingSchemaSourceException(
            "Source provider is not available", sourceIdentifier));
    }

    return (CheckedFuture<YangTextSchemaSource, SchemaSourceException>) yangProvider.getSource(sourceIdentifier);
}

private void notifyExistingNodes(final EventSourceTopology eventSourceTopology){
    LOG.debug("Notify existing nodes");
    final Pattern nodeRegex = this.nodeIdPattern;

    final ReadOnlyTransaction tx = eventSourceTopology.getDataBroker().newReadOnlyTransaction();
    final CheckedFuture<Optional<Topology>, ReadFailedException> future =
            tx.read(LogicalDatastoreType.OPERATIONAL, EventSourceTopology.EVENT_SOURCE_TOPOLOGY_PATH);

    Futures.addCallback(future, new FutureCallback<Optional<Topology>>(){

        @Override
        public void onSuccess(final Optional<Topology> data) {
            if(data.isPresent()) {
                 final List<Node> nodes = data.get().getNode();
                 if(nodes != null){
                    for (final Node node : nodes) {
                         if (nodeRegex.matcher(node.getNodeId().getValue()).matches()) {
                             notifyNode(EventSourceTopology.EVENT_SOURCE_TOPOLOGY_PATH.child(Node.class, node.getKey()));
                         }
                     }
                 }
            }
            tx.close();
        }

        @Override
        public void onFailure(final Throwable t) {
            LOG.error("Can not notify existing nodes", t);
            tx.close();
        }

    });

}

protected final <D extends DataObject> CheckedFuture<Optional<D>,ReadFailedException> doRead(
        final DOMDataReadTransaction readTx, final LogicalDatastoreType store,
        final InstanceIdentifier<D> path) {
    Preconditions.checkArgument(!path.isWildcarded(), "Invalid read of wildcarded path %s", path);

    return MappingCheckedFuture.create(
                Futures.transform(readTx.read(store, codec.toYangInstanceIdentifierBlocking(path)),
                                  codec.deserializeFunction(path)),
                ReadFailedException.MAPPER);
}

private <T> CheckedFuture<T, ReadFailedException> sendReadRequest(final AbstractReadTransactionRequest<?> request,
        final Consumer<Response<?, ?>> completer, final ListenableFuture<T> future) {
    // Check if a previous operation failed. If it has, do not bother sending anything and report a failure
    final Exception local = operationFailure;
    if (local != null) {
        return Futures.immediateFailedCheckedFuture(new ReadFailedException("Previous operation failed", local));
    }

    // Make sure we send any modifications before issuing a read
    ensureFlushedBuider();
    sendRequest(request, completer);
    return MappingCheckedFuture.create(future, ReadFailedException.MAPPER);
}