2 * Copyright © 2017 Orange, Inc. and others. All rights reserved.
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
8 package org.opendaylight.transportpce.common.network;
10 import com.google.common.base.Optional;
11 import com.google.common.util.concurrent.ListenableFuture;
13 import java.util.concurrent.locks.ReentrantReadWriteLock;
15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
16 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
17 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
19 import org.opendaylight.controller.md.sal.common.api.data.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;
26 public class RequestProcessor {
28 private static final Logger LOG = LoggerFactory.getLogger(RequestProcessor.class);
30 private final DataBroker dataBroker;
31 private ReadWriteTransaction rwTx;
32 private ReadOnlyTransaction readTx;
33 private ReentrantReadWriteLock lock;
37 public RequestProcessor(DataBroker dataBroker) {
38 this.dataBroker = dataBroker;
39 rwTx = dataBroker.newReadWriteTransaction();
40 readTx = dataBroker.newReadOnlyTransaction();
41 lock = new ReentrantReadWriteLock();
42 LOG.info("RequestProcessor instantiated");
46 public <T extends DataObject> ListenableFuture<Optional<T>>
47 read(LogicalDatastoreType store,InstanceIdentifier<T> path) {
49 ListenableFuture<Optional<T>> result = null;
51 LOG.debug("Number of threads in queue to read {}", lock.getQueueLength());
52 result = rwTx.read(store, path);
58 public <T extends DataObject> void delete(LogicalDatastoreType store, InstanceIdentifier<?> path) {
61 LOG.info("Number of delete requests waiting in queue :{}", lock.getQueueLength());
62 rwTx.delete(store, path);
65 public <T extends DataObject> void put(LogicalDatastoreType store,
66 InstanceIdentifier<T> path, T data, boolean createMissingParents) {
69 LOG.debug("Number of put requests waiting in queue :{}", lock.getQueueLength());
70 rwTx.put(store, path, data, createMissingParents);
73 public <T extends DataObject> void put(LogicalDatastoreType store,
74 InstanceIdentifier<T> path, T data) {
77 LOG.debug("Number of put requests waiting in queue :{}", lock.getQueueLength());
78 rwTx.put(store, path, data);
82 public <T extends DataObject> void merge(LogicalDatastoreType store,
83 InstanceIdentifier<T> path, T data, boolean createMissingParents) {
86 LOG.debug("Number of merge requests waiting in queue :{}", lock.getQueueLength());
87 rwTx.merge(store, path, data, createMissingParents);
90 public <T extends DataObject> void merge(LogicalDatastoreType store,
91 InstanceIdentifier<T> path, T data) {
94 LOG.debug("Number of merge requests waiting in queue :{}", lock.getQueueLength());
95 rwTx.merge(store, path, data);
98 public ListenableFuture<Void> submit() {
100 ListenableFuture<Void> future = null;
101 future = rwTx.submit();
107 public void close() {
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());
118 LOG.debug("Lock already acquired by : {}", Thread.currentThread().getName());
122 private void acquireReadLock() {
123 if (lock.getReadHoldCount() > 0) {
124 LOG.info("Read Lock already acquired by : {}", Thread.currentThread().getName());
126 lock.readLock().lock();
128 LOG.info("Read Lock acquired by : {}", Thread.currentThread().getName());
132 private void releaseLock() {
133 if (lock.writeLock().isHeldByCurrentThread()) {
134 LOG.info("Write Lock released by : {}", Thread.currentThread().getName());
135 lock.writeLock().unlock();
139 private void releaseReadLock() {
140 LOG.info("Read Lock released by : {}", Thread.currentThread().getName());
141 lock.readLock().unlock();
144 private ReadWriteTransaction resetRwTx() {
145 LOG.info("Resetting the read write transaction .....");
146 rwTx = dataBroker.newReadWriteTransaction();