X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fservices%2FSalFlowServiceImpl.java;h=5a4ed71f75fa31bd683ccf159873f79946e18a8a;hb=0ac8877c3539d82e55ffe49dc0a2c8bcb0c82b19;hp=c0ab557425994f2d06426f6820db02fdbb4099ab;hpb=e8cbd779e310a5877d2a5cad5bed75ba9839bb5e;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java index c0ab557425..5a4ed71f75 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java @@ -12,7 +12,6 @@ import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import java.util.ArrayList; -import java.util.Collection; import java.util.List; import java.util.Objects; import java.util.concurrent.Future; @@ -26,6 +25,8 @@ import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource; import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener; import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory; import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory; +import org.opendaylight.openflowplugin.impl.util.ErrorUtil; +import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor; import org.opendaylight.openflowplugin.openflow.md.util.FlowCreatorUtil; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId; @@ -44,12 +45,10 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.Upda import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.OriginalFlow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.UpdatedFlow; -import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder; import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier; -import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -62,11 +61,11 @@ public class SalFlowServiceImpl implements SalFlowService, ItemLifeCycleSource { private final DeviceContext deviceContext; private ItemLifecycleListener itemLifecycleListener; - public SalFlowServiceImpl(final RequestContextStack requestContextStack, final DeviceContext deviceContext) { + public SalFlowServiceImpl(final RequestContextStack requestContextStack, final DeviceContext deviceContext, final ConvertorExecutor convertorExecutor) { this.deviceContext = deviceContext; - flowRemove = new FlowService(requestContextStack, deviceContext, RemoveFlowOutput.class); - flowAdd = new FlowService<>(requestContextStack, deviceContext, AddFlowOutput.class); - flowUpdate = new FlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class); + flowRemove = new FlowService<>(requestContextStack, deviceContext, RemoveFlowOutput.class, convertorExecutor); + flowAdd = new FlowService<>(requestContextStack, deviceContext, AddFlowOutput.class, convertorExecutor); + flowUpdate = new FlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class, convertorExecutor); } @Override @@ -76,103 +75,25 @@ public class SalFlowServiceImpl implements SalFlowService, ItemLifeCycleSource { @Override public Future> addFlow(final AddFlowInput input) { - final FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(input); - final FlowId flowId; - final FlowDescriptor flowDescriptor; - - if (Objects.nonNull(input.getFlowRef())) { - flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); - flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId); - deviceContext.getDeviceFlowRegistry().store(flowRegistryKey, flowDescriptor); - } else { - flowId = deviceContext.getDeviceFlowRegistry().storeIfNecessary(flowRegistryKey); - flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId); - } - - LOG.trace("Calling add flow for flow with ID ={}.", flowId); + final FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input); final ListenableFuture> future = flowAdd.processFlowModInputBuilders(flowAdd.toFlowModInputs(input)); - Futures.addCallback(future, new FutureCallback>() { - @Override - public void onSuccess(final RpcResult rpcResult) { - if (rpcResult.isSuccessful()) { - if(LOG.isDebugEnabled()) { - LOG.debug("flow add with id={},finished without error,", flowId.getValue()); - } - if (itemLifecycleListener != null) { - KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, - deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); - final FlowBuilder flowBuilder = new FlowBuilder(input).setId(flowDescriptor.getFlowId()); - itemLifecycleListener.onAdded(flowPath, flowBuilder.build()); - } - } else { - LOG.error("flow add failed for id={}, errors={}", flowId.getValue(), errorsToString(rpcResult.getErrors())); - } - } - - @Override - public void onFailure(final Throwable throwable) { - deviceContext.getDeviceFlowRegistry().markToBeremoved(flowRegistryKey); - LOG.error("Service call for adding flow with id={} failed, reason {} .", flowId.getValue(), throwable); - } - }); - + Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey)); return future; } @Override public Future> removeFlow(final RemoveFlowInput input) { - LOG.trace("Calling remove flow for flow with ID ={}.", input.getFlowRef()); - final ListenableFuture> future = flowRemove.processFlowModInputBuilders(flowRemove.toFlowModInputs(input)); - Futures.addCallback(future, new FutureCallback>() { - @Override - public void onSuccess(final RpcResult result) { - if (result.isSuccessful()) { - if(LOG.isDebugEnabled()) { - LOG.debug("flow removed finished without error,"); - } - FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(input); - deviceContext.getDeviceFlowRegistry().markToBeremoved(flowRegistryKey); - if (itemLifecycleListener != null) { - final FlowDescriptor flowDescriptor = - deviceContext.getDeviceFlowRegistry().retrieveIdForFlow(flowRegistryKey); - if (flowDescriptor != null) { - KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, - deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); - itemLifecycleListener.onRemoved(flowPath); - } - } - } else { - LOG.error("Flow remove failed with errors : {}",errorsToString(result.getErrors())); - } - } - - @Override - public void onFailure(final Throwable throwable) { - LOG.error("Service call for removing flow with id {} failed ,reason {}",input.getFlowRef().getValue(), throwable); - } - }); - + Futures.addCallback(future, new RemoveFlowCallback(input)); return future; } - private final String errorsToString(final Collection rpcErrors) { - final StringBuilder errors = new StringBuilder(); - if ((null != rpcErrors) && (rpcErrors.size() > 0)) { - for (final RpcError rpcError : rpcErrors) { - errors.append(rpcError.getMessage()); - } - } - return errors.toString(); - } - @Override public Future> updateFlow(final UpdateFlowInput input) { - final UpdateFlowInput in = input; - final UpdatedFlow updated = in.getUpdatedFlow(); - final OriginalFlow original = in.getOriginalFlow(); + final UpdatedFlow updated = input.getUpdatedFlow(); + final OriginalFlow original = input.getOriginalFlow(); final List allFlowMods = new ArrayList<>(); final List ofFlowModInputs; @@ -193,62 +114,153 @@ public class SalFlowServiceImpl implements SalFlowService, ItemLifeCycleSource { allFlowMods.addAll(ofFlowModInputs); ListenableFuture> future = flowUpdate.processFlowModInputBuilders(allFlowMods); - Futures.addCallback(future, new FutureCallback>() { - @Override - public void onSuccess(final RpcResult o) { - FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(original); + Futures.addCallback(future, new UpdateFlowCallback(input)); + return future; + } - FlowRegistryKey updatedflowRegistryKey = FlowRegistryKeyFactory.create(updated); - final FlowRef flowRef = input.getFlowRef(); - final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry(); + @VisibleForTesting + private static KeyedInstanceIdentifier createFlowPath(FlowDescriptor flowDescriptor, + KeyedInstanceIdentifier nodePath) { + return nodePath.augmentation(FlowCapableNode.class) + .child(Table.class, flowDescriptor.getTableKey()) + .child(Flow.class, new FlowKey(flowDescriptor.getFlowId())); + } + + private class AddFlowCallback implements FutureCallback> { + private final AddFlowInput input; + private final FlowRegistryKey flowRegistryKey; + + private AddFlowCallback(final AddFlowInput input, + final FlowRegistryKey flowRegistryKey) { + this.input = input; + this.flowRegistryKey = flowRegistryKey; + } - if (flowRef == null) { //then this is equivalent to a delete - deviceFlowRegistry.markToBeremoved(flowRegistryKey); + @Override + public void onSuccess(final RpcResult rpcResult) { + if (rpcResult.isSuccessful()) { + final FlowDescriptor flowDescriptor; + + if (Objects.nonNull(input.getFlowRef())) { + final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); + flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId); + deviceContext.getDeviceFlowRegistry().store(flowRegistryKey, flowDescriptor); + } else { + final FlowId flowId = deviceContext.getDeviceFlowRegistry().storeIfNecessary(flowRegistryKey); + flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId); + } + + if (LOG.isDebugEnabled()) { + LOG.debug("Flow add with id={} finished without error", flowDescriptor.getFlowId().getValue()); + } - if (itemLifecycleListener != null) { - final FlowDescriptor flowDescriptor = - deviceContext.getDeviceFlowRegistry().retrieveIdForFlow( flowRegistryKey); + if (itemLifecycleListener != null) { + KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, + deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); + final FlowBuilder flowBuilder = new FlowBuilder(input).setId(flowDescriptor.getFlowId()); + itemLifecycleListener.onAdded(flowPath, flowBuilder.build()); + } + } else { + if (LOG.isDebugEnabled()) { + LOG.debug("Flow add failed for flow={}, errors={}", input, + ErrorUtil.errorsToString(rpcResult.getErrors())); + } + } + } + + @Override + public void onFailure(final Throwable throwable) { + LOG.warn("Service call for adding flow={} failed, reason: {}", input, throwable); + } + } + + private class RemoveFlowCallback implements FutureCallback> { + private final RemoveFlowInput input; + + private RemoveFlowCallback(final RemoveFlowInput input) { + this.input = input; + } + + @Override + public void onSuccess(final RpcResult result) { + if (result.isSuccessful()) { + if (LOG.isDebugEnabled()) { + LOG.debug("Flow remove finished without error for flow={}", input); + } + FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input); + deviceContext.getDeviceFlowRegistry().removeDescriptor(flowRegistryKey); + + if (itemLifecycleListener != null) { + final FlowDescriptor flowDescriptor = + deviceContext.getDeviceFlowRegistry().retrieveIdForFlow(flowRegistryKey); + if (flowDescriptor != null) { KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); itemLifecycleListener.onRemoved(flowPath); } - } else { //this is either an add or an update - final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); - final FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(updated.getTableId(), flowId); - deviceFlowRegistry.store(updatedflowRegistryKey, flowDescriptor); + } + } else { + if (LOG.isDebugEnabled()) { + LOG.debug("Flow remove failed for flow={}, errors={}", input, + ErrorUtil.errorsToString(result.getErrors())); + } + } + } - if (itemLifecycleListener != null) { - KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, - deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); - final FlowBuilder flowBuilder = new FlowBuilder( - input.getUpdatedFlow()).setId(flowDescriptor.getFlowId()); - - boolean isUpdate = null != - deviceFlowRegistry.retrieveIdForFlow(flowRegistryKey); - if (isUpdate) { - itemLifecycleListener.onUpdated(flowPath, flowBuilder.build()); - } else { - itemLifecycleListener.onAdded(flowPath, flowBuilder.build()); - } - } + @Override + public void onFailure(final Throwable throwable) { + LOG.warn("Service call for removing flow={} failed, reason: {}", input, throwable); + } + } + private class UpdateFlowCallback implements FutureCallback> { + private final UpdateFlowInput input; - } + private UpdateFlowCallback(UpdateFlowInput input) { + this.input = input; + } + + @Override + public void onSuccess(final RpcResult o) { + final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry(); + + final UpdatedFlow updated = input.getUpdatedFlow(); + final OriginalFlow original = input.getOriginalFlow(); + final FlowRegistryKey origFlowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), original); + final FlowRegistryKey updatedFlowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), updated); + final FlowDescriptor origFlowDescriptor = deviceFlowRegistry.retrieveIdForFlow(origFlowRegistryKey); + + final boolean isUpdate = Objects.nonNull(origFlowDescriptor); + final FlowId fLowId = Objects.nonNull(input.getFlowRef()) + ? input.getFlowRef().getValue().firstKeyOf(Flow.class).getId() + : isUpdate ? origFlowDescriptor.getFlowId() : deviceFlowRegistry.storeIfNecessary(updatedFlowRegistryKey); + final FlowDescriptor updatedFlowDescriptor = FlowDescriptorFactory.create(updated.getTableId(), fLowId); + if (isUpdate) { + deviceFlowRegistry.removeDescriptor(origFlowRegistryKey); + deviceFlowRegistry.store(updatedFlowRegistryKey, updatedFlowDescriptor); } - @Override - public void onFailure(final Throwable throwable) { - LOG.error("Service call for updating flow failed, reason{}", throwable); + if (itemLifecycleListener != null) { + final KeyedInstanceIdentifier flowPath = + createFlowPath( + updatedFlowDescriptor, + deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); + + final Flow flow = new FlowBuilder(updated) + .setId(updatedFlowDescriptor.getFlowId()) + .build(); + + if (Objects.nonNull(origFlowDescriptor)) { + itemLifecycleListener.onUpdated(flowPath, flow); + } else { + itemLifecycleListener.onAdded(flowPath, flow); + } } - }); - return future; - } + } - @VisibleForTesting - static KeyedInstanceIdentifier createFlowPath(FlowDescriptor flowDescriptor, - KeyedInstanceIdentifier nodePath) { - return nodePath.augmentation(FlowCapableNode.class) - .child(Table.class, flowDescriptor.getTableKey()) - .child(Flow.class, new FlowKey(flowDescriptor.getFlowId())); + @Override + public void onFailure(final Throwable throwable) { + LOG.warn("Service call for updating flow={} failed, reason: {}", input, throwable); + } } }