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