Skip to content

Commit

Permalink
refactoring
Browse files Browse the repository at this point in the history
  • Loading branch information
Nikita Koksharov committed May 17, 2024
1 parent 86c8892 commit 6bfd7c9
Show file tree
Hide file tree
Showing 36 changed files with 157 additions and 198 deletions.
8 changes: 4 additions & 4 deletions ...ring-data-20/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
Expand Down Expand Up @@ -51,7 +51,7 @@ public void onMessage(Message message, byte[] pattern) {
RedisConnection c = factory.getConnection();
c.publish("test".getBytes(), "sdfdsf".getBytes());

Awaitility.await().atMost(Duration.FIVE_SECONDS).until(() -> {
Awaitility.await().atMost(Durations.FIVE_SECONDS).until(() -> {
return counterTest.get() == 2;
});
Assertions.assertThat(counterTest2.get()).isZero();
Expand All @@ -72,7 +72,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand All @@ -92,7 +92,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...a-21/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeReactiveTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import java.util.concurrent.atomic.AtomicReference;

import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveSubscription;
Expand All @@ -30,7 +30,7 @@ public void testTemplate() {
ReactiveRedisConnection connection = factory.getReactiveConnection();
connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();

Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));
}

Expand All @@ -49,7 +49,7 @@ public void testSubscribe() {

connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();

Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

ss.unsubscribe();
Expand All @@ -71,7 +71,7 @@ public void testUnSubscribe() {
}).subscribe();

connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

ss.unsubscribe();
Expand Down
6 changes: 3 additions & 3 deletions ...ring-data-21/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import java.util.concurrent.atomic.AtomicReference;

import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
Expand All @@ -23,7 +23,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand All @@ -43,7 +43,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...-22/src/main/java/org/redisson/spring/data/connection/RedissonReactiveStreamCommands.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -214,15 +214,15 @@ public Flux<ReactiveRedisConnection.CommandResponse<ReadCommand, Flux<ByteBuffer

if (command.getConsumer() == null) {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD, params.toArray());
}
} else {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP, params.toArray());
}
}

Expand Down
6 changes: 3 additions & 3 deletions ...ring-data-22/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import java.util.concurrent.atomic.AtomicReference;

import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
Expand All @@ -23,7 +23,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand All @@ -43,7 +43,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...-23/src/main/java/org/redisson/spring/data/connection/RedissonReactiveStreamCommands.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -412,15 +412,15 @@ public Flux<ReactiveRedisConnection.CommandResponse<ReadCommand, Flux<ByteBuffer

if (command.getConsumer() == null) {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD, params.toArray());
}
} else {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP, params.toArray());
}
}

Expand Down
8 changes: 4 additions & 4 deletions ...a-23/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeReactiveTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import java.util.concurrent.atomic.AtomicReference;

import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveSubscription;
Expand Down Expand Up @@ -36,7 +36,7 @@ public void testTemplate() {
ReactiveRedisConnection connection = factory.getReactiveConnection();
connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();

Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> counter.get() == 1);
}

Expand All @@ -55,7 +55,7 @@ public void testSubscribe() {

connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();

Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

ss.unsubscribe();
Expand All @@ -77,7 +77,7 @@ public void testUnSubscribe() {
}).subscribe();

connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

ss.unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...ring-data-23/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.redisson.RedisRunner;
import org.redisson.Redisson;
Expand Down Expand Up @@ -65,7 +65,7 @@ public void onMessage(Message message, byte[] pattern) {
c.set("mykey".getBytes(), "2".getBytes());
c.del("mykey".getBytes());

Awaitility.await().atMost(Duration.ONE_SECOND).until(() -> {
Awaitility.await().atMost(Durations.ONE_SECOND).until(() -> {
return counterTest.get() == 3;
});

Expand All @@ -85,7 +85,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand All @@ -105,7 +105,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...-24/src/main/java/org/redisson/spring/data/connection/RedissonReactiveStreamCommands.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -412,15 +412,15 @@ public Flux<ReactiveRedisConnection.CommandResponse<ReadCommand, Flux<ByteBuffer

if (command.getConsumer() == null) {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD, params.toArray());
}
} else {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP, params.toArray());
}
}

Expand Down
8 changes: 4 additions & 4 deletions ...a-24/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeReactiveTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import java.util.concurrent.atomic.AtomicReference;

import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveSubscription;
Expand Down Expand Up @@ -36,7 +36,7 @@ public void testTemplate() {
ReactiveRedisConnection connection = factory.getReactiveConnection();
connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();

Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> counter.get() == 1);
}

Expand All @@ -55,7 +55,7 @@ public void testSubscribe() {

connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();

Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

ss.unsubscribe();
Expand All @@ -77,7 +77,7 @@ public void testUnSubscribe() {
}).subscribe();

connection.pubSubCommands().publish(ByteBuffer.wrap("test".getBytes()), ByteBuffer.wrap("msg".getBytes())).block();
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

ss.unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...ring-data-24/src/test/java/org/redisson/spring/data/connection/RedissonSubscribeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
import org.awaitility.Duration;
import org.awaitility.Durations;
import org.junit.Test;
import org.redisson.RedisRunner;
import org.redisson.Redisson;
Expand Down Expand Up @@ -65,7 +65,7 @@ public void onMessage(Message message, byte[] pattern) {
c.set("mykey".getBytes(), "2".getBytes());
c.del("mykey".getBytes());

Awaitility.await().atMost(Duration.FIVE_SECONDS).until(() -> {
Awaitility.await().atMost(Durations.FIVE_SECONDS).until(() -> {
return counterTest.get() == 3;
});

Expand All @@ -85,7 +85,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand All @@ -105,7 +105,7 @@ public void onMessage(Message message, byte[] pattern) {
}, "test".getBytes());

connection.publish("test".getBytes(), "msg".getBytes());
Awaitility.await().atMost(Duration.ONE_SECOND)
Awaitility.await().atMost(Durations.ONE_SECOND)
.until(() -> Arrays.equals("msg".getBytes(), msg.get()));

connection.getSubscription().unsubscribe();
Expand Down
8 changes: 4 additions & 4 deletions ...-25/src/main/java/org/redisson/spring/data/connection/RedissonReactiveStreamCommands.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -412,15 +412,15 @@ public Flux<ReactiveRedisConnection.CommandResponse<ReadCommand, Flux<ByteBuffer

if (command.getConsumer() == null) {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREAD, params.toArray());
}
} else {
if (command.getReadOptions().getBlock() != null && command.getReadOptions().getBlock() > 0) {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupBlockingCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP_BLOCKING, params.toArray());
} else {
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, executorService.getServiceManager().getXReadGroupCommand(), params.toArray());
m = read(toByteArray(command.getStreamOffsets().get(0).getKey()), ByteArrayCodec.INSTANCE, RedisCommands.XREADGROUP, params.toArray());
}
}

Expand Down

0 comments on commit 6bfd7c9

Please sign in to comment.