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