Merge branch 'master' into feature_json_mget
seakider
7 months ago
commit
d418fb648e
@ -0,0 +1,61 @@
|
||||
/**
|
||||
* Copyright (c) 2013-2024 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.connection.decoder;
|
||||
|
||||
import org.redisson.client.handler.State;
|
||||
import org.redisson.client.protocol.decoder.MultiDecoder;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class MapNativeAllDecoder implements MultiDecoder<Map<Object, Object>> {
|
||||
|
||||
private final List<Object> args;
|
||||
private final Class<?> valueClass;
|
||||
|
||||
public MapNativeAllDecoder(List<Object> args, Class<?> valueClass) {
|
||||
this.args = args;
|
||||
this.valueClass = valueClass;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<Object, Object> decode(List<Object> parts, State state) {
|
||||
if (parts.isEmpty()) {
|
||||
return new HashMap<>();
|
||||
}
|
||||
Map<Object, Object> result = new LinkedHashMap<>(parts.size());
|
||||
for (int index = 0; index < parts.size(); index++) {
|
||||
Long value = (Long) parts.get(index);
|
||||
if (value == -2 && valueClass != Long.class) {
|
||||
continue;
|
||||
}
|
||||
if (valueClass == Boolean.class) {
|
||||
result.put(args.get(index), value == 1);
|
||||
} else {
|
||||
result.put(args.get(index), value);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,79 @@
|
||||
package org.redisson.pubsub;
|
||||
|
||||
import mockit.Injectable;
|
||||
import mockit.Mock;
|
||||
import mockit.MockUp;
|
||||
import mockit.Tested;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.redisson.RedissonLockEntry;
|
||||
import org.redisson.client.ChannelName;
|
||||
import org.redisson.client.RedisPubSubListener;
|
||||
import org.redisson.client.RedisTimeoutException;
|
||||
import org.redisson.client.codec.Codec;
|
||||
import org.redisson.misc.AsyncSemaphore;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
public class PublishSubscribeTest {
|
||||
|
||||
@Tested
|
||||
private LockPubSub lockPubSub;
|
||||
|
||||
@Injectable
|
||||
private PublishSubscribeService publishSubscribeService;
|
||||
|
||||
@Test
|
||||
public void testSubscribeForRaceCondition() throws InterruptedException {
|
||||
AtomicReference<CompletableFuture<PubSubConnectionEntry>> sRef = new AtomicReference<>();
|
||||
new MockUp<PublishSubscribeService>() {
|
||||
|
||||
@Mock
|
||||
AsyncSemaphore getSemaphore(ChannelName channelName) {
|
||||
return new AsyncSemaphore(1);
|
||||
}
|
||||
|
||||
@Mock
|
||||
CompletableFuture<PubSubConnectionEntry> subscribeNoTimeout(
|
||||
Codec codec, String channelName,
|
||||
AsyncSemaphore semaphore, RedisPubSubListener<?>... listeners) {
|
||||
sRef.set(new CompletableFuture<>());
|
||||
return sRef.get();
|
||||
}
|
||||
};
|
||||
|
||||
CompletableFuture<RedissonLockEntry> newPromise = lockPubSub.subscribe(
|
||||
"test", "redisson_lock__channel__test"
|
||||
);
|
||||
sRef.get().whenComplete((r, e) -> {
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException ex) {
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
});
|
||||
|
||||
Thread thread1 = new Thread(() -> sRef.get().complete(null));
|
||||
Thread thread2 = new Thread(() -> newPromise.completeExceptionally(new RedisTimeoutException("test")));
|
||||
|
||||
thread1.start();
|
||||
thread2.start();
|
||||
thread1.join();
|
||||
thread2.join();
|
||||
|
||||
assertTrue(newPromise.isCompletedExceptionally());
|
||||
assertTrue(sRef.get().isDone());
|
||||
assertFalse(sRef.get().isCompletedExceptionally());
|
||||
|
||||
CompletableFuture<RedissonLockEntry> secondPromise = lockPubSub.subscribe(
|
||||
"test", "redisson_lock__channel__test"
|
||||
);
|
||||
Thread thread3 = new Thread(() -> secondPromise.complete(null));
|
||||
thread3.start();
|
||||
thread3.join();
|
||||
assertTrue(secondPromise.isDone());
|
||||
assertFalse(secondPromise.isCompletedExceptionally());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue