Fixed - TransactionalBucket#set(V, Duration), PSETEX is called before MULTI #5561

Signed-off-by: seakider <seakider@gmail.com>

redisson/src/main/java/org/redisson/transaction/RedissonTransactionalBucket.java

@@ -25,6 +25,7 @@ import org.redisson.misc.CompletableFutureWrapper;
 import org.redisson.transaction.operation.*;
 import org.redisson.transaction.operation.bucket.*;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -336,7 +337,13 @@ public class RedissonTransactionalBucket<V> extends RedissonBucket<V> {
         long currentThreadId = Thread.currentThread().getId();
         return setAsync(value, new BucketSetOperation<V>(getName(), getLockName(), getCodec(), value, timeToLive, timeUnit, transactionId, currentThreadId));
     }
-    
+
+    @Override
+    public RFuture<Void> setAsync(V value, Duration duration) {
+        long currentThreadId = Thread.currentThread().getId();
+        return setAsync(value, new BucketSetOperation<V>(getName(), getLockName(), getCodec(), value, duration.toMillis(), TimeUnit.MILLISECONDS, transactionId, currentThreadId));
+    }
+
     @Override
     public RFuture<Boolean> trySetAsync(V newValue) {
         long currentThreadId = Thread.currentThread().getId();

redisson/src/test/java/org/redisson/transaction/RedissonTransactionalBucketTest.java

@@ -69,15 +69,33 @@ public class RedissonTransactionalBucketTest extends RedisDockerTest {
         
         RTransaction transaction = redisson.createTransaction(TransactionOptions.defaults());
         RBucket<String> bucket = transaction.getBucket("test");
+        assertThat(bucket.get()).isEqualTo("123");
         bucket.set("234");
         assertThat(bucket.get()).isEqualTo("234");
-        
+
         transaction.commit();
         
         assertThat(redisson.getKeys().count()).isEqualTo(1);
         assertThat(b.get()).isEqualTo("234");
     }
 
+    @Test
+    public void testSetDuration() throws InterruptedException {
+        RBucket<String> b = redisson.getBucket("test");
+        b.set("123");
+
+        RTransaction transaction = redisson.createTransaction(TransactionOptions.defaults());
+        RBucket<String> bucket = transaction.getBucket("test");
+        bucket.set("234", Duration.ofSeconds(2));
+        assertThat(bucket.get()).isEqualTo("234");
+        assertThat(b.get()).isEqualTo("123");
+        transaction.commit();
+
+        assertThat(b.get()).isEqualTo("234");
+        Thread.sleep(2200);
+        assertThat(b.get()).isNull();
+    }
+
     @Test
     public void testExpire() throws InterruptedException {
         RBucket<String> b = redisson.getBucket("test");