import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import org.opendaylight.infrautils.utils.concurrent.JdkFutures;
+import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
import org.opendaylight.mdsal.binding.api.ReadTransaction;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.openflowplugin.applications.frm.ForwardingRulesManager;
.getSalBundleService().addBundleMessages(addBundleMessagesInput);
LOG.trace("Pushing flow remove message {} to bundle {} for device {}", addBundleMessagesInput,
bundleId.getValue(), node);
- JdkFutures.addErrorLogging(resultFuture, LOG, "removeBundleFlow");
+ LoggingFutures.addErrorLogging(resultFuture, LOG, "removeBundleFlow");
}
public void update(final InstanceIdentifier<Flow> identifier, final Flow originalFlow, final Flow updatedFlow,
final InstanceIdentifier<FlowCapableNode> nodeIdent, final BundleId bundleId) {
final NodeId nodeId = getNodeIdFromNodeIdentifier(nodeIdent);
nodeConfigurator.enqueueJob(nodeId.getValue(), () -> {
- BundleInnerMessage bundleInnerMessage = new BundleUpdateFlowCaseBuilder()
+ BundleInnerMessage innerDeleteMessage = new BundleRemoveFlowCaseBuilder()
+ .setRemoveFlowCaseData(new RemoveFlowCaseDataBuilder(originalFlow).build()).build();
+ Message deleteMessage = new MessageBuilder().setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setBundleInnerMessage(innerDeleteMessage).build();
+ BundleInnerMessage innerUpdateMessage = new BundleUpdateFlowCaseBuilder()
.setUpdateFlowCaseData(new UpdateFlowCaseDataBuilder(updatedFlow).build()).build();
- Message message = new MessageBuilder().setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
- .setBundleInnerMessage(bundleInnerMessage).build();
+ Message updateMessage = new MessageBuilder().setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)))
+ .setBundleInnerMessage(innerUpdateMessage).build();
ListenableFuture<RpcResult<AddBundleMessagesOutput>> groupFuture = pushDependentGroup(nodeIdent,
updatedFlow, identifier, bundleId);
+ List<Message> messages = Arrays.asList(deleteMessage, updateMessage);
SettableFuture<RpcResult<AddBundleMessagesOutput>> resultFuture = SettableFuture.create();
- Futures.addCallback(groupFuture, new BundleFlowCallBack(nodeIdent, bundleId, message, resultFuture),
+ Futures.addCallback(groupFuture, new BundleFlowCallBack(nodeIdent, bundleId, messages, resultFuture),
MoreExecutors.directExecutor());
return resultFuture;
});
ListenableFuture<RpcResult<AddBundleMessagesOutput>> groupFuture = pushDependentGroup(nodeIdent, flow,
identifier, bundleId);
SettableFuture<RpcResult<AddBundleMessagesOutput>> resultFuture = SettableFuture.create();
- Futures.addCallback(groupFuture, new BundleFlowCallBack(nodeIdent, bundleId, message, resultFuture),
+ Futures.addCallback(groupFuture, new BundleFlowCallBack(nodeIdent, bundleId,
+ Collections.singletonList(message), resultFuture),
MoreExecutors.directExecutor());
return resultFuture;
});
private final class BundleFlowCallBack implements FutureCallback<RpcResult<AddBundleMessagesOutput>> {
private final InstanceIdentifier<FlowCapableNode> nodeIdent;
private final BundleId bundleId;
- private final Message message;
+ private final List<Message> messages;
private final NodeId nodeId;
private final SettableFuture<RpcResult<AddBundleMessagesOutput>> resultFuture;
- BundleFlowCallBack(InstanceIdentifier<FlowCapableNode> nodeIdent, BundleId bundleId, Message message,
+ BundleFlowCallBack(InstanceIdentifier<FlowCapableNode> nodeIdent, BundleId bundleId, List<Message> messages,
SettableFuture<RpcResult<AddBundleMessagesOutput>> resultFuture) {
this.nodeIdent = nodeIdent;
this.bundleId = bundleId;
- this.message = message;
+ this.messages = messages;
this.resultFuture = resultFuture;
nodeId = getNodeIdFromNodeIdentifier(nodeIdent);
}
if (rpcResult.isSuccessful()) {
AddBundleMessagesInput addBundleMessagesInput = new AddBundleMessagesInputBuilder()
.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class))).setBundleId(bundleId)
- .setFlags(BUNDLE_FLAGS).setMessages(new MessagesBuilder().setMessage(
- Collections.singletonList(message)).build()).build();
+ .setFlags(BUNDLE_FLAGS).setMessages(new MessagesBuilder().setMessage(messages).build()).build();
LOG.trace("Pushing flow add message {} to bundle {} for device {}", addBundleMessagesInput,
bundleId.getValue(), nodeId.getValue());
@Override
public void onFailure(Throwable throwable) {
- LOG.error("Error while pushing flow add bundle {} for device {}", message, nodeId);
+ LOG.error("Error while pushing flow add bundle {} for device {}", messages, nodeId);
resultFuture.setException(throwable);
}
}