interfacemanager/interfacemanager-impl/src/main/java/org/opendaylight/genius/interfacemanager/renderer/ovs/utilities/BatchingUtils.java

   1 /*
   2  * Copyright (c) 2016, 2017 Ericsson India Global Services Pvt Ltd. 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 package org.opendaylight.genius.interfacemanager.renderer.ovs.utilities;
   9
  10 import java.util.concurrent.BlockingQueue;
  11 import java.util.concurrent.LinkedBlockingQueue;
  12 import javax.annotation.PostConstruct;
  13 import javax.annotation.PreDestroy;
  14 import javax.inject.Inject;
  15 import javax.inject.Singleton;
  16 import org.apache.aries.blueprint.annotation.service.Reference;
  17 import org.opendaylight.genius.utils.batching.ActionableResource;
  18 import org.opendaylight.genius.utils.batching.ActionableResources;
  19 import org.opendaylight.genius.utils.batching.ResourceBatchingManager;
  20 import org.opendaylight.mdsal.binding.api.DataBroker;
  21 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  22 import org.opendaylight.yangtools.yang.binding.DataObject;
  23 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  24
  25 @Singleton
  26 public class BatchingUtils implements AutoCloseable {
  27     public enum EntityType {
  28         DEFAULT_CONFIG,
  29         DEFAULT_OPERATIONAL,
  30         TOPOLOGY_CONFIG
  31     }
  32
  33     private static final String DEFAULT_OPERATIONAL_RES_TYPE = "INTERFACEMGR-DEFAULT-OPERATIONAL";
  34     private static final String DEFAULT_CONFIG_RES_TYPE = "INTERFACEMGR-DEFAULT-CONFIG";
  35     private static final String TOPOLOGY_CONFIG_RES_TYPE = "INTERFACEMGR-TOPOLOGY-CONFIG";
  36
  37     private static final int DEFAULT_BATCH_SIZE = 1000;
  38     private static final int DEFAULT_BATCH_INTERVAL = 500;
  39
  40     private final BlockingQueue<ActionableResource<?>> topologyConfigShardBufferQ = new LinkedBlockingQueue<>();
  41     private final BlockingQueue<ActionableResource<?>> defaultConfigShardBufferQ = new LinkedBlockingQueue<>();
  42     private final BlockingQueue<ActionableResource<?>> defaultOperationalShardBufferQ = new LinkedBlockingQueue<>();
  43
  44     private final DataBroker dataBroker;
  45     private final ResourceBatchingManager resourceBatchingManager = ResourceBatchingManager.getInstance();
  46
  47     @Inject
  48     public BatchingUtils(@Reference DataBroker dataBroker) {
  49         this.dataBroker = dataBroker;
  50     }
  51
  52     @PostConstruct
  53     public void init() {
  54         int batchSize = Integer.getInteger("batch.size", DEFAULT_BATCH_SIZE);
  55         int batchInterval = Integer.getInteger("batch.wait.time", DEFAULT_BATCH_INTERVAL);
  56
  57         resourceBatchingManager.registerBatchableResource(TOPOLOGY_CONFIG_RES_TYPE, topologyConfigShardBufferQ,
  58                 new InterfaceBatchHandler(dataBroker, LogicalDatastoreType.CONFIGURATION, batchSize, batchInterval));
  59         resourceBatchingManager.registerBatchableResource(DEFAULT_CONFIG_RES_TYPE, defaultConfigShardBufferQ,
  60                 new InterfaceBatchHandler(dataBroker, LogicalDatastoreType.CONFIGURATION, batchSize, batchInterval));
  61         resourceBatchingManager.registerBatchableResource(DEFAULT_OPERATIONAL_RES_TYPE, defaultOperationalShardBufferQ,
  62                 new InterfaceBatchHandler(dataBroker, LogicalDatastoreType.OPERATIONAL, batchSize, batchInterval));
  63     }
  64
  65     @Override
  66     @PreDestroy
  67     public void close() {
  68         resourceBatchingManager.deregisterBatchableResource(TOPOLOGY_CONFIG_RES_TYPE);
  69         resourceBatchingManager.deregisterBatchableResource(DEFAULT_CONFIG_RES_TYPE);
  70         resourceBatchingManager.deregisterBatchableResource(DEFAULT_OPERATIONAL_RES_TYPE);
  71     }
  72
  73     <T extends DataObject> void update(InstanceIdentifier<T> path, T data, EntityType entityType) {
  74         getQueue(entityType).add(ActionableResources.update(path, data));
  75     }
  76
  77     public <T extends DataObject> void write(InstanceIdentifier<T> path, T data, EntityType entityType) {
  78         getQueue(entityType).add(ActionableResources.create(path, data));
  79     }
  80
  81     public BlockingQueue<ActionableResource<?>> getQueue(EntityType entityType) {
  82         switch (entityType) {
  83             case DEFAULT_CONFIG:
  84                 return defaultConfigShardBufferQ;
  85             case TOPOLOGY_CONFIG:
  86                 return topologyConfigShardBufferQ;
  87             case DEFAULT_OPERATIONAL:
  88                 return defaultOperationalShardBufferQ;
  89             default:
  90                 return null;
  91         }
  92     }
  93
  94     public <T extends DataObject> void delete(InstanceIdentifier<T> path, EntityType entityType) {
  95         getQueue(entityType).add(ActionableResources.delete(path));
  96     }
  97 }