tryAcquire();
+
+ /**
+ * Acquires a permit from this semaphore, if one becomes available
+ * within the given waiting time and the current thread has not
+ * been {@linkplain Thread#interrupt interrupted}.
+ *
+ * Acquires a permit, if one is available and returns immediately,
+ * with the permit id,
+ * reducing the number of available permits by one.
+ *
+ *
If no permit is available then the current thread becomes
+ * disabled for thread scheduling purposes and lies dormant until
+ * one of three things happens:
+ *
+ * - Some other thread invokes the {@link #release(String)} method for this
+ * semaphore and the current thread is next to be assigned a permit; or
+ *
- Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
+ *
- The specified waiting time elapses.
+ *
+ *
+ * If a permit is acquired then the permit id is returned.
+ *
+ *
If the specified waiting time elapses then the value {@code null}
+ * is returned. If the time is less than or equal to zero, the method
+ * will not wait at all.
+ *
+ * @param waitTime the maximum time to wait for a permit
+ * @param unit the time unit of the {@code timeout} argument
+ * @return permit id if a permit was acquired and {@code null}
+ * if the waiting time elapsed before a permit was acquired
+ */
+ Publisher tryAcquire(long waitTime, TimeUnit unit);
+
+ /**
+ * Acquires a permit with defined lease time from this semaphore,
+ * if one becomes available
+ * within the given waiting time and the current thread has not
+ * been {@linkplain Thread#interrupt interrupted}.
+ *
+ * Acquires a permit, if one is available and returns immediately,
+ * with the permit id,
+ * reducing the number of available permits by one.
+ *
+ *
If no permit is available then the current thread becomes
+ * disabled for thread scheduling purposes and lies dormant until
+ * one of three things happens:
+ *
+ * - Some other thread invokes the {@link #release(String)} method for this
+ * semaphore and the current thread is next to be assigned a permit; or
+ *
- Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
+ *
- The specified waiting time elapses.
+ *
+ *
+ * If a permit is acquired then the permit id is returned.
+ *
+ *
If the specified waiting time elapses then the value {@code null}
+ * is returned. If the time is less than or equal to zero, the method
+ * will not wait at all.
+ *
+ * @param waitTime the maximum time to wait for a permit
+ * @param leaseTime permit lease time
+ * @param unit the time unit of the {@code timeout} argument
+ * @return permit id if a permit was acquired and {@code null}
+ * if the waiting time elapsed before a permit was acquired
+ */
+ Publisher tryAcquire(long waitTime, long leaseTime, TimeUnit unit);
+
+ /**
+ * Releases a permit by its id, returning it to the semaphore.
+ *
+ * Releases a permit, increasing the number of available permits by
+ * one. If any threads of Redisson client are trying to acquire a permit,
+ * then one is selected and given the permit that was just released.
+ *
+ *
There is no requirement that a thread that releases a permit must
+ * have acquired that permit by calling {@link #acquire()}.
+ * Correct usage of a semaphore is established by programming convention
+ * in the application.
+ *
+ * @param permitId - permit id
+ * @return {@code true} if a permit has been released and {@code false}
+ * otherwise
+ */
+ Publisher tryRelease(String permitId);
+
+ /**
+ * Releases a permit by its id, returning it to the semaphore.
+ *
+ * Releases a permit, increasing the number of available permits by
+ * one. If any threads of Redisson client are trying to acquire a permit,
+ * then one is selected and given the permit that was just released.
+ *
+ *
There is no requirement that a thread that releases a permit must
+ * have acquired that permit by calling {@link #acquire()}.
+ * Correct usage of a semaphore is established by programming convention
+ * in the application.
+ *
+ *
Throws an exception if permit id doesn't exist or has already been release
+ *
+ * @param permitId - permit id
+ * @return void
+ */
+ Publisher release(String permitId);
+
+ /**
+ * Returns the current number of available permits.
+ *
+ * @return number of available permits
+ */
+ Publisher availablePermits();
+
+ /**
+ * Sets number of permits.
+ *
+ * @param permits - number of permits
+ * @return true if permits has been set successfully, otherwise false.
+ */
+ Publisher trySetPermits(int permits);
+
+ /**
+ * Increases or decreases the number of available permits by defined value.
+ *
+ * @param permits - number of permits to add/remove
+ * @return void
+ */
+ Publisher addPermits(int permits);
+
+}
diff --git a/redisson/src/main/java/org/redisson/api/RedissonReactiveClient.java b/redisson/src/main/java/org/redisson/api/RedissonReactiveClient.java
index c576c605a..2ac9b3fe4 100644
--- a/redisson/src/main/java/org/redisson/api/RedissonReactiveClient.java
+++ b/redisson/src/main/java/org/redisson/api/RedissonReactiveClient.java
@@ -38,6 +38,15 @@ public interface RedissonReactiveClient {
*/
RSemaphoreReactive getSemaphore(String name);
+ /**
+ * Returns semaphore instance by name.
+ * Supports lease time parameter for each acquired permit.
+ *
+ * @param name - name of object
+ * @return PermitExpirableSemaphore object
+ */
+ RPermitExpirableSemaphoreReactive getPermitExpirableSemaphore(String name);
+
/**
* Returns readWriteLock instance by name.
*
diff --git a/redisson/src/main/java/org/redisson/reactive/RedissonPermitExpirableSemaphoreReactive.java b/redisson/src/main/java/org/redisson/reactive/RedissonPermitExpirableSemaphoreReactive.java
new file mode 100644
index 000000000..f6047026e
--- /dev/null
+++ b/redisson/src/main/java/org/redisson/reactive/RedissonPermitExpirableSemaphoreReactive.java
@@ -0,0 +1,151 @@
+/**
+ * Copyright 2018 Nikita Koksharov
+ *
+ * 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
+ *
+ * http://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.
+ */
+package org.redisson.reactive;
+
+import java.util.concurrent.TimeUnit;
+
+import org.reactivestreams.Publisher;
+import org.redisson.RedissonLock;
+import org.redisson.RedissonPermitExpirableSemaphore;
+import org.redisson.api.RFuture;
+import org.redisson.api.RLockAsync;
+import org.redisson.api.RPermitExpirableSemaphoreAsync;
+import org.redisson.api.RPermitExpirableSemaphoreReactive;
+import org.redisson.command.CommandAsyncExecutor;
+import org.redisson.command.CommandReactiveExecutor;
+import org.redisson.pubsub.SemaphorePubSub;
+
+import reactor.fn.Supplier;
+
+/**
+ *
+ * @author Nikita Koksharov
+ *
+ */
+public class RedissonPermitExpirableSemaphoreReactive extends RedissonExpirableReactive implements RPermitExpirableSemaphoreReactive {
+
+ private final RPermitExpirableSemaphoreAsync instance;
+
+ public RedissonPermitExpirableSemaphoreReactive(CommandReactiveExecutor connectionManager, String name, SemaphorePubSub semaphorePubSub) {
+ super(connectionManager, name);
+ instance = new RedissonPermitExpirableSemaphore(commandExecutor, name, semaphorePubSub);
+ }
+
+ protected RLockAsync createLock(CommandAsyncExecutor connectionManager, String name) {
+ return new RedissonLock(commandExecutor, name);
+ }
+
+ @Override
+ public Publisher acquire() {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.acquireAsync();
+ }
+ });
+ }
+
+ @Override
+ public Publisher acquire(final long leaseTime, final TimeUnit unit) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.acquireAsync(leaseTime, unit);
+ }
+ });
+ }
+
+ @Override
+ public Publisher tryAcquire() {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.tryAcquireAsync();
+ }
+ });
+ }
+
+ @Override
+ public Publisher tryAcquire(final long waitTime, final TimeUnit unit) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.tryAcquireAsync(waitTime, unit);
+ }
+ });
+ }
+
+ @Override
+ public Publisher tryAcquire(final long waitTime, final long leaseTime, final TimeUnit unit) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.tryAcquireAsync(waitTime, leaseTime, unit);
+ }
+ });
+ }
+
+ @Override
+ public Publisher tryRelease(final String permitId) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.tryReleaseAsync(permitId);
+ }
+ });
+ }
+
+ @Override
+ public Publisher release(final String permitId) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.releaseAsync(permitId);
+ }
+ });
+ }
+
+ @Override
+ public Publisher availablePermits() {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.availablePermitsAsync();
+ }
+ });
+ }
+
+ @Override
+ public Publisher trySetPermits(final int permits) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.trySetPermitsAsync(permits);
+ }
+ });
+ }
+
+ @Override
+ public Publisher addPermits(final int permits) {
+ return reactive(new Supplier>() {
+ @Override
+ public RFuture get() {
+ return instance.addPermitsAsync(permits);
+ }
+ });
+ }
+
+}