import static org.opendaylight.openflowplugin.applications.frm.util.FrmUtil.OPENFLOW_PREFIX;
-import java.util.ArrayList;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Collection;
import java.util.Formatter;
-import java.util.List;
import java.util.Map;
-import java.util.Queue;
import org.apache.karaf.shell.commands.Command;
import org.apache.karaf.shell.commands.Option;
import org.apache.karaf.shell.console.OsgiCommandSupport;
-import org.opendaylight.openflowplugin.api.openflow.FlowGroupCache;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager;
+import org.opendaylight.openflowplugin.api.openflow.FlowGroupInfo;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
@Command(scope = "openflow", name = "getflownodecache", description = "Print all flow/group cache")
public class GetFlowGroupCacheProvider extends OsgiCommandSupport {
this.flowGroupCacheManager = flowGroupCacheManager;
}
- @SuppressWarnings("checkstyle:RegexpSinglelineJava")
@Override
- protected Object doExecute() throws Exception {
- String nodeId = OPENFLOW_PREFIX + dpnId;
- List<String> result = new ArrayList<>();
-
+ @SuppressWarnings("checkstyle:RegexpSinglelineJava")
+ protected Object doExecute() {
if (dpnId == null) {
- Map<String, Queue<FlowGroupCache>> flowGroupCacheListForAllNodes = flowGroupCacheManager
- .getAllNodesFlowGroupCache();
- if (!flowGroupCacheListForAllNodes.isEmpty()) {
- StringBuilder stringBuilder = new StringBuilder();
- Formatter formatter = new Formatter(stringBuilder);
- result.add(getAllLocalNodesHeaderOutput());
- result.add(getLineSeparator());
- for (Map.Entry<String, Queue<FlowGroupCache>> cacheEntry : flowGroupCacheListForAllNodes.entrySet()) {
- String[] temp = cacheEntry.getKey().split(":");
- String node = temp[1];
- Queue<FlowGroupCache> flowGroupCacheList = cacheEntry.getValue();
- synchronized (flowGroupCacheList) {
- for (FlowGroupCache cache : flowGroupCacheList) {
- result.add(formatter.format("%-15s %1s %-10s %1s %-8s %1s %-21s %1s %-60s",
- node, "", cache.getDescription(), "", cache.getStatus(), "",
- cache.getTime(), "", cache.getId()).toString());
- stringBuilder.setLength(0);
- }
- }
- }
- formatter.close();
- result.stream().forEach(p -> System.out.println(p));
- } else {
- session.getConsole().println("No flow/group is programmed yet");
- }
- } else {
- if (!flowGroupCacheManager.getAllNodesFlowGroupCache().containsKey(nodeId)) {
- session.getConsole().println("No node available for this NodeID");
- return null;
- }
- Queue<FlowGroupCache> flowGroupCacheList = flowGroupCacheManager.getAllNodesFlowGroupCache()
- .get(nodeId);
- if (!flowGroupCacheList.isEmpty()) {
- StringBuilder stringBuilder = new StringBuilder();
- Formatter formatter = new Formatter(stringBuilder);
- result.add(String.format("Number of flows and groups in cache for node %s : %d", nodeId,
- flowGroupCacheList.size()));
- result.add(getLocalNodeHeaderOutput());
- result.add(getLineSeparator());
- for (FlowGroupCache cache : flowGroupCacheList) {
- result.add(formatter.format("%-10s %1s %-8s %1s %-23s %1s %-60s",
- cache.getDescription(), "", cache.getStatus(), "",
- cache.getTime(), "", cache.getId()).toString());
- stringBuilder.setLength(0);
+ printAllNodes();
+ return null;
+ }
+
+ final String nodeId = OPENFLOW_PREFIX + dpnId;
+ Collection<FlowGroupInfo> flowGroupCacheList = flowGroupCacheManager.getFlowGroupCache(new NodeId(nodeId));
+ if (flowGroupCacheList == null) {
+ session.getConsole().println("No node available for this NodeID");
+ return null;
+ }
+ if (flowGroupCacheList.isEmpty()) {
+ session.getConsole().println("No flow/group is programmed yet for the the node " + nodeId);
+ return null;
+ }
+
+ StringBuilder sb = new StringBuilder();
+ Formatter fmt = new Formatter(sb);
+ System.out.println(String.format("Number of flows and groups in cache for node %s : %d", nodeId,
+ flowGroupCacheList.size()));
+ System.out.println(getLocalNodeHeaderOutput());
+ System.out.println(getLineSeparator());
+
+ for (FlowGroupInfo cache : flowGroupCacheList) {
+ System.out.println(fmt.format("%-10s %1s %-8s %1s %-23s %1s %-60s", cache.getDescription(), "",
+ cache.getStatus(), "", getTime(cache), "", cache.getId()).toString());
+ sb.setLength(0);
+ }
+ fmt.close();
+ return null;
+ }
+
+ private static LocalDateTime getTime(final FlowGroupInfo info) {
+ return LocalDateTime.ofInstant(info.getInstantUTC(), ZoneOffset.UTC);
+ }
+
+ @SuppressWarnings("checkstyle:RegexpSinglelineJava")
+ private void printAllNodes() {
+ final Map<NodeId, Collection<FlowGroupInfo>> allGroupInfos = flowGroupCacheManager.getAllNodesFlowGroupCache();
+ if (allGroupInfos.isEmpty()) {
+ session.getConsole().println("No flow/group is programmed yet");
+ return;
+ }
+
+ StringBuilder sb = new StringBuilder();
+ Formatter fmt = new Formatter(sb);
+ System.out.println(getAllLocalNodesHeaderOutput());
+ System.out.println(getLineSeparator());
+ for (Map.Entry<NodeId, Collection<FlowGroupInfo>> cacheEntry : allGroupInfos.entrySet()) {
+ // FIXME: just seek/substring
+ String[] temp = cacheEntry.getKey().getValue().split(":");
+ String node = temp[1];
+ Collection<FlowGroupInfo> flowGroupCacheList = cacheEntry.getValue();
+ synchronized (flowGroupCacheList) {
+ for (FlowGroupInfo cache : flowGroupCacheList) {
+ System.out.println(fmt.format("%-15s %1s %-10s %1s %-8s %1s %-21s %1s %-60s", node, "",
+ cache.getDescription(), "", cache.getStatus(), "", getTime(cache), "",
+ cache.getId()).toString());
+ sb.setLength(0);
}
- formatter.close();
- result.stream().forEach(p -> System.out.println(p));
- } else {
- session.getConsole().println("No flow/group is programmed yet for the the node " + nodeId);
}
}
- return null;
+ fmt.close();
}
private static String getLocalNodeHeaderOutput() {
+++ /dev/null
-/*
- * Copyright (c) 2020 Ericsson India Global Services Pvt Ltd. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.openflowplugin.api.openflow;
-
-import java.time.LocalDateTime;
-
-public class FlowGroupCache {
- private final String id;
- private final String description;
- private final FlowGroupStatus status;
- private final LocalDateTime time;
-
- public FlowGroupCache(String id, String description, FlowGroupStatus status,
- LocalDateTime time) {
- this.id = id;
- this.description = description;
- this.status = status;
- this.time = time;
- }
-
- public String getId() {
- return id;
- }
-
- public String getDescription() {
- return description;
- }
-
- public FlowGroupStatus getStatus() {
- return status;
- }
-
- public LocalDateTime getTime() {
- return time;
- }
-}
*/
package org.opendaylight.openflowplugin.api.openflow;
+import java.util.Collection;
import java.util.Map;
-import java.util.Queue;
+import org.eclipse.jdt.annotation.NonNull;
+import org.eclipse.jdt.annotation.Nullable;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupTypes;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
+import org.opendaylight.yangtools.yang.common.Uint8;
public interface FlowGroupCacheManager {
Map<String, ReconciliationState> getReconciliationStates();
- Map<String, Queue<FlowGroupCache>> getAllNodesFlowGroupCache();
+ // FIXME: this quite unrelated to getReconciliationStates() to the point
+ Map<NodeId, Collection<FlowGroupInfo>> getAllNodesFlowGroupCache();
+
+ @Nullable Collection<FlowGroupInfo> getFlowGroupCache(@NonNull NodeId nodeId);
+
+ // FIXME: these two methods should live in a separate interface
+ void appendFlow(@NonNull NodeId nodeId, @NonNull FlowId id, Uint8 tableId, @NonNull FlowGroupStatus status);
+
+ void appendGroup(@NonNull NodeId nodeId, @NonNull GroupId id, @NonNull GroupTypes type,
+ @NonNull FlowGroupStatus status);
}
--- /dev/null
+/*
+ * Copyright (c) 2020 Ericsson India Global Services Pvt Ltd. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.openflowplugin.api.openflow;
+
+import static java.util.Objects.requireNonNull;
+
+import java.time.Instant;
+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupTypes;
+import org.opendaylight.yangtools.concepts.Immutable;
+import org.opendaylight.yangtools.yang.common.Uint8;
+
+@NonNullByDefault
+public abstract class FlowGroupInfo implements Immutable {
+ private static final class Flow extends FlowGroupInfo {
+ private final Uint8 tableId;
+ private final FlowId id;
+
+ Flow(final FlowGroupStatus status, final FlowId id, final Uint8 tableId) {
+ super(status);
+ this.id = requireNonNull(id);
+ this.tableId = requireNonNull(tableId);
+ }
+
+ @Override
+ public String getId() {
+ return id.getValue();
+ }
+
+ @Override
+ public String getDescription() {
+ return tableId.toString();
+ }
+ }
+
+ private static final class Group extends FlowGroupInfo {
+ private final GroupTypes type;
+ private final GroupId id;
+
+ Group(final FlowGroupStatus status, final GroupId id, final GroupTypes type) {
+ super(status);
+ this.id = requireNonNull(id);
+ this.type = requireNonNull(type);
+ }
+
+ @Override
+ public String getId() {
+ // FIXME: GroupId.toString() is not pretty, can we do something else?
+ return id.toString();
+ }
+
+ @Override
+ public String getDescription() {
+ return type.getName();
+ }
+ }
+
+ private final Instant time = Instant.now();
+ private final FlowGroupStatus status;
+
+ private FlowGroupInfo(final FlowGroupStatus status) {
+ this.status = requireNonNull(status);
+ }
+
+ public static FlowGroupInfo ofFlow(final FlowId id, final Uint8 tableId, final FlowGroupStatus status) {
+ return new Flow(status, id, tableId);
+ }
+
+ public static FlowGroupInfo ofGroup(final GroupId id, final GroupTypes type, final FlowGroupStatus status) {
+ return new Group(status, id, type);
+ }
+
+ public abstract String getId();
+
+ public abstract String getDescription();
+
+ public final FlowGroupStatus getStatus() {
+ return status;
+ }
+
+ public final Instant getInstantUTC() {
+ return time;
+ }
+}
*/
package org.opendaylight.openflowplugin.impl.services.cache;
+import com.google.common.collect.EvictingQueue;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import javax.inject.Singleton;
import org.apache.aries.blueprint.annotation.service.Service;
-import org.opendaylight.openflowplugin.api.openflow.FlowGroupCache;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager;
+import org.opendaylight.openflowplugin.api.openflow.FlowGroupInfo;
+import org.opendaylight.openflowplugin.api.openflow.FlowGroupStatus;
import org.opendaylight.openflowplugin.api.openflow.ReconciliationState;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupTypes;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
+import org.opendaylight.yangtools.yang.common.Uint8;
@Singleton
@Service(classes = FlowGroupCacheManager.class)
public class FlowGroupCacheManagerImpl implements FlowGroupCacheManager {
+ // FIXME: make this configurable and expose a different implementation at least for OSGi when this is switched off
+ private static final int FLOWGROUP_CACHE_SIZE = 10000;
- private Map<String, ReconciliationState> reconciliationStates = new ConcurrentHashMap<>();
- private Map<String, Queue<FlowGroupCache>> allNodesFlowGroupCache = new ConcurrentHashMap<>();
+ private final Map<String, ReconciliationState> reconciliationStates = new ConcurrentHashMap<>();
+ private final Map<NodeId, Queue<FlowGroupInfo>> flowGroups = new ConcurrentHashMap<>();
@Override
- public Map<String, Queue<FlowGroupCache>> getAllNodesFlowGroupCache() {
- return allNodesFlowGroupCache;
+ public Map<String, ReconciliationState> getReconciliationStates() {
+ return reconciliationStates;
}
@Override
- public Map<String, ReconciliationState> getReconciliationStates() {
- return reconciliationStates;
+ public Map<NodeId, Collection<FlowGroupInfo>> getAllNodesFlowGroupCache() {
+ return Collections.unmodifiableMap(Maps.transformValues(flowGroups, Collections::unmodifiableCollection));
+ }
+
+ @Override
+ public Collection<FlowGroupInfo> getFlowGroupCache(final NodeId nodeId) {
+ final Collection<FlowGroupInfo> result = flowGroups.get(nodeId);
+ return result == null ? List.of() : Collections.unmodifiableCollection(result);
+ }
+
+ @Override
+ public void appendFlow(final NodeId nodeId, final FlowId id, final Uint8 tableId, final FlowGroupStatus status) {
+ flowGroup(nodeId).add(FlowGroupInfo.ofFlow(id, tableId, status));
+ }
+
+ @Override
+ public void appendGroup(final NodeId nodeId, final GroupId id, final GroupTypes type,
+ final FlowGroupStatus status) {
+ flowGroup(nodeId).add(FlowGroupInfo.ofGroup(id, type, status));
+ }
+
+ // FIXME: we really want to split out the 'nodeId' lookup and provide an internal interface for the plugin to
+ // contribute directly to the queue.
+ private @NonNull Queue<FlowGroupInfo> flowGroup(final NodeId nodeId) {
+ return flowGroups.computeIfAbsent(nodeId,
+ // FIXME: synchronized queue relies on locking -- and most of the time all we access it from the same(-ish)
+ // context. We should be able to do better.
+ key -> Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE)));
}
}
\ No newline at end of file
*/
package org.opendaylight.openflowplugin.impl.services.sal;
-import com.google.common.collect.EvictingQueue;
-import com.google.common.collect.Queues;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
-import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
-import java.util.Queue;
import org.opendaylight.openflowplugin.api.OFConstants;
-import org.opendaylight.openflowplugin.api.openflow.FlowGroupCache;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupStatus;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
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.NodeId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder;
import org.opendaylight.yangtools.yang.common.RpcError;
public class SalFlowServiceImpl implements SalFlowService {
private static final Logger LOG = LoggerFactory.getLogger(SalFlowServiceImpl.class);
+ private static final Uint8 OFPTT_ALL = Uint8.MAX_VALUE;
+
private final MultiLayerFlowService<UpdateFlowOutput> flowUpdate;
private final MultiLayerFlowService<AddFlowOutput> flowAdd;
private final MultiLayerFlowService<RemoveFlowOutput> flowRemove;
private final SingleLayerFlowService<UpdateFlowOutput> flowUpdateMessage;
private final SingleLayerFlowService<RemoveFlowOutput> flowRemoveMessage;
private final DeviceContext deviceContext;
- private static final Uint8 OFPTT_ALL = Uint8.MAX_VALUE;
private final FlowGroupCacheManager provider;
- public static final int FLOWGROUP_CACHE_SIZE = 10000;
-
public SalFlowServiceImpl(final RequestContextStack requestContextStack,
final DeviceContext deviceContext,
public ListenableFuture<RpcResult<UpdateFlowOutput>> updateFlow(final UpdateFlowInput input) {
final UpdatedFlow updated = input.getUpdatedFlow();
final OriginalFlow original = input.getOriginalFlow();
- String nodeId = PathUtil.extractNodeId(input.getNode()).getValue();
+ final NodeId nodeId = PathUtil.extractNodeId(input.getNode());
final List<FlowModInputBuilder> allFlowMods = new ArrayList<>();
final List<FlowModInputBuilder> ofFlowModInputs;
Futures.addCallback(listListenableFuture, new FutureCallback<List<RpcResult<UpdateFlowOutput>>>() {
@Override
public void onSuccess(final List<RpcResult<UpdateFlowOutput>> results) {
- final ArrayList<RpcError> errors = new ArrayList();
+ final ArrayList<RpcError> errors = new ArrayList<>();
for (RpcResult<UpdateFlowOutput> flowModResult : results) {
if (flowModResult == null) {
errors.add(RpcResultBuilder.newError(
@Override
public void onSuccess(final RpcResult<AddFlowOutput> rpcResult) {
- if (rpcResult.isSuccessful()) {
- final FlowDescriptor flowDescriptor;
- final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class).getId();
- FlowGroupCache cache = new FlowGroupCache(flowId.getValue(), input.getTableId().toString(),
- FlowGroupStatus.ADDED, LocalDateTime.now());
- if (provider.getAllNodesFlowGroupCache().containsKey(nodeId.getValue())) {
- provider.getAllNodesFlowGroupCache().get(nodeId.getValue()).add(cache);
- } else {
- Queue<FlowGroupCache> flowGroupCacheList =
- Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE));
- flowGroupCacheList.add(cache);
- provider.getAllNodesFlowGroupCache().put(nodeId.getValue(), flowGroupCacheList);
- }
- if (input.getFlowRef() != null) {
- flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId);
- deviceContext.getDeviceFlowRegistry().storeDescriptor(flowRegistryKey, flowDescriptor);
- } else {
- deviceContext.getDeviceFlowRegistry().store(flowRegistryKey);
- flowDescriptor = deviceContext.getDeviceFlowRegistry().retrieveDescriptor(flowRegistryKey);
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Flow add with id={} finished without error", flowDescriptor.getFlowId().getValue());
- }
- } else {
+ if (!rpcResult.isSuccessful()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Flow add failed for flow={}, errors={}", input,
ErrorUtil.errorsToString(rpcResult.getErrors()));
}
+ return;
+ }
+
+ final FlowDescriptor flowDescriptor;
+ final FlowRef flowRef = input.getFlowRef();
+ if (flowRef != null) {
+ final Uint8 tableId = input.getTableId();
+ final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class).getId();
+ provider.appendFlow(nodeId, flowId, tableId, FlowGroupStatus.ADDED);
+
+ flowDescriptor = FlowDescriptorFactory.create(tableId, flowId);
+ deviceContext.getDeviceFlowRegistry().storeDescriptor(flowRegistryKey, flowDescriptor);
+ } else {
+ deviceContext.getDeviceFlowRegistry().store(flowRegistryKey);
+ flowDescriptor = deviceContext.getDeviceFlowRegistry().retrieveDescriptor(flowRegistryKey);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Flow add with id={} finished without error", flowDescriptor.getFlowId().getValue());
}
}
FlowRegistryKey flowRegistryKey =
FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input);
deviceContext.getDeviceFlowRegistry().addMark(flowRegistryKey);
- final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class).getId();
- FlowGroupCache cache = new FlowGroupCache(flowId.getValue(),
- input.getTableId().toString(), FlowGroupStatus.REMOVED,
- LocalDateTime.now());
- if (provider.getAllNodesFlowGroupCache().containsKey(nodeId.getValue())) {
- provider.getAllNodesFlowGroupCache().get(nodeId.getValue()).add(cache);
- } else {
- Queue<FlowGroupCache> flowGroupCacheList =
- Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE));
- flowGroupCacheList.add(cache);
- provider.getAllNodesFlowGroupCache().put(nodeId.getValue(), flowGroupCacheList);
+
+ final FlowRef flowRef = input.getFlowRef();
+ if (flowRef != null) {
+ final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class).getId();
+ provider.appendFlow(nodeId, flowId, input.getTableId(), FlowGroupStatus.REMOVED);
}
} else {
deviceContext.getDeviceFlowRegistry().clearFlowRegistry();
private final class UpdateFlowCallback implements FutureCallback<RpcResult<UpdateFlowOutput>> {
private final UpdateFlowInput input;
- private final String nodeId;
+ private final NodeId nodeId;
- private UpdateFlowCallback(UpdateFlowInput input, String nodeId) {
+ private UpdateFlowCallback(final UpdateFlowInput input, final NodeId nodeId) {
this.input = input;
this.nodeId = nodeId;
}
@Override
public void onSuccess(final RpcResult<UpdateFlowOutput> updateFlowOutputRpcResult) {
final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry();
- final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class).getId();
final UpdatedFlow updated = input.getUpdatedFlow();
final OriginalFlow original = input.getOriginalFlow();
final FlowRegistryKey origFlowRegistryKey =
final boolean isUpdate = origFlowDescriptor != null;
final FlowDescriptor updatedFlowDescriptor;
- FlowGroupCache cache = new FlowGroupCache(flowId.getValue(), updated.getTableId().toString(),
- FlowGroupStatus.MODIFIED,
- LocalDateTime.now());
- if (provider.getAllNodesFlowGroupCache().containsKey(nodeId)) {
- provider.getAllNodesFlowGroupCache().get(nodeId).add(cache);
- } else {
- Queue<FlowGroupCache> flowGroupCacheList =
- Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE));
- flowGroupCacheList.add(cache);
- provider.getAllNodesFlowGroupCache().put(nodeId, flowGroupCacheList);
- }
-
- if (input.getFlowRef() != null) {
- updatedFlowDescriptor =
- FlowDescriptorFactory.create(updated.getTableId(),
- input.getFlowRef().getValue().firstKeyOf(Flow.class).getId());
+ final FlowRef flowRef = input.getFlowRef();
+ if (flowRef != null) {
+ final Uint8 tableId = updated.getTableId();
+ final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class).getId();
+ provider.appendFlow(nodeId, flowId, tableId, FlowGroupStatus.MODIFIED);
+
+ updatedFlowDescriptor = FlowDescriptorFactory.create(tableId, flowId);
+ } else if (isUpdate) {
+ updatedFlowDescriptor = origFlowDescriptor;
} else {
- if (isUpdate) {
- updatedFlowDescriptor = origFlowDescriptor;
- } else {
- deviceFlowRegistry.store(updatedFlowRegistryKey);
- updatedFlowDescriptor = deviceFlowRegistry.retrieveDescriptor(updatedFlowRegistryKey);
- }
+ deviceFlowRegistry.store(updatedFlowRegistryKey);
+ updatedFlowDescriptor = deviceFlowRegistry.retrieveDescriptor(updatedFlowRegistryKey);
}
if (isUpdate) {
*/
package org.opendaylight.openflowplugin.impl.services.sal;
-import com.google.common.collect.EvictingQueue;
-import com.google.common.collect.Queues;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
-import java.time.LocalDateTime;
-import java.util.Queue;
-import org.opendaylight.openflowplugin.api.openflow.FlowGroupCache;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupCacheManager;
import org.opendaylight.openflowplugin.api.openflow.FlowGroupStatus;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.SalGroupService;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.UpdateGroupInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.UpdateGroupOutput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.group.service.rev130918.group.update.UpdatedGroup;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.Group;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
import org.opendaylight.yangtools.yang.common.RpcResult;
addGroupMessage.canUseSingleLayerSerialization()
? addGroupMessage.handleServiceCall(input)
: addGroup.handleServiceCall(input);
- String nodeId = PathUtil.extractNodeId(input.getNode()).getValue();
+ NodeId nodeId = PathUtil.extractNodeId(input.getNode());
Futures.addCallback(resultFuture, new FutureCallback<RpcResult<AddGroupOutput>>() {
@Override
- public void onSuccess(RpcResult<AddGroupOutput> result) {
+ public void onSuccess(
final RpcResult<AddGroupOutput> result) {
if (result.isSuccessful()) {
LOG.debug("adding group successful {}", input.getGroupId());
- FlowGroupCache cache = new FlowGroupCache(input.getGroupId().toString(),
- input.getGroupType().getName(), FlowGroupStatus.ADDED,
- LocalDateTime.now());
- if (provider.getAllNodesFlowGroupCache().containsKey(nodeId)) {
- provider.getAllNodesFlowGroupCache().get(nodeId).add(cache);
- } else {
- Queue<FlowGroupCache> flowGroupCacheList =
- Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE));
- flowGroupCacheList.add(cache);
- provider.getAllNodesFlowGroupCache().put(nodeId, flowGroupCacheList);
- }
+ provider.appendGroup(nodeId, input.getGroupId(), input.getGroupType(), FlowGroupStatus.ADDED);
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("Group add with id={} failed, errors={}", input.getGroupId().getValue(),
}
@Override
- public void onFailure(Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Service call for adding group={} failed",
input.getGroupId().getValue(),
throwable);
Futures.addCallback(resultFuture, new FutureCallback<RpcResult<UpdateGroupOutput>>() {
@Override
- public void onSuccess(RpcResult<UpdateGroupOutput> result) {
+ public void onSuccess(final RpcResult<UpdateGroupOutput> result) {
if (result.isSuccessful()) {
- NodeId nodeId = PathUtil.extractNodeId(input.getNode());
- FlowGroupCache cache = new FlowGroupCache(
- input.getUpdatedGroup().getGroupId().getValue().toString(),
- input.getUpdatedGroup().getGroupType().getName(), FlowGroupStatus.MODIFIED,
- LocalDateTime.now());
- if (provider.getAllNodesFlowGroupCache().containsKey(nodeId.getValue())) {
- provider.getAllNodesFlowGroupCache().get(nodeId.getValue()).add(cache);
- } else {
- Queue<FlowGroupCache> flowGroupCacheList =
- Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE));
- flowGroupCacheList.add(cache);
- provider.getAllNodesFlowGroupCache().put(nodeId.getValue(), flowGroupCacheList);
- }
+ UpdatedGroup updatedGroup = input.getUpdatedGroup();
+ provider.appendGroup(PathUtil.extractNodeId(input.getNode()),
+ updatedGroup.getGroupId(), updatedGroup.getGroupType(), FlowGroupStatus.MODIFIED);
if (LOG.isDebugEnabled()) {
LOG.debug("Group update with original id={} finished without error",
input.getOriginalGroup().getGroupId().getValue());
}
@Override
- public void onFailure(Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Service call for updating group={} failed",
input.getOriginalGroup().getGroupId(), throwable);
}
Futures.addCallback(resultFuture, new FutureCallback<RpcResult<RemoveGroupOutput>>() {
@Override
- public void onSuccess(RpcResult<RemoveGroupOutput> result) {
+ public void onSuccess(final RpcResult<RemoveGroupOutput> result) {
if (result.isSuccessful()) {
if (LOG.isDebugEnabled()) {
- NodeId nodeId = PathUtil.extractNodeId(input.getNode());
LOG.debug("Group remove with id={} finished without error", input.getGroupId().getValue());
- FlowGroupCache cache = new FlowGroupCache(input.getGroupId().getValue().toString(),
- input.getGroupType().getName(), FlowGroupStatus.REMOVED,
- LocalDateTime.now());
- if (provider.getAllNodesFlowGroupCache().containsKey(nodeId.getValue())) {
- provider.getAllNodesFlowGroupCache().get(nodeId.getValue()).add(cache);
- } else {
- Queue<FlowGroupCache> flowGroupCacheList =
- Queues.synchronizedQueue(EvictingQueue.create(FLOWGROUP_CACHE_SIZE));
- flowGroupCacheList.add(cache);
- provider.getAllNodesFlowGroupCache().put(nodeId.getValue(), flowGroupCacheList);
- }
+ provider.appendGroup(PathUtil.extractNodeId(input.getNode()), input.getGroupId(),
+ input.getGroupType(), FlowGroupStatus.REMOVED);
}
} else {
LOG.warn("Group remove with id={} failed, errors={}", input.getGroupId().getValue(),
}
@Override
- public void onFailure(Throwable throwable) {
+ public void onFailure(final Throwable throwable) {
LOG.warn("Service call for removing group={} failed",
input.getGroupId().getValue(), throwable);
}
import static org.mockito.Mockito.when;
import com.google.common.util.concurrent.Futures;
-import java.time.LocalDateTime;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.Queue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import junit.framework.TestCase;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
import org.opendaylight.openflowplugin.api.OFConstants;
-import org.opendaylight.openflowplugin.api.openflow.FlowGroupCache;
-import org.opendaylight.openflowplugin.api.openflow.FlowGroupStatus;
import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
import org.opendaylight.yangtools.yang.common.Uint64;
import org.opendaylight.yangtools.yang.common.Uint8;
-@RunWith(MockitoJUnitRunner.class)
+@RunWith(MockitoJUnitRunner.StrictStubs.class)
public class SalFlowServiceImplTest extends TestCase {
private static final Uint64 DUMMY_DATAPATH_ID = Uint64.valueOf(444);
= NODE_II.augmentation(FlowCapableNode.class).child(Table.class, new TableKey(DUMMY_TABLE_ID));
private final NodeRef noderef = new NodeRef(NODE_II);
- private static final String KEY = "0";
- private static FlowGroupCache flowcache =
- new FlowGroupCache("0","mock class", FlowGroupStatus.ADDED, LocalDateTime.MAX);
-
- private static Queue<FlowGroupCache> caches() {
- Queue<FlowGroupCache> cache = new LinkedList<>();
- cache.add(flowcache);
- return cache;
- }
-
- private static final Queue<FlowGroupCache> CACHE = caches();
-
- private static Map<String, Queue<FlowGroupCache>> createMap() {
- Map<String,Queue<FlowGroupCache>> myMap = new HashMap<>();
- myMap.put(KEY, CACHE);
- return myMap;
- }
-
- private static final Map<String, Queue<FlowGroupCache>> MYMAP = createMap();
@Mock
private RequestContextStack mockedRequestContextStack;
when(mockedDeviceInfo.getDatapathId()).thenReturn(DUMMY_DATAPATH_ID);
when(mockedDeviceContext.getDeviceInfo()).thenReturn(mockedDeviceInfo);
- when(flowGroupCacheManager.getAllNodesFlowGroupCache()).thenReturn(MYMAP);
}
private SalFlowServiceImpl mockSalFlowService(final short version) {
Code Review / openflowplugin.git / commitdiff