openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/OutboundQueueEntry.java

   1 /*
   2  * Copyright (c) 2015 Pantheon Technologies s.r.o. 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.openflowjava.protocol.impl.core.connection;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Preconditions;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueException;
  14 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
  15 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReplyMessage;
  16 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  17 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketOutInput;
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20
  21 final class OutboundQueueEntry {
  22     private static final Logger LOG = LoggerFactory.getLogger(OutboundQueueEntry.class);
  23     private FutureCallback<OfHeader> callback;
  24     private OfHeader message;
  25     private boolean completed;
  26     private boolean barrier;
  27     private volatile boolean committed;
  28     private OutboundQueueException lastException = null;
  29
  30     void commit(final OfHeader message, final FutureCallback<OfHeader> callback) {
  31         if (this.completed) {
  32             LOG.warn("Can't commit a completed message.");
  33             if (callback != null) {
  34                 callback.onFailure(lastException);
  35             }
  36         } else {
  37             this.message = message;
  38             this.callback = callback;
  39             this.barrier = message instanceof BarrierInput;
  40
  41             // Volatile write, needs to be last
  42             this.committed = true;
  43         }
  44     }
  45
  46     void reset() {
  47         barrier = false;
  48         callback = null;
  49         completed = false;
  50         message = null;
  51
  52         // Volatile write, needs to be last
  53         committed = false;
  54     }
  55
  56     boolean isBarrier() {
  57         return barrier;
  58     }
  59
  60     boolean isCommitted() {
  61         return committed;
  62     }
  63
  64     boolean isCompleted() {
  65         return completed;
  66     }
  67
  68     OfHeader takeMessage() {
  69         final OfHeader ret = message;
  70         if (!barrier) {
  71             checkCompletionNeed();
  72         }
  73         message = null;
  74         return ret;
  75     }
  76
  77     private void checkCompletionNeed() {
  78         if (callback == null || (message instanceof PacketOutInput)) {
  79             completed = true;
  80             if (callback != null) {
  81                 callback.onSuccess(null);
  82                 callback = null;
  83             }
  84             committed = false;
  85         }
  86     }
  87
  88     boolean complete(final OfHeader response) {
  89         Preconditions.checkState(!completed, "Attempted to complete a completed message with response %s", response);
  90
  91         // Multipart requests are special, we have to look at them to see
  92         // if there is something outstanding and adjust ourselves accordingly
  93         final boolean reallyComplete;
  94         if (response instanceof MultipartReplyMessage) {
  95             reallyComplete = !((MultipartReplyMessage) response).getFlags().isOFPMPFREQMORE();
  96             LOG.debug("Multipart reply {}", response);
  97         } else {
  98             reallyComplete = true;
  99         }
 100
 101         completed = reallyComplete;
 102         if (callback != null) {
 103             callback.onSuccess(response);
 104             if (reallyComplete) {
 105                 // We will not need the callback anymore, make sure it can be GC'd
 106                 callback = null;
 107             }
 108         }
 109         LOG.debug("Entry {} completed {} with response {}", this, completed, response);
 110         return reallyComplete;
 111     }
 112
 113     void fail(final OutboundQueueException cause) {
 114         if (!completed) {
 115             lastException = cause;
 116             completed = true;
 117             if (callback != null) {
 118                 callback.onFailure(cause);
 119                 callback = null;
 120             }
 121         } else {
 122             LOG.warn("Ignoring failure {} for completed message", cause);
 123         }
 124     }
 125
 126     @VisibleForTesting
 127     /** This method is only for testing to prove that after queue entry is completed there is not callback future */
 128     boolean hasCallback() {
 129         return (callback != null);
 130     }
 131 }