Reactor Locks - Locks, RWLocks, Semaphores

Reactor Locks is a library providing reactive access to locks with Project Reactor.

import party.iroiro.lock.Lock;
import party.iroiro.lock.ReactiveLock;

class Example {
    Lock lock = new ReactiveLock();
    
    <T> Mono<T> reactive(Mono<T> mono) {
        return mono
                .as(Lock::begin)
                .flatMap(t -> someJob(t))
                .map(t -> someOtherJob(t))
                .with(lock);
    }
}

Getting It

Maven

<dependency>
  <groupId>party.iroiro</groupId>
  <artifactId>reactor-locks</artifactId>
  <version>1.1.0</version>
</dependency>

Gradle

implementation 'party.iroiro:reactor-locks:1.1.0'

Usage

Fluent API

import party.iroiro.lock.Lock;

// ...

Lock lock = getLock();
// (1)    Start a locked scope builder
   mono.as(Lock::begin)

//    (2) Register operations in the locked scope
//    (2) Operations inside will be performed with the lock locked
      .map(t -> t + 1)
      .flatMap(this::fm)

// (3)   Build the locked scope with a lock
    .with(lock);

For RWLocks, you may choose between .with and .withR to distinguish reader-locking and writer-locking.

Scoped Function

A little lengthy.

mono.flatMap(t -> lock.withLock(() -> {
    return Mono.just(t)
               .map(t -> t + 1)
               .flatMap(this::fm);
});

For RWLockss, .withLock or .withRLock.

Advanced

The APIs below utilize LockHandle (Javadoc):

public interface LockHandle {
    Mono<Void> mono();
    boolean cancel();
}

Subscribe to mono() to get informed when the lock is ready. Or you may cancel the request. In case of a failed cancellation (false returned), you need to unlock the lock yourself.

Function	Returns
Lock::tryLock() RWLock::tryRLock() Immediately requests the lock. Use it with Mono.defer if you want laziness.	A LockHandle which you may use to cancel the lock request any time you want to. Just make sure to check the return value of cancel in case that the lock is already granted (which means you need to unlock it yourself.
Lock::unlock() RWLock::rUnlock()	Returns void since you do not need to wait to unlock.
Lock::isLocked() RWLock::isRLocked()	Whether the lock has been (either reader- or writer-) locked. For semaphores, it means whether the lock has reached the max lock holders. Never rely on the result of this.

Wrapped Operators

In previous versions of this library, we have lockOnNext unlockOnNext to make locking easier. However, all these operators does not handle Mono cancellations (from downstream timeout for example) and we are deprecating them.

Use the fluent API or withLock / withRLock instead.

Locks

All the lock implementations use CAS operations and are non-blocking.

Lock Implementations

• ReactiveLock

  A basic lock. Internally using a queue.

• BroadcastingLock

  Functionally equivalent to ReactiveLock. Internally using a broadcast, which might degrade performance in really extreme cases.

• ReactiveSemaphore

  A lock allowing multiple lock holders.

RWLock Implementations

• ReactiveRWLock

  A basic readers-writer lock. Handling writer hunger by blocking (reactively, of course) further readers if there is a writer waiting.

License

This project is licensed under the Apache License Version 2.0.

Copyright 2022 the original author or authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

     https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

The javadoc.gradle file is actually modified from that of reactor-core, which is licensed under the Apache License Version 2.0 as well.