common/src/main/java/org/opendaylight/transportpce/common/network/RequestProcessor.java

   1 /*
   2  * Copyright © 2017 Orange, 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 package org.opendaylight.transportpce.common.network;
   9
  10 import com.google.common.base.Optional;
  11 import com.google.common.util.concurrent.CheckedFuture;
  12 import com.google.common.util.concurrent.ListenableFuture;
  13
  14 import java.util.concurrent.locks.ReentrantReadWriteLock;
  15
  16 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  17 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  18 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  19
  20 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  21 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  22
  23 import org.opendaylight.yangtools.yang.binding.DataObject;
  24 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28
  29
  30 public class RequestProcessor {
  31
  32     private static final Logger LOG = LoggerFactory.getLogger(RequestProcessor.class);
  33
  34     private final DataBroker dataBroker;
  35     private ReadWriteTransaction rwTx;
  36     private ReadOnlyTransaction readTx;
  37     private ReentrantReadWriteLock lock;
  38
  39
  40
  41     public RequestProcessor(DataBroker dataBroker) {
  42         this.dataBroker = dataBroker;
  43         rwTx = dataBroker.newReadWriteTransaction();
  44         readTx = dataBroker.newReadOnlyTransaction();
  45         lock = new ReentrantReadWriteLock();
  46         LOG.info("RequestProcessor instantiated");
  47
  48     }
  49
  50     public <T extends DataObject> CheckedFuture<Optional<T>,
  51         ReadFailedException> read(LogicalDatastoreType store,InstanceIdentifier<T> path) {
  52
  53         CheckedFuture<Optional<T>, ReadFailedException> result = null;
  54         acquireReadLock();
  55         LOG.debug("Number of threads in queue to read " + lock.getQueueLength());
  56         result = rwTx.read(store, path);
  57
  58         releaseReadLock();
  59         return result;
  60     }
  61
  62     public <T extends DataObject> void delete(LogicalDatastoreType store, InstanceIdentifier<?> path) {
  63
  64         acquireLock();
  65         LOG.info("Number of delete requests waiting in queue :" + lock.getQueueLength());
  66         rwTx.delete(store, path);
  67     }
  68
  69     public <T extends DataObject> void put(LogicalDatastoreType store,
  70         InstanceIdentifier<T> path, T data, boolean createMissingParents) {
  71
  72         acquireLock();
  73         LOG.debug("Number of put requests waiting in queue :" + lock.getQueueLength());
  74         rwTx.put(store, path, data, createMissingParents);
  75     }
  76
  77     public <T extends DataObject> void put(LogicalDatastoreType store,
  78         InstanceIdentifier<T> path, T data) {
  79
  80         acquireLock();
  81         LOG.debug("Number of put requests waiting in queue :" + lock.getQueueLength());
  82         rwTx.put(store, path, data);
  83     }
  84
  85
  86     public <T extends DataObject> void merge(LogicalDatastoreType store,
  87         InstanceIdentifier<T> path, T data, boolean createMissingParents) {
  88
  89         acquireLock();
  90         LOG.debug("Number of merge requests waiting in queue :" + lock.getQueueLength());
  91         rwTx.merge(store, path, data, createMissingParents);
  92     }
  93
  94     public <T extends DataObject> void merge(LogicalDatastoreType store,
  95         InstanceIdentifier<T> path, T data) {
  96
  97         acquireLock();
  98         LOG.debug("Number of merge requests waiting in queue :" + lock.getQueueLength());
  99         rwTx.merge(store, path, data);
 100     }
 101
 102     public ListenableFuture<Void> submit() {
 103         acquireLock();
 104         ListenableFuture<Void> future = null;
 105         future = rwTx.submit();
 106         releaseLock();
 107         resetRwTx();
 108         return future;
 109     }
 110
 111     public void close() {
 112         releaseLock();
 113     }
 114
 115     private void acquireLock() {
 116         if (!lock.writeLock().isHeldByCurrentThread()) {
 117             lock.writeLock().lock();
 118             LOG.debug("Number of write lock requests waiting in queue :" + lock.getQueueLength());
 119             LOG.info("Write Lock acquired by : " + Thread.currentThread().getName());
 120             rwTx = resetRwTx();
 121         } else {
 122             LOG.debug("Lock already acquired by : " + Thread.currentThread().getName());
 123         }
 124     }
 125
 126     private void acquireReadLock() {
 127         if (lock.getReadHoldCount() > 0) {
 128             LOG.info("Read Lock already acquired by : " + Thread.currentThread().getName());
 129         } else {
 130             lock.readLock().lock();
 131             rwTx = resetRwTx();
 132             LOG.info("Read Lock acquired by : " + Thread.currentThread().getName());
 133         }
 134     }
 135
 136     private void releaseLock() {
 137         if (lock.writeLock().isHeldByCurrentThread()) {
 138             LOG.info("Write Lock released by : " + Thread.currentThread().getName());
 139             lock.writeLock().unlock();
 140         }
 141     }
 142
 143     private void releaseReadLock() {
 144         LOG.info("Read Lock released by : " + Thread.currentThread().getName());
 145         lock.readLock().unlock();
 146     }
 147
 148     private ReadWriteTransaction resetRwTx() {
 149         LOG.info("Resetting the read write transaction .....");
 150         rwTx = dataBroker.newReadWriteTransaction();
 151         return rwTx;
 152     }
 153 }