2 * ========================LICENSE_START=================================
5 * Copyright (C) 2019 Nordix Foundation
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ========================LICENSE_END===================================
21 package org.oransc.policyagent.repository;
23 import java.util.ArrayList;
24 import java.util.Iterator;
25 import java.util.LinkedList;
26 import java.util.List;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
31 import reactor.core.publisher.Mono;
32 import reactor.core.publisher.MonoSink;
35 * A resource lock. Exclusive means that the caller takes exclusive ownership of
36 * the resurce. Non exclusive lock means that several users can lock the
37 * resource (for shared usage).
40 private static final Logger logger = LoggerFactory.getLogger(Lock.class);
42 private boolean isExclusive = false;
43 private int lockCounter = 0;
44 private final List<LockRequest> lockRequestQueue = new LinkedList<>();
46 private static class AsynchCallbackExecutor implements Runnable {
47 private List<LockRequest> lockRequestQueue = new LinkedList<>();
49 public AsynchCallbackExecutor() {
50 Thread thread = new Thread(this);
54 public synchronized void addAll(List<LockRequest> requests) {
55 this.lockRequestQueue.addAll(requests);
62 for (LockRequest request : consume()) {
63 request.callback.success(request.lock);
69 private synchronized List<LockRequest> consume() {
70 List<LockRequest> q = this.lockRequestQueue;
71 this.lockRequestQueue = new LinkedList<>();
75 private synchronized void waitForNewEntries() {
77 if (this.lockRequestQueue.isEmpty()) {
80 } catch (InterruptedException e) {
81 logger.warn("waitForUnlock interrupted", e);
86 private static AsynchCallbackExecutor callbackProcessor = new AsynchCallbackExecutor();
88 public static enum LockType {
92 /** The caller thread will be blocked util the lock is granted. */
93 public synchronized void lockBlocking(LockType locktype) {
94 while (!tryLock(locktype)) {
99 /** Reactive version. The Lock will be emitted when the lock is granted */
100 public synchronized Mono<Lock> lock(LockType lockType) {
101 if (tryLock(lockType)) {
102 return Mono.just(this);
104 return Mono.create(monoSink -> addToQueue(monoSink, lockType));
108 public Mono<Lock> unlock() {
109 return Mono.create(monoSink -> {
111 monoSink.success(this);
115 public void unlockBlocking() {
116 synchronized (this) {
117 if (lockCounter <= 0) {
118 lockCounter = -1; // Might as well stop, to make it easier to find the problem
119 throw new RuntimeException("Number of unlocks must match the number of locks");
122 if (lockCounter == 0) {
127 this.processQueuedEntries();
131 public String toString() {
132 return "Lock cnt: " + this.lockCounter + " exclusive: " + this.isExclusive;
135 /** returns the current number of granted locks */
136 public synchronized int getLockCounter() {
137 return this.lockCounter;
140 private void processQueuedEntries() {
141 List<LockRequest> granted = new ArrayList<>();
142 synchronized (this) {
143 for (Iterator<LockRequest> i = lockRequestQueue.iterator(); i.hasNext();) {
144 LockRequest request = i.next();
145 if (tryLock(request.lockType)) {
147 granted.add(request);
153 * for (LockRequest request : granted) { request.callback.success(this); }
155 callbackProcessor.addAll(granted);
158 private static class LockRequest {
159 final MonoSink<Lock> callback;
160 final LockType lockType;
163 LockRequest(MonoSink<Lock> callback, LockType lockType, Lock lock) {
164 this.callback = callback;
165 this.lockType = lockType;
170 private synchronized void addToQueue(MonoSink<Lock> callback, LockType lockType) {
171 lockRequestQueue.add(new LockRequest(callback, lockType, this));
174 private void waitForUnlock() {
177 } catch (InterruptedException e) {
178 logger.warn("waitForUnlock interrupted", e);
182 private boolean tryLock(LockType lockType) {
183 if (this.isExclusive) {
186 if (lockType == LockType.EXCLUSIVE && lockCounter > 0) {
190 this.isExclusive = lockType == LockType.EXCLUSIVE;