* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.clustering.it.provider.impl;
-import com.google.common.base.Preconditions;
+import static java.util.Objects.requireNonNull;
+
+import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.HashSet;
import java.util.Set;
import java.util.SplittableRandom;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ProduceTransactionsHandler extends AbstractTransactionHandler {
+@Deprecated(forRemoval = true)
+public final class ProduceTransactionsHandler extends AbstractTransactionHandler {
private static final Logger LOG = LoggerFactory.getLogger(ProduceTransactionsHandler.class);
private final SettableFuture<RpcResult<ProduceTransactionsOutput>> future = SettableFuture.create();
private ProduceTransactionsHandler(final DOMDataTreeProducer producer, final DOMDataTreeIdentifier idListItem,
final ProduceTransactionsInput input) {
super(input);
- this.itemProducer = Preconditions.checkNotNull(producer);
- this.idListItem = Preconditions.checkNotNull(idListItem);
+ this.itemProducer = requireNonNull(producer);
+ this.idListItem = requireNonNull(idListItem);
}
public static ListenableFuture<RpcResult<ProduceTransactionsOutput>> start(
final String id = input.getId();
LOG.debug("Filling the item list {} with initial values.", id);
- final YangInstanceIdentifier idListWithKey = ID_INT_YID.node(new NodeIdentifierWithPredicates(ID_INT, ID, id));
+ final YangInstanceIdentifier idListWithKey = ID_INT_YID.node(NodeIdentifierWithPredicates.of(ID_INT, ID, id));
final DOMDataTreeProducer itemProducer = domDataTreeService.createProducer(
Collections.singleton(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, idListWithKey)));
cursor.close();
try {
- tx.submit().checkedGet(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- } catch (final Exception e) {
+ tx.commit().get(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException | ExecutionException | TimeoutException e) {
LOG.warn("Unable to fill the initial item list.", e);
-
- try {
- itemProducer.close();
- } catch (final DOMDataTreeProducerException exception) {
- LOG.warn("Failure while closing producer.", exception);
- }
+ closeProducer(itemProducer);
return Futures.immediateFuture(RpcResultBuilder.<ProduceTransactionsOutput>failed()
.withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
final ProduceTransactionsHandler handler = new ProduceTransactionsHandler(itemProducer,
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, idListWithKey.node(list.getIdentifier())
.toOptimized()), input);
+ // It is handler's responsibility to close itemProducer when the work is finished.
handler.doStart();
return handler.future;
}
+ private static void closeProducer(final DOMDataTreeProducer producer) {
+ try {
+ producer.close();
+ } catch (final DOMDataTreeProducerException exception) {
+ LOG.warn("Failure while closing producer.", exception);
+ }
+ }
+
@Override
- ListenableFuture<Void> execWrite(final long txId) {
+ FluentFuture<? extends @NonNull CommitInfo> execWrite(final long txId) {
final int i = random.nextInt(MAX_ITEM + 1);
final DOMDataTreeCursorAwareTransaction tx = itemProducer.createTransaction(false);
final DOMDataTreeWriteCursor cursor = tx.createCursor(idListItem);
- final NodeIdentifierWithPredicates entryId = new NodeIdentifierWithPredicates(ITEM, NUMBER, i);
+ final NodeIdentifierWithPredicates entryId = NodeIdentifierWithPredicates.of(ITEM, NUMBER, i);
if (usedValues.contains(i)) {
LOG.debug("Deleting item: {}", i);
deleteTx++;
cursor.close();
- return tx.submit();
+ return tx.commit();
}
@Override
- void runFailed(final Throwable cause) {
+ void runFailed(final Throwable cause, final long txId) {
+ closeProducer(itemProducer);
future.set(RpcResultBuilder.<ProduceTransactionsOutput>failed()
- .withError(RpcError.ErrorType.APPLICATION, "Submit failed", cause).build());
+ .withError(RpcError.ErrorType.APPLICATION, "Commit failed for tx # " + txId, cause).build());
}
@Override
void runSuccessful(final long allTx) {
+ closeProducer(itemProducer);
final ProduceTransactionsOutput output = new ProduceTransactionsOutputBuilder()
.setAllTx(allTx)
.setInsertTx(insertTx)
}
@Override
- void runTimedOut(final Exception cause) {
+ void runTimedOut(final String cause) {
+ closeProducer(itemProducer);
future.set(RpcResultBuilder.<ProduceTransactionsOutput>failed()
- .withError(RpcError.ErrorType.APPLICATION,
- "Final submit was timed out by the test provider or was interrupted", cause).build());
+ .withError(RpcError.ErrorType.APPLICATION, cause).build());
}
}