import com.google.common.base.Function;
import com.google.common.base.Optional;
-import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
import java.util.stream.Collectors;
import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceRegistry;
if (!rpcResultIsNull && rpcResult.isSuccessful()) {
LOG.debug("Stats reply successfully received for node {} of type {}", deviceInfo.getNodeId(), type);
- // TODO: in case the result value is null then multipart data probably got processed
- // TODO: on the fly. This contract should by clearly stated and enforced.
- // TODO: Now simple true value is returned
+ // TODO: in case the result value is null then multipart data probably got processed
+ // TODO: on the fly. This contract should by clearly stated and enforced.
+ // TODO: Now simple true value is returned
if (Objects.nonNull(rpcResult.getResult()) && !rpcResult.getResult().isEmpty()) {
final List<DataContainer> allMultipartData = rpcResult.getResult().stream()
.map(reply -> MultipartReplyTranslatorUtil
- .translate(reply, deviceInfo, convertorExecutor, null))
+ .translate(reply, deviceInfo, convertorExecutor, null))
.filter(java.util.Optional::isPresent).map(java.util.Optional::get)
- .collect(Collectors.toList());
+ .collect(Collectors.toList());
return processStatistics(type, allMultipartData, txFacade, registry, deviceInfo,
- statisticsWriterProvider);
+ statisticsWriterProvider);
} else {
LOG.debug("Stats reply was empty for node {} of type {}", deviceInfo.getNodeId(), type);
}
} else {
LOG.warn("Stats reply FAILED for node {} of type {}: {}", deviceInfo.getNodeId(), type,
- rpcResultIsNull ? "" : rpcResult.getErrors());
+ rpcResultIsNull ? "" : rpcResult.getErrors());
}
return false;
- }));
+ }), MoreExecutors.directExecutor());
}
private static boolean processStatistics(final MultipartType type, final List<? extends DataContainer> statistics,
switch (type) {
case OFPMPFLOW:
deleteAllKnownFlows(txFacade, instanceIdentifier, deviceRegistry.getDeviceFlowRegistry());
+ deviceRegistry.getDeviceFlowRegistry().processMarks();
break;
case OFPMPMETERCONFIG:
deleteAllKnownMeters(txFacade, instanceIdentifier, deviceRegistry.getDeviceMeterRegistry());
+ deviceRegistry.getDeviceMeterRegistry().processMarks();
break;
case OFPMPGROUPDESC:
deleteAllKnownGroups(txFacade, instanceIdentifier, deviceRegistry.getDeviceGroupRegistry());
+ deviceRegistry.getDeviceGroupRegistry().processMarks();
break;
default:
// no operation
if (writeStatistics(type, statistics, deviceInfo, statisticsWriterProvider)) {
txFacade.submitTransaction();
- switch (type) {
- case OFPMPFLOW:
- deviceRegistry.getDeviceFlowRegistry().processMarks();
- break;
- case OFPMPMETERCONFIG:
- deviceRegistry.getDeviceMeterRegistry().processMarks();
- break;
- case OFPMPGROUPDESC:
- deviceRegistry.getDeviceGroupRegistry().processMarks();
- break;
- default:
- // no operation
- }
-
LOG.debug("Stats reply added to transaction for node {} of type {}", deviceInfo.getNodeId(), type);
return true;
}
return;
}
- final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> future;
+ final ListenableFuture<Optional<FlowCapableNode>> future;
try (ReadOnlyTransaction readTx = txFacade.getReadTransaction()) {
future = readTx.read(LogicalDatastoreType.OPERATIONAL, instanceIdentifier);
}
try {
Futures.transform(Futures.catchingAsync(future, Throwable.class, throwable -> {
return Futures.immediateFailedFuture(throwable);
- }), (Function<Optional<FlowCapableNode>, Void>) flowCapNodeOpt -> {
+ }, MoreExecutors.directExecutor()), (Function<Optional<FlowCapableNode>, Void>) flowCapNodeOpt -> {
// we have to read actual tables with all information before we set empty Flow list,
// merge is expensive and not applicable for lists
if (flowCapNodeOpt != null && flowCapNodeOpt.isPresent()) {
for (final Table tableData : flowCapNodeOpt.get().getTable()) {
final Table table = new TableBuilder(tableData).setFlow(Collections.emptyList()).build();
final InstanceIdentifier<Table> iiToTable = instanceIdentifier
- .child(Table.class, tableData.getKey());
+ .child(Table.class, tableData.key());
txFacade.writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToTable, table);
}
}
return null;
- }).get();
+ }, MoreExecutors.directExecutor()).get();
} catch (InterruptedException | ExecutionException ex) {
- LOG.debug("Failed to delete {} flows, exception: {}", deviceFlowRegistry.size(), ex);
+ LOG.debug("Failed to delete {} flows", deviceFlowRegistry.size(), ex);
}
}
public static void deleteAllKnownMeters(final TxFacade txFacade,
final InstanceIdentifier<FlowCapableNode> instanceIdentifier,
final DeviceMeterRegistry meterRegistry) {
- meterRegistry.forEach(meterId -> txFacade.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL,
- instanceIdentifier.child(Meter.class,
- new MeterKey(meterId))));
+ meterRegistry.forEach(meterId -> {
+ txFacade
+ .addDeleteToTxChain(
+ LogicalDatastoreType.OPERATIONAL,
+ instanceIdentifier.child(Meter.class, new MeterKey(meterId)));
+ meterRegistry.addMark(meterId);
+ });
}
public static void deleteAllKnownGroups(final TxFacade txFacade,
final InstanceIdentifier<FlowCapableNode> instanceIdentifier,
final DeviceGroupRegistry groupRegistry) {
- groupRegistry.forEach(groupId -> txFacade.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL,
- instanceIdentifier.child(Group.class,
- new GroupKey(groupId))));
+ LOG.debug("deleteAllKnownGroups on device targetType {}", instanceIdentifier.getTargetType());
+ groupRegistry.forEach(groupId -> {
+ txFacade
+ .addDeleteToTxChain(
+ LogicalDatastoreType.OPERATIONAL,
+ instanceIdentifier.child(Group.class, new GroupKey(groupId)));
+ groupRegistry.addMark(groupId);
+ });
}
/**
* Writes snapshot gathering start timestamp + cleans end mark.
*
* @param deviceInfo device info
- * @param txFacade tx manager
+ * @param txFacade tx manager
*/
static void markDeviceStateSnapshotStart(final DeviceInfo deviceInfo, final TxFacade txFacade) {
final InstanceIdentifier<FlowCapableStatisticsGatheringStatus> statusPath = deviceInfo
final SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_AND_TIME_FORMAT);
final FlowCapableStatisticsGatheringStatus gatheringStatus = new FlowCapableStatisticsGatheringStatusBuilder()
.setSnapshotGatheringStatusStart(new SnapshotGatheringStatusStartBuilder()
- .setBegin(new DateAndTime(simpleDateFormat.format(new Date())))
- .build())
+ .setBegin(new DateAndTime(simpleDateFormat.format(new Date())))
+ .build())
.setSnapshotGatheringStatusEnd(null) // TODO: reconsider if really need to clean end mark here
.build();
try {
* Writes snapshot gathering end timestamp + outcome.
*
* @param deviceInfo device info
- * @param txFacade tx manager
- * @param succeeded outcome of currently finished gathering
+ * @param txFacade tx manager
+ * @param succeeded outcome of currently finished gathering
*/
- static void markDeviceStateSnapshotEnd(final DeviceInfo deviceInfo, final TxFacade txFacade,
- final boolean succeeded) {
- final InstanceIdentifier<SnapshotGatheringStatusEnd> statusEndPath = deviceInfo.getNodeInstanceIdentifier()
- .augmentation(FlowCapableStatisticsGatheringStatus.class).child(SnapshotGatheringStatusEnd.class);
+ static void markDeviceStateSnapshotEnd(final DeviceInfo deviceInfo,
+ final TxFacade txFacade, final boolean succeeded) {
+ final InstanceIdentifier<SnapshotGatheringStatusEnd> statusEndPath = deviceInfo
+ .getNodeInstanceIdentifier().augmentation(FlowCapableStatisticsGatheringStatus.class)
+ .child(SnapshotGatheringStatusEnd.class);
final SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_AND_TIME_FORMAT);
final SnapshotGatheringStatusEnd gatheringStatus = new SnapshotGatheringStatusEndBuilder()
- .setEnd(new DateAndTime(simpleDateFormat.format(new Date()))).setSucceeded(succeeded).build();
+ .setEnd(new DateAndTime(simpleDateFormat.format(new Date())))
+ .setSucceeded(succeeded)
+ .build();
try {
txFacade.writeToTransaction(LogicalDatastoreType.OPERATIONAL, statusEndPath, gatheringStatus);
} catch (TransactionChainClosedException e) {
Code Review / openflowplugin.git / blobdiff