openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/BarrierTaskBuilder.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
   9 package org.opendaylight.openflowplugin.impl.device;
  10
  11 import com.google.common.base.Preconditions;
  12 import com.google.common.util.concurrent.FutureCallback;
  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 io.netty.util.Timeout;
  17 import io.netty.util.TimerTask;
  18 import java.util.concurrent.Future;
  19 import java.util.concurrent.TimeUnit;
  20 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInputBuilder;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierOutput;
  24 import org.opendaylight.yangtools.yang.common.RpcError;
  25 import org.opendaylight.yangtools.yang.common.RpcResult;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 /**
  30  * openflowplugin-impl
  31  * org.opendaylight.openflowplugin.impl.device
  32  * Barrier message self restarting builder.
  33  *
  34  * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
  35  *         Created: Apr 3, 2015
  36  */
  37 public class BarrierTaskBuilder {
  38
  39     private static final Logger LOG = LoggerFactory.getLogger(BarrierTaskBuilder.class);
  40     public static final long DELAY = 500L;
  41
  42     private final DeviceContext deviceCtx;
  43
  44     public BarrierTaskBuilder(final DeviceContext deviceCtx) {
  45         this.deviceCtx = Preconditions.checkNotNull(deviceCtx);
  46         Preconditions.checkNotNull(deviceCtx.getTimer());
  47     }
  48
  49     public void buildAndFireBarrierTask() {
  50         Timeout timeout = deviceCtx.getTimer().newTimeout(new BarrierTask(deviceCtx), DELAY, TimeUnit.MILLISECONDS);
  51         deviceCtx.setCurrentBarrierTimeout(timeout);
  52     }
  53
  54
  55     private final class BarrierTask implements TimerTask {
  56
  57         private final DeviceContext deviceCtx;
  58
  59         public BarrierTask(final DeviceContext deviceCtx) {
  60             this.deviceCtx = deviceCtx;
  61         }
  62
  63         /**
  64          * @return OF-message, ready to send
  65          */
  66         private BarrierInput makeBarrier() {
  67             final BarrierInputBuilder biBuilder = new BarrierInputBuilder();
  68             biBuilder.setVersion(deviceCtx.getDeviceState().getVersion());
  69             biBuilder.setXid(deviceCtx.getNextXid().getValue());
  70             return biBuilder.build();
  71         }
  72
  73         @Override
  74         public void run(final Timeout timeout) throws Exception {
  75             // check outstanding requests first
  76             if (deviceCtx.getDeviceState().isValid()) {
  77                 if (deviceCtx.getNumberOfOutstandingRequests() > 0) {
  78                     BarrierInput barrierInput = makeBarrier();
  79                     LOG.trace("sending out barrier [{}]", barrierInput.getXid());
  80                     synchronized (deviceCtx) {
  81                         final Future<RpcResult<BarrierOutput>> future = deviceCtx.getPrimaryConnectionContext()
  82                                 .getConnectionAdapter().barrier(barrierInput);
  83                         final ListenableFuture<RpcResult<BarrierOutput>> lsFuture = JdkFutureAdapters.listenInPoolThread(future);
  84                         Futures.addCallback(lsFuture, makeCallBack());
  85                     }
  86                 } else {
  87                     // if no requests
  88                     buildAndFireBarrierTask();
  89                 }
  90             } else {
  91                 LOG.trace("DeviceContext is not valid, will not create next barrier task.");
  92             }
  93         }
  94
  95         private FutureCallback<RpcResult<BarrierOutput>> makeCallBack() {
  96             return new FutureCallback<RpcResult<BarrierOutput>>() {
  97                 @Override
  98                 public void onSuccess(final RpcResult<BarrierOutput> result) {
  99                     synchronized (deviceCtx) {
 100                         if (!result.isSuccessful()) {
 101                             for (RpcError rpcError : result.getErrors()) {
 102                                 LOG.trace("Barrier response with error {}", rpcError, rpcError.getCause());
 103                             }
 104                         } else if (null != result.getResult().getXid()) {
 105                             BarrierProcessor.processOutstandingRequests(result.getResult().getXid(), deviceCtx);
 106                         }
 107                         buildAndFireBarrierTask();
 108                     }
 109                 }
 110
 111                 @Override
 112                 public void onFailure(final Throwable t) {
 113                     LOG.info("Barrier has failed {} ", t.getMessage());
 114                     LOG.trace("Barrier has failed", t);
 115                 }
 116             };
 117         }
 118
 119     }
 120
 121 }