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