2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.openflowplugin.impl.util;
11 import com.google.common.base.Function;
12 import com.google.common.util.concurrent.AsyncFunction;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.JdkFutureAdapters;
15 import com.google.common.util.concurrent.ListenableFuture;
16 import javax.annotation.Nullable;
17 import org.apache.commons.lang3.tuple.MutablePair;
18 import org.apache.commons.lang3.tuple.Pair;
19 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.FlowCapableTransactionService;
20 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierInput;
21 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierInputBuilder;
22 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
23 import org.opendaylight.yangtools.yang.common.RpcResult;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
28 * provides barrier message chaining and factory methods
30 public final class BarrierUtil {
32 private static final Logger LOG = LoggerFactory.getLogger(BarrierUtil.class);
35 private BarrierUtil() {
36 throw new IllegalStateException("This class should not be instantiated.");
41 * chain a barrier message - regardless of previous result and use given {@link Function} to combine
42 * original result and barrier result
44 * @param <T> type of input future
45 * @param input future to chain barrier to
46 * @param nodeRef target device
47 * @param transactionService barrier service
48 * @param compositeTransform
49 * @return future holding both results (input and of the barrier)
51 public static <T> ListenableFuture<RpcResult<T>> chainBarrier(
52 final ListenableFuture<RpcResult<T>> input, final NodeRef nodeRef,
53 final FlowCapableTransactionService transactionService,
54 final Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>> compositeTransform) {
55 final MutablePair<RpcResult<T>, RpcResult<Void>> resultPair = new MutablePair<>();
57 // store input result and append barrier
58 final ListenableFuture<RpcResult<Void>> barrierResult = Futures.transform(input,
59 new AsyncFunction<RpcResult<T>, RpcResult<Void>>() {
61 public ListenableFuture<RpcResult<Void>> apply(@Nullable final RpcResult<T> interInput) throws Exception {
62 resultPair.setLeft(interInput);
63 final SendBarrierInput barrierInput = createSendBarrierInput(nodeRef);
64 return JdkFutureAdapters.listenInPoolThread(transactionService.sendBarrier(barrierInput));
67 // store barrier result and return initiated pair
68 final ListenableFuture<Pair<RpcResult<T>, RpcResult<Void>>> compositeResult = Futures.transform(
69 barrierResult, new Function<RpcResult<Void>, Pair<RpcResult<T>, RpcResult<Void>>>() {
72 public Pair<RpcResult<T>, RpcResult<Void>> apply(@Nullable final RpcResult<Void> input) {
73 resultPair.setRight(input);
77 // append assembling transform to barrier result
78 return Futures.transform(compositeResult, compositeTransform);
82 * @param nodeRef rpc routing context
83 * @return input for {@link FlowCapableTransactionService#sendBarrier(SendBarrierInput)}
85 public static SendBarrierInput createSendBarrierInput(final NodeRef nodeRef) {
86 return new SendBarrierInputBuilder()