* 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.impl.services.sal;
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Iterables;
+import static java.util.Objects.requireNonNull;
+
import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.JdkFutureAdapters;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.Future;
-import javax.annotation.Nullable;
+import java.util.stream.Collectors;
import org.opendaylight.openflowplugin.impl.util.BarrierUtil;
import org.opendaylight.openflowplugin.impl.util.GroupUtil;
import org.opendaylight.openflowplugin.impl.util.PathUtil;
import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupRef;
import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.AddGroupsBatchInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.AddGroupsBatchOutput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.BatchGroupInputUpdateGrouping;
import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.RemoveGroupsBatchInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.RemoveGroupsBatchOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.SalGroupsBatchService;
* Default implementation of {@link SalGroupsBatchService} - delegates work to {@link SalGroupService}.
*/
public class SalGroupsBatchServiceImpl implements SalGroupsBatchService {
-
private static final Logger LOG = LoggerFactory.getLogger(SalGroupsBatchServiceImpl.class);
private final SalGroupService salGroupService;
private final FlowCapableTransactionService transactionService;
- public SalGroupsBatchServiceImpl(final SalGroupService salGroupService, final FlowCapableTransactionService transactionService) {
- this.salGroupService = Preconditions.checkNotNull(salGroupService);
- this.transactionService = Preconditions.checkNotNull(transactionService);
+ public SalGroupsBatchServiceImpl(final SalGroupService salGroupService,
+ final FlowCapableTransactionService transactionService) {
+ this.salGroupService = requireNonNull(salGroupService);
+ this.transactionService = requireNonNull(transactionService);
}
@Override
- public Future<RpcResult<UpdateGroupsBatchOutput>> updateGroupsBatch(final UpdateGroupsBatchInput input) {
+ public ListenableFuture<RpcResult<UpdateGroupsBatchOutput>> updateGroupsBatch(final UpdateGroupsBatchInput input) {
final List<BatchUpdateGroups> batchUpdateGroups = input.getBatchUpdateGroups();
LOG.trace("Updating groups @ {} : {}", PathUtil.extractNodeId(input.getNode()), batchUpdateGroups.size());
final UpdateGroupInput updateGroupInput = new UpdateGroupInputBuilder(input)
.setOriginalGroup(new OriginalGroupBuilder(batchGroup.getOriginalBatchedGroup()).build())
.setUpdatedGroup(new UpdatedGroupBuilder(batchGroup.getUpdatedBatchedGroup()).build())
- .setGroupRef(createGroupRef(input.getNode(), batchGroup))
- .setNode(input.getNode())
- .build();
- resultsLot.add(JdkFutureAdapters.listenInPoolThread(salGroupService.updateGroup(updateGroupInput)));
+ .setGroupRef(createGroupRef(input.getNode(), batchGroup)).setNode(input.getNode()).build();
+ resultsLot.add(salGroupService.updateGroup(updateGroupInput));
}
- final Iterable<Group> groups = Iterables.transform(batchUpdateGroups, new Function<BatchUpdateGroups, Group>() {
- @Nullable
- @Override
- public Group apply(@Nullable final BatchUpdateGroups input) {
- return input.getUpdatedBatchedGroup();
- }
- }
- );
-
- final ListenableFuture<RpcResult<List<BatchFailedGroupsOutput>>> commonResult =
- Futures.transform(Futures.allAsList(resultsLot), GroupUtil.<UpdateGroupOutput>createCumulatingFunction(
- groups, batchUpdateGroups.size()));
-
- ListenableFuture<RpcResult<UpdateGroupsBatchOutput>> updateGroupsBulkFuture = Futures.transform(
- commonResult, GroupUtil.GROUP_UPDATE_TRANSFORM);
-
- if (input.isBarrierAfter()) {
- updateGroupsBulkFuture = BarrierUtil.chainBarrier(updateGroupsBulkFuture, input.getNode(),
- transactionService, GroupUtil.GROUP_UPDATE_COMPOSING_TRANSFORM);
+ final Iterable<Group> groups = batchUpdateGroups.stream()
+ .map(BatchGroupInputUpdateGrouping::getUpdatedBatchedGroup).collect(Collectors.toList());
+
+ final ListenableFuture<RpcResult<List<BatchFailedGroupsOutput>>> commonResult = Futures
+ .transform(Futures.allAsList(resultsLot),
+ GroupUtil.createCumulatingFunction(groups, batchUpdateGroups.size()),
+ MoreExecutors.directExecutor());
+
+ ListenableFuture<RpcResult<UpdateGroupsBatchOutput>> updateGroupsBulkFuture = Futures
+ .transform(commonResult, GroupUtil.GROUP_UPDATE_TRANSFORM, MoreExecutors.directExecutor());
+
+ if (input.getBarrierAfter()) {
+ updateGroupsBulkFuture = BarrierUtil
+ .chainBarrier(updateGroupsBulkFuture, input.getNode(), transactionService,
+ GroupUtil.GROUP_UPDATE_COMPOSING_TRANSFORM);
}
return updateGroupsBulkFuture;
}
@Override
- public Future<RpcResult<AddGroupsBatchOutput>> addGroupsBatch(final AddGroupsBatchInput input) {
+ public
ListenableFuture<RpcResult<AddGroupsBatchOutput>> addGroupsBatch(final AddGroupsBatchInput input) {
LOG.trace("Adding groups @ {} : {}", PathUtil.extractNodeId(input.getNode()), input.getBatchAddGroups().size());
final ArrayList<ListenableFuture<RpcResult<AddGroupOutput>>> resultsLot = new ArrayList<>();
- for (BatchAddGroups addGroup : input.getBatchAddGroups()) {
+ for (BatchAddGroups addGroup : input.nonnullBatchAddGroups().values()) {
final AddGroupInput addGroupInput = new AddGroupInputBuilder(addGroup)
- .setGroupRef(createGroupRef(input.getNode(), addGroup))
- .setNode(input.getNode())
- .build();
- resultsLot.add(JdkFutureAdapters.listenInPoolThread(salGroupService.addGroup(addGroupInput)));
+ .setGroupRef(createGroupRef(input.getNode(), addGroup)).setNode(input.getNode()).build();
+ resultsLot.add(salGroupService.addGroup(addGroupInput));
}
- final ListenableFuture<RpcResult<List<BatchFailedGroupsOutput>>> commonResult =
- Futures.transform(Futures.allAsList(resultsLot),
- GroupUtil.<AddGroupOutput>createCumulatingFunction(input.getBatchAddGroups()));
+ final ListenableFuture<RpcResult<List<BatchFailedGroupsOutput>>> commonResult = Futures
+ .transform(Futures.allAsList(resultsLot),
+ GroupUtil.createCumulatingFunction(input.nonnullBatchAddGroups().values()),
+ MoreExecutors.directExecutor());
- ListenableFuture<RpcResult<AddGroupsBatchOutput>> addGroupsBulkFuture =
- Futures.transform(commonResult, GroupUtil.GROUP_ADD_TRANSFORM);
+ ListenableFuture<RpcResult<AddGroupsBatchOutput>> addGroupsBulkFuture = Futures
+ .transform(commonResult, GroupUtil.GROUP_ADD_TRANSFORM, MoreExecutors.directExecutor());
- if (input.isBarrierAfter()) {
- addGroupsBulkFuture = BarrierUtil.chainBarrier(addGroupsBulkFuture, input.getNode(),
- transactionService, GroupUtil.GROUP_ADD_COMPOSING_TRANSFORM);
+ if (input.getBarrierAfter()) {
+ addGroupsBulkFuture = BarrierUtil.chainBarrier(addGroupsBulkFuture, input.getNode(), transactionService,
+ GroupUtil.GROUP_ADD_COMPOSING_TRANSFORM);
}
return addGroupsBulkFuture;
}
@Override
- public Future<RpcResult<RemoveGroupsBatchOutput>> removeGroupsBatch(final RemoveGroupsBatchInput input) {
- LOG.trace("Removing groups @ {} : {}", PathUtil.extractNodeId(input.getNode()), input.getBatchRemoveGroups().size());
+ public ListenableFuture<RpcResult<RemoveGroupsBatchOutput>> removeGroupsBatch(final RemoveGroupsBatchInput input) {
+ LOG.trace("Removing groups @ {} : {}", PathUtil.extractNodeId(input.getNode()),
+ input.getBatchRemoveGroups().size());
final ArrayList<ListenableFuture<RpcResult<RemoveGroupOutput>>> resultsLot = new ArrayList<>();
- for (BatchRemoveGroups addGroup : input.getBatchRemoveGroups()) {
+ for (BatchRemoveGroups addGroup : input.nonnullBatchRemoveGroups().values()) {
final RemoveGroupInput removeGroupInput = new RemoveGroupInputBuilder(addGroup)
- .setGroupRef(createGroupRef(input.getNode(), addGroup))
- .setNode(input.getNode())
- .build();
- resultsLot.add(JdkFutureAdapters.listenInPoolThread(salGroupService.removeGroup(removeGroupInput)));
+ .setGroupRef(createGroupRef(input.getNode(), addGroup)).setNode(input.getNode()).build();
+ resultsLot.add(salGroupService.removeGroup(removeGroupInput));
}
- final ListenableFuture<RpcResult<List<BatchFailedGroupsOutput>>> commonResult =
- Futures.transform(Futures.allAsList(resultsLot),
- GroupUtil.<RemoveGroupOutput>createCumulatingFunction(input.getBatchRemoveGroups()));
+ final ListenableFuture<RpcResult<List<BatchFailedGroupsOutput>>> commonResult = Futures
+ .transform(Futures.allAsList(resultsLot),
+ GroupUtil.createCumulatingFunction(input.nonnullBatchRemoveGroups().values()),
+ MoreExecutors.directExecutor());
- ListenableFuture<RpcResult<RemoveGroupsBatchOutput>> removeGroupsBulkFuture =
- Futures.transform(commonResult, GroupUtil.GROUP_REMOVE_TRANSFORM);
+ ListenableFuture<RpcResult<RemoveGroupsBatchOutput>> removeGroupsBulkFuture = Futures
+ .transform(commonResult, GroupUtil.GROUP_REMOVE_TRANSFORM, MoreExecutors.directExecutor());
- if (input.isBarrierAfter()) {
- removeGroupsBulkFuture = BarrierUtil.chainBarrier(removeGroupsBulkFuture, input.getNode(),
- transactionService, GroupUtil.GROUP_REMOVE_COMPOSING_TRANSFORM);
+ if (input.getBarrierAfter()) {
+ removeGroupsBulkFuture = BarrierUtil
+ .chainBarrier(removeGroupsBulkFuture, input.getNode(), transactionService,
+ GroupUtil.GROUP_REMOVE_COMPOSING_TRANSFORM);
}
return removeGroupsBulkFuture;
private static GroupRef createGroupRef(final NodeRef nodeRef, final BatchUpdateGroups batchGroup) {
return GroupUtil.buildGroupPath((InstanceIdentifier<Node>) nodeRef.getValue(),
- batchGroup.getUpdatedBatchedGroup().getGroupId());
+ batchGroup.getUpdatedBatchedGroup().getGroupId());
}
}
Code Review / openflowplugin.git / blobdiff