applications/forwardingrules-sync/src/main/java/org/opendaylight/openflowplugin/applications/frsync/impl/strategy/SyncPlanPushStrategyFlatBatchImpl.java

   1 /**
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.openflowplugin.applications.frsync.impl.strategy;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.collect.Iterators;
  13 import com.google.common.collect.PeekingIterator;
  14 import com.google.common.collect.Range;
  15 import com.google.common.util.concurrent.AsyncFunction;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.JdkFutureAdapters;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import java.util.ArrayList;
  21 import java.util.LinkedHashMap;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.concurrent.Future;
  25 import javax.annotation.Nullable;
  26 import org.opendaylight.openflowplugin.applications.frsync.SyncPlanPushStrategy;
  27 import org.opendaylight.openflowplugin.applications.frsync.impl.TableForwarder;
  28 import org.opendaylight.openflowplugin.applications.frsync.util.FxChainUtil;
  29 import org.opendaylight.openflowplugin.applications.frsync.util.ItemSyncBox;
  30 import org.opendaylight.openflowplugin.applications.frsync.util.PathUtil;
  31 import org.opendaylight.openflowplugin.applications.frsync.util.ReconcileUtil;
  32 import org.opendaylight.openflowplugin.applications.frsync.util.SyncCrudCounters;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.ProcessFlatBatchInput;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.ProcessFlatBatchInputBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.ProcessFlatBatchOutput;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.SalFlatBatchService;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.Batch;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.BatchBuilder;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.BatchChoice;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchAddFlowCase;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchAddFlowCaseBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchAddGroupCase;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchAddGroupCaseBuilder;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchAddMeterCase;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchAddMeterCaseBuilder;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchRemoveFlowCase;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchRemoveFlowCaseBuilder;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchRemoveGroupCase;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchRemoveGroupCaseBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchRemoveMeterCase;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchRemoveMeterCaseBuilder;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchUpdateFlowCase;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchUpdateFlowCaseBuilder;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchUpdateGroupCase;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchUpdateGroupCaseBuilder;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchUpdateMeterCase;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.FlatBatchUpdateMeterCaseBuilder;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.add.flow._case.FlatBatchAddFlow;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.add.flow._case.FlatBatchAddFlowBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.add.group._case.FlatBatchAddGroup;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.add.group._case.FlatBatchAddGroupBuilder;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.add.meter._case.FlatBatchAddMeter;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.add.meter._case.FlatBatchAddMeterBuilder;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.remove.flow._case.FlatBatchRemoveFlow;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.remove.flow._case.FlatBatchRemoveFlowBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.remove.group._case.FlatBatchRemoveGroup;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.remove.group._case.FlatBatchRemoveGroupBuilder;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.remove.meter._case.FlatBatchRemoveMeter;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.remove.meter._case.FlatBatchRemoveMeterBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.update.flow._case.FlatBatchUpdateFlow;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.update.flow._case.FlatBatchUpdateFlowBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.update.group._case.FlatBatchUpdateGroup;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.update.group._case.FlatBatchUpdateGroupBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.update.meter._case.FlatBatchUpdateMeter;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.input.batch.batch.choice.flat.batch.update.meter._case.FlatBatchUpdateMeterBuilder;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.flat.batch.service.rev160321.process.flat.batch.output.BatchFailure;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.input.update.grouping.OriginalBatchedFlowBuilder;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.input.update.grouping.UpdatedBatchedFlowBuilder;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.batch.group.input.update.grouping.OriginalBatchedGroupBuilder;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.groups.service.rev160315.batch.group.input.update.grouping.UpdatedBatchedGroupBuilder;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.meters.service.rev160316.batch.meter.input.update.grouping.OriginalBatchedMeterBuilder;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.meters.service.rev160316.batch.meter.input.update.grouping.UpdatedBatchedMeterBuilder;
  90 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  91 import org.opendaylight.yangtools.yang.common.RpcResult;
  92 import org.slf4j.Logger;
  93 import org.slf4j.LoggerFactory;
  94
  95 /**
  96  * Execute CRUD API for flow + group + meter involving flat-batch strategy.
  97  */
  98 public class SyncPlanPushStrategyFlatBatchImpl implements SyncPlanPushStrategy {
  99
 100     private static final Logger LOG = LoggerFactory.getLogger(SyncPlanPushStrategyFlatBatchImpl.class);
 101
 102     private SalFlatBatchService flatBatchService;
 103     private TableForwarder tableForwarder;
 104
 105     @Override
 106     public ListenableFuture<RpcResult<Void>> executeSyncStrategy(ListenableFuture<RpcResult<Void>> resultVehicle,
 107                                                                  final SynchronizationDiffInput diffInput,
 108                                                                  final SyncCrudCounters counters) {
 109         final InstanceIdentifier<FlowCapableNode> nodeIdent = diffInput.getNodeIdent();
 110         final NodeId nodeId = PathUtil.digNodeId(nodeIdent);
 111
 112         // prepare default (full) counts
 113         counters.getGroupCrudCounts().setAdded(ReconcileUtil.countTotalPushed(diffInput.getGroupsToAddOrUpdate()));
 114         counters.getGroupCrudCounts().setUpdated(ReconcileUtil.countTotalUpdated(diffInput.getGroupsToAddOrUpdate()));
 115         counters.getGroupCrudCounts().setRemoved(ReconcileUtil.countTotalPushed(diffInput.getGroupsToRemove()));
 116
 117         counters.getFlowCrudCounts().setAdded(ReconcileUtil.countTotalPushed(diffInput.getFlowsToAddOrUpdate().values()));
 118         counters.getFlowCrudCounts().setUpdated(ReconcileUtil.countTotalUpdated(diffInput.getFlowsToAddOrUpdate().values()));
 119         counters.getFlowCrudCounts().setRemoved(ReconcileUtil.countTotalPushed(diffInput.getFlowsToRemove().values()));
 120
 121         counters.getMeterCrudCounts().setAdded(diffInput.getMetersToAddOrUpdate().getItemsToPush().size());
 122         counters.getMeterCrudCounts().setUpdated(diffInput.getMetersToAddOrUpdate().getItemsToUpdate().size());
 123         counters.getMeterCrudCounts().setRemoved(diffInput.getMetersToRemove().getItemsToPush().size());
 124
 125         /* Tables - have to be pushed before groups */
 126         // TODO enable table-update when ready
 127         //resultVehicle = updateTableFeatures(nodeIdent, configTree);
 128
 129         resultVehicle = Futures.transform(resultVehicle, new AsyncFunction<RpcResult<Void>, RpcResult<Void>>() {
 130             @Override
 131             public ListenableFuture<RpcResult<Void>> apply(final RpcResult<Void> input) throws Exception {
 132                 if (!input.isSuccessful()) {
 133                     //TODO chain errors but not skip processing on first error return Futures.immediateFuture(input);
 134                     //final ListenableFuture<RpcResult<Void>> singleVoidUpdateResult = Futures.transform(
 135                     //        Futures.asList Arrays.asList(input, output),
 136                     //        ReconcileUtil.<UpdateFlowOutput>createRpcResultCondenser("TODO"));
 137                 }
 138
 139                 final List<Batch> batchBag = new ArrayList<>();
 140                 int batchOrder = 0;
 141
 142                 batchOrder = assembleAddOrUpdateGroups(batchBag, batchOrder, diffInput.getGroupsToAddOrUpdate());
 143                 batchOrder = assembleAddOrUpdateMeters(batchBag, batchOrder, diffInput.getMetersToAddOrUpdate());
 144                 batchOrder = assembleAddOrUpdateFlows(batchBag, batchOrder, diffInput.getFlowsToAddOrUpdate());
 145
 146                 batchOrder = assembleRemoveFlows(batchBag, batchOrder, diffInput.getFlowsToRemove());
 147                 batchOrder = assembleRemoveMeters(batchBag, batchOrder, diffInput.getMetersToRemove());
 148                 batchOrder = assembleRemoveGroups(batchBag, batchOrder, diffInput.getGroupsToRemove());
 149
 150                 LOG.trace("Index of last batch step: {}", batchOrder);
 151
 152                 final ProcessFlatBatchInput flatBatchInput = new ProcessFlatBatchInputBuilder()
 153                         .setNode(new NodeRef(PathUtil.digNodePath(diffInput.getNodeIdent())))
 154                         .setExitOnFirstError(false) // TODO: propagate from input
 155                         .setBatch(batchBag)
 156                         .build();
 157
 158                 final Future<RpcResult<ProcessFlatBatchOutput>> rpcResultFuture = flatBatchService.processFlatBatch(flatBatchInput);
 159
 160                 final int failureIndexLimit = batchOrder;
 161
 162                 if (LOG.isDebugEnabled()) {
 163                     Futures.addCallback(JdkFutureAdapters.listenInPoolThread(rpcResultFuture),
 164                             createCounterCallback(batchBag, failureIndexLimit, counters));
 165                 }
 166
 167                 return Futures.transform(JdkFutureAdapters.listenInPoolThread(rpcResultFuture),
 168                         ReconcileUtil.<ProcessFlatBatchOutput>createRpcResultToVoidFunction("flat-batch"));
 169             }
 170         });
 171
 172         Futures.addCallback(resultVehicle, FxChainUtil.logResultCallback(nodeId, "flat-batch"));
 173         return resultVehicle;
 174     }
 175
 176     private FutureCallback<RpcResult<ProcessFlatBatchOutput>> createCounterCallback(final List<Batch> inputBatchBag,
 177                                                                                     final int failureIndexLimit,
 178                                                                                     final SyncCrudCounters counters) {
 179         return new FutureCallback<RpcResult<ProcessFlatBatchOutput>>() {
 180             @Override
 181             public void onSuccess(@Nullable final RpcResult<ProcessFlatBatchOutput> result) {
 182                 if (!result.isSuccessful() && result.getResult() != null && !result.getResult().getBatchFailure().isEmpty()) {
 183                     Map<Range<Integer>, Batch> batchMap = mapBachesToRanges(inputBatchBag, failureIndexLimit);
 184
 185                     for (BatchFailure batchFailure : result.getResult().getBatchFailure()) {
 186                         for (Map.Entry<Range<Integer>, Batch> rangeBatchEntry : batchMap.entrySet()) {
 187                             if (rangeBatchEntry.getKey().contains(batchFailure.getBatchOrder())) {
 188                                 // get type and decrease
 189                                 final BatchChoice batchChoice = rangeBatchEntry.getValue().getBatchChoice();
 190                                 decrementCounters(batchChoice, counters);
 191                                 break;
 192                             }
 193                         }
 194                     }
 195                 }
 196             }
 197
 198             @Override
 199             public void onFailure(final Throwable t) {
 200                 counters.resetAll();
 201             }
 202         };
 203     }
 204
 205     static void decrementCounters(final BatchChoice batchChoice, final SyncCrudCounters counters) {
 206         if (batchChoice instanceof FlatBatchAddFlowCase) {
 207             counters.getFlowCrudCounts().decAdded();
 208         } else if (batchChoice instanceof FlatBatchUpdateFlowCase) {
 209             counters.getFlowCrudCounts().decUpdated();
 210         } else if (batchChoice instanceof FlatBatchRemoveFlowCase) {
 211             counters.getFlowCrudCounts().decRemoved();
 212         } else if (batchChoice instanceof FlatBatchAddGroupCase) {
 213             counters.getGroupCrudCounts().decAdded();
 214         } else if (batchChoice instanceof FlatBatchUpdateGroupCase) {
 215             counters.getGroupCrudCounts().decUpdated();
 216         } else if (batchChoice instanceof FlatBatchRemoveGroupCase) {
 217             counters.getGroupCrudCounts().decRemoved();
 218         } else if (batchChoice instanceof FlatBatchAddMeterCase) {
 219             counters.getMeterCrudCounts().decAdded();
 220         } else if (batchChoice instanceof FlatBatchUpdateMeterCase) {
 221             counters.getMeterCrudCounts().decUpdated();
 222         } else if (batchChoice instanceof FlatBatchRemoveMeterCase) {
 223             counters.getMeterCrudCounts().decRemoved();
 224         }
 225     }
 226
 227     static Map<Range<Integer>, Batch> mapBachesToRanges(final List<Batch> inputBatchBag, final int failureIndexLimit) {
 228         final Map<Range<Integer>, Batch> batchMap = new LinkedHashMap<>();
 229         final PeekingIterator<Batch> batchPeekingIterator = Iterators.peekingIterator(inputBatchBag.iterator());
 230         while (batchPeekingIterator.hasNext()) {
 231             final Batch batch = batchPeekingIterator.next();
 232             final int nextBatchOrder = batchPeekingIterator.hasNext()
 233                     ? batchPeekingIterator.peek().getBatchOrder()
 234                     : failureIndexLimit;
 235             batchMap.put(Range.closed(batch.getBatchOrder(), nextBatchOrder - 1), batch);
 236         }
 237         return batchMap;
 238     }
 239
 240     private int getNextBatchLimit(final PeekingIterator<Batch> inputBatchIterator, final int failureIndexLimit) {
 241         return inputBatchIterator.hasNext()
 242                 ? inputBatchIterator.peek().getBatchOrder()
 243                 : failureIndexLimit;
 244     }
 245
 246     @VisibleForTesting
 247     static int assembleRemoveFlows(final List<Batch> batchBag, int batchOrder, final Map<TableKey, ItemSyncBox<Flow>> flowItemSyncTableMap) {
 248         // process flow remove
 249         if (flowItemSyncTableMap != null) {
 250             for (Map.Entry<TableKey, ItemSyncBox<Flow>> syncBoxEntry : flowItemSyncTableMap.entrySet()) {
 251                 final TableKey tableKey = syncBoxEntry.getKey();
 252                 final ItemSyncBox<Flow> flowItemSyncBox = syncBoxEntry.getValue();
 253
 254                 if (!flowItemSyncBox.getItemsToPush().isEmpty()) {
 255                     final List<FlatBatchRemoveFlow> flatBatchRemoveFlowBag =
 256                             new ArrayList<>(flowItemSyncBox.getItemsToUpdate().size());
 257                     int itemOrder = 0;
 258                     for (Flow flow : flowItemSyncBox.getItemsToPush()) {
 259                         flatBatchRemoveFlowBag.add(new FlatBatchRemoveFlowBuilder(flow)
 260                                 .setBatchOrder(itemOrder++)
 261                                 .setFlowId(flow.getId())
 262                                 .build());
 263                     }
 264                     final Batch batch = new BatchBuilder()
 265                             .setBatchChoice(new FlatBatchRemoveFlowCaseBuilder()
 266                                     .setFlatBatchRemoveFlow(flatBatchRemoveFlowBag)
 267                                     .build())
 268                             .setBatchOrder(batchOrder)
 269                             .build();
 270                     batchOrder += itemOrder;
 271                     batchBag.add(batch);
 272                 }
 273             }
 274         }
 275         return batchOrder;
 276     }
 277
 278     @VisibleForTesting
 279     static int assembleAddOrUpdateGroups(final List<Batch> batchBag, int batchOrder, final List<ItemSyncBox<Group>> groupsToAddOrUpdate) {
 280         // process group add+update
 281         if (groupsToAddOrUpdate != null) {
 282             for (ItemSyncBox<Group> groupItemSyncBox : groupsToAddOrUpdate) {
 283                 if (!groupItemSyncBox.getItemsToPush().isEmpty()) {
 284                     final List<FlatBatchAddGroup> flatBatchAddGroupBag =
 285                             new ArrayList<>(groupItemSyncBox.getItemsToUpdate().size());
 286                     int itemOrder = 0;
 287                     for (Group group : groupItemSyncBox.getItemsToPush()) {
 288                         flatBatchAddGroupBag.add(new FlatBatchAddGroupBuilder(group).setBatchOrder(itemOrder++).build());
 289                     }
 290                     final Batch batch = new BatchBuilder()
 291                             .setBatchChoice(new FlatBatchAddGroupCaseBuilder()
 292                                     .setFlatBatchAddGroup(flatBatchAddGroupBag)
 293                                     .build())
 294                             .setBatchOrder(batchOrder)
 295                             .build();
 296                     batchOrder += itemOrder;
 297                     batchBag.add(batch);
 298                 }
 299
 300                 if (!groupItemSyncBox.getItemsToUpdate().isEmpty()) {
 301                     final List<FlatBatchUpdateGroup> flatBatchUpdateGroupBag =
 302                             new ArrayList<>(groupItemSyncBox.getItemsToUpdate().size());
 303                     int itemOrder = 0;
 304                     for (ItemSyncBox.ItemUpdateTuple<Group> groupUpdate : groupItemSyncBox.getItemsToUpdate()) {
 305                         flatBatchUpdateGroupBag.add(new FlatBatchUpdateGroupBuilder()
 306                                 .setBatchOrder(itemOrder++)
 307                                 .setOriginalBatchedGroup(new OriginalBatchedGroupBuilder(groupUpdate.getOriginal()).build())
 308                                 .setUpdatedBatchedGroup(new UpdatedBatchedGroupBuilder(groupUpdate.getUpdated()).build())
 309                                 .build());
 310                     }
 311                     final Batch batch = new BatchBuilder()
 312                             .setBatchChoice(new FlatBatchUpdateGroupCaseBuilder()
 313                                     .setFlatBatchUpdateGroup(flatBatchUpdateGroupBag)
 314                                     .build())
 315                             .setBatchOrder(batchOrder)
 316                             .build();
 317                     batchOrder += itemOrder;
 318                     batchBag.add(batch);
 319                 }
 320             }
 321         }
 322         return batchOrder;
 323     }
 324
 325     @VisibleForTesting
 326     static int assembleRemoveGroups(final List<Batch> batchBag, int batchOrder, final List<ItemSyncBox<Group>> groupsToRemoveOrUpdate) {
 327         // process group add+update
 328         if (groupsToRemoveOrUpdate != null) {
 329             for (ItemSyncBox<Group> groupItemSyncBox : groupsToRemoveOrUpdate) {
 330                 if (!groupItemSyncBox.getItemsToPush().isEmpty()) {
 331                     final List<FlatBatchRemoveGroup> flatBatchRemoveGroupBag =
 332                             new ArrayList<>(groupItemSyncBox.getItemsToUpdate().size());
 333                     int itemOrder = 0;
 334                     for (Group group : groupItemSyncBox.getItemsToPush()) {
 335                         flatBatchRemoveGroupBag.add(new FlatBatchRemoveGroupBuilder(group).setBatchOrder(itemOrder++).build());
 336                     }
 337                     final Batch batch = new BatchBuilder()
 338                             .setBatchChoice(new FlatBatchRemoveGroupCaseBuilder()
 339                                     .setFlatBatchRemoveGroup(flatBatchRemoveGroupBag)
 340                                     .build())
 341                             .setBatchOrder(batchOrder)
 342                             .build();
 343                     batchOrder += itemOrder;
 344                     batchBag.add(batch);
 345                 }
 346             }
 347         }
 348         return batchOrder;
 349     }
 350
 351     @VisibleForTesting
 352     static int assembleAddOrUpdateMeters(final List<Batch> batchBag, int batchOrder, final ItemSyncBox<Meter> meterItemSyncBox) {
 353         // process meter add+update
 354         if (meterItemSyncBox != null) {
 355             if (!meterItemSyncBox.getItemsToPush().isEmpty()) {
 356                 final List<FlatBatchAddMeter> flatBatchAddMeterBag =
 357                         new ArrayList<>(meterItemSyncBox.getItemsToUpdate().size());
 358                 int itemOrder = 0;
 359                 for (Meter meter : meterItemSyncBox.getItemsToPush()) {
 360                     flatBatchAddMeterBag.add(new FlatBatchAddMeterBuilder(meter).setBatchOrder(itemOrder++).build());
 361                 }
 362                 final Batch batch = new BatchBuilder()
 363                         .setBatchChoice(new FlatBatchAddMeterCaseBuilder()
 364                                 .setFlatBatchAddMeter(flatBatchAddMeterBag)
 365                                 .build())
 366                         .setBatchOrder(batchOrder)
 367                         .build();
 368                 batchOrder += itemOrder;
 369                 batchBag.add(batch);
 370             }
 371
 372             if (!meterItemSyncBox.getItemsToUpdate().isEmpty()) {
 373                 final List<FlatBatchUpdateMeter> flatBatchUpdateMeterBag =
 374                         new ArrayList<>(meterItemSyncBox.getItemsToUpdate().size());
 375                 int itemOrder = 0;
 376                 for (ItemSyncBox.ItemUpdateTuple<Meter> meterUpdate : meterItemSyncBox.getItemsToUpdate()) {
 377                     flatBatchUpdateMeterBag.add(new FlatBatchUpdateMeterBuilder()
 378                             .setBatchOrder(itemOrder++)
 379                             .setOriginalBatchedMeter(new OriginalBatchedMeterBuilder(meterUpdate.getOriginal()).build())
 380                             .setUpdatedBatchedMeter(new UpdatedBatchedMeterBuilder(meterUpdate.getUpdated()).build())
 381                             .build());
 382                 }
 383                 final Batch batch = new BatchBuilder()
 384                         .setBatchChoice(new FlatBatchUpdateMeterCaseBuilder()
 385                                 .setFlatBatchUpdateMeter(flatBatchUpdateMeterBag)
 386                                 .build())
 387                         .setBatchOrder(batchOrder)
 388                         .build();
 389                 batchOrder += itemOrder;
 390                 batchBag.add(batch);
 391             }
 392         }
 393         return batchOrder;
 394     }
 395
 396     @VisibleForTesting
 397     static int assembleRemoveMeters(final List<Batch> batchBag, int batchOrder, final ItemSyncBox<Meter> meterItemSyncBox) {
 398         // process meter remove
 399         if (meterItemSyncBox != null) {
 400             if (!meterItemSyncBox.getItemsToPush().isEmpty()) {
 401                 final List<FlatBatchRemoveMeter> flatBatchRemoveMeterBag =
 402                         new ArrayList<>(meterItemSyncBox.getItemsToUpdate().size());
 403                 int itemOrder = 0;
 404                 for (Meter meter : meterItemSyncBox.getItemsToPush()) {
 405                     flatBatchRemoveMeterBag.add(new FlatBatchRemoveMeterBuilder(meter).setBatchOrder(itemOrder++).build());
 406                 }
 407                 final Batch batch = new BatchBuilder()
 408                         .setBatchChoice(new FlatBatchRemoveMeterCaseBuilder()
 409                                 .setFlatBatchRemoveMeter(flatBatchRemoveMeterBag)
 410                                 .build())
 411                         .setBatchOrder(batchOrder)
 412                         .build();
 413                 batchOrder += itemOrder;
 414                 batchBag.add(batch);
 415             }
 416         }
 417         return batchOrder;
 418     }
 419
 420     @VisibleForTesting
 421     static int assembleAddOrUpdateFlows(final List<Batch> batchBag, int batchOrder, final Map<TableKey, ItemSyncBox<Flow>> flowItemSyncTableMap) {
 422         // process flow add+update
 423         if (flowItemSyncTableMap != null) {
 424             for (Map.Entry<TableKey, ItemSyncBox<Flow>> syncBoxEntry : flowItemSyncTableMap.entrySet()) {
 425                 final TableKey tableKey = syncBoxEntry.getKey();
 426                 final ItemSyncBox<Flow> flowItemSyncBox = syncBoxEntry.getValue();
 427
 428                 if (!flowItemSyncBox.getItemsToPush().isEmpty()) {
 429                     final List<FlatBatchAddFlow> flatBatchAddFlowBag =
 430                             new ArrayList<>(flowItemSyncBox.getItemsToUpdate().size());
 431                     int itemOrder = 0;
 432                     for (Flow flow : flowItemSyncBox.getItemsToPush()) {
 433                         flatBatchAddFlowBag.add(new FlatBatchAddFlowBuilder(flow)
 434                                 .setBatchOrder(itemOrder++)
 435                                 .setFlowId(flow.getId())
 436                                 .build());
 437                     }
 438                     final Batch batch = new BatchBuilder()
 439                             .setBatchChoice(new FlatBatchAddFlowCaseBuilder()
 440                                     .setFlatBatchAddFlow(flatBatchAddFlowBag)
 441                                     .build())
 442                             .setBatchOrder(batchOrder)
 443                             .build();
 444                     batchOrder += itemOrder;
 445                     batchBag.add(batch);
 446                 }
 447
 448                 if (!flowItemSyncBox.getItemsToUpdate().isEmpty()) {
 449                     final List<FlatBatchUpdateFlow> flatBatchUpdateFlowBag =
 450                             new ArrayList<>(flowItemSyncBox.getItemsToUpdate().size());
 451                     int itemOrder = 0;
 452                     for (ItemSyncBox.ItemUpdateTuple<Flow> flowUpdate : flowItemSyncBox.getItemsToUpdate()) {
 453                         flatBatchUpdateFlowBag.add(new FlatBatchUpdateFlowBuilder()
 454                                 .setBatchOrder(itemOrder++)
 455                                 .setFlowId(flowUpdate.getUpdated().getId())
 456                                 .setOriginalBatchedFlow(new OriginalBatchedFlowBuilder(flowUpdate.getOriginal()).build())
 457                                 .setUpdatedBatchedFlow(new UpdatedBatchedFlowBuilder(flowUpdate.getUpdated()).build())
 458                                 .build());
 459                     }
 460                     final Batch batch = new BatchBuilder()
 461                             .setBatchChoice(new FlatBatchUpdateFlowCaseBuilder()
 462                                     .setFlatBatchUpdateFlow(flatBatchUpdateFlowBag)
 463                                     .build())
 464                             .setBatchOrder(batchOrder)
 465                             .build();
 466                     batchOrder += itemOrder;
 467                     batchBag.add(batch);
 468                 }
 469             }
 470         }
 471         return batchOrder;
 472     }
 473
 474     public SyncPlanPushStrategyFlatBatchImpl setFlatBatchService(final SalFlatBatchService flatBatchService) {
 475         this.flatBatchService = flatBatchService;
 476         return this;
 477     }
 478
 479     public SyncPlanPushStrategyFlatBatchImpl setTableForwarder(final TableForwarder tableForwarder) {
 480         this.tableForwarder = tableForwarder;
 481         return this;
 482     }
 483 }