Merge branch 'mrniko/master' into feature/travis-ci
Rui Gu
9 years ago
commit
4b87837111
@ -0,0 +1,129 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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;
|
||||||
|
|
||||||
|
import java.util.Arrays;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
import org.redisson.client.codec.Codec;
|
||||||
|
import org.redisson.client.codec.LongCodec;
|
||||||
|
import org.redisson.client.protocol.RedisCommand;
|
||||||
|
import org.redisson.client.protocol.RedisCommands;
|
||||||
|
import org.redisson.client.protocol.RedisCommand.ValueType;
|
||||||
|
import org.redisson.client.protocol.convertor.BooleanReplayConvertor;
|
||||||
|
import org.redisson.command.CommandAsyncExecutor;
|
||||||
|
|
||||||
|
import io.netty.util.concurrent.Future;
|
||||||
|
|
||||||
|
public class RedissonMultimapCache<K> {
|
||||||
|
|
||||||
|
private static final RedisCommand<Boolean> EVAL_EXPIRE_KEY = new RedisCommand<Boolean>("EVAL", new BooleanReplayConvertor(), 6, ValueType.MAP_KEY);
|
||||||
|
|
||||||
|
private final CommandAsyncExecutor commandExecutor;
|
||||||
|
private final String name;
|
||||||
|
private final Codec codec;
|
||||||
|
private final String timeoutSetName;
|
||||||
|
|
||||||
|
public RedissonMultimapCache(CommandAsyncExecutor commandExecutor, String name, Codec codec, String timeoutSetName) {
|
||||||
|
this.commandExecutor = commandExecutor;
|
||||||
|
this.name = name;
|
||||||
|
this.codec = codec;
|
||||||
|
this.timeoutSetName = timeoutSetName;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Future<Boolean> expireKeyAsync(K key, long timeToLive, TimeUnit timeUnit) {
|
||||||
|
long ttlTimeout = System.currentTimeMillis() + timeUnit.toMillis(timeToLive);
|
||||||
|
|
||||||
|
return commandExecutor.evalWriteAsync(name, codec, EVAL_EXPIRE_KEY,
|
||||||
|
"if redis.call('hexists', KEYS[1], ARGV[2]) == 1 then "
|
||||||
|
+ "if tonumber(ARGV[1]) > 0 then "
|
||||||
|
+ "redis.call('zadd', KEYS[2], ARGV[1], ARGV[2]); " +
|
||||||
|
"else " +
|
||||||
|
"redis.call('zrem', KEYS[2], ARGV[2]); "
|
||||||
|
+ "end; "
|
||||||
|
+ "return 1; "
|
||||||
|
+ "else "
|
||||||
|
+ "return 0; "
|
||||||
|
+ "end",
|
||||||
|
Arrays.<Object>asList(name, timeoutSetName), ttlTimeout, key);
|
||||||
|
}
|
||||||
|
|
||||||
|
public Future<Boolean> deleteAsync() {
|
||||||
|
return commandExecutor.evalWriteAsync(name, LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN_AMOUNT,
|
||||||
|
"local entries = redis.call('hgetall', KEYS[1]); " +
|
||||||
|
"local keys = {KEYS[1], KEYS[2]}; " +
|
||||||
|
"for i, v in ipairs(entries) do " +
|
||||||
|
"if i % 2 == 0 then " +
|
||||||
|
"local name = '{' .. KEYS[1] .. '}:' .. v; " +
|
||||||
|
"table.insert(keys, name); " +
|
||||||
|
"end;" +
|
||||||
|
"end; " +
|
||||||
|
|
||||||
|
"local n = 0 "
|
||||||
|
+ "for i=1, #keys,5000 do "
|
||||||
|
+ "n = n + redis.call('del', unpack(keys, i, math.min(i+4999, table.getn(keys)))) "
|
||||||
|
+ "end; "
|
||||||
|
+ "return n;",
|
||||||
|
Arrays.<Object>asList(name, timeoutSetName));
|
||||||
|
}
|
||||||
|
|
||||||
|
public Future<Boolean> expireAsync(long timeToLive, TimeUnit timeUnit) {
|
||||||
|
return commandExecutor.evalWriteAsync(name, LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
|
||||||
|
"redis.call('zadd', KEYS[2], 92233720368547758, 'redisson__expiretag'); " +
|
||||||
|
"local entries = redis.call('hgetall', KEYS[1]); " +
|
||||||
|
"for i, v in ipairs(entries) do " +
|
||||||
|
"if i % 2 == 0 then " +
|
||||||
|
"local name = '{' .. KEYS[1] .. '}:' .. v; " +
|
||||||
|
"redis.call('pexpire', name, ARGV[1]); " +
|
||||||
|
"end;" +
|
||||||
|
"end; " +
|
||||||
|
"redis.call('pexpire', KEYS[2], ARGV[1]); " +
|
||||||
|
"return redis.call('pexpire', KEYS[1], ARGV[1]); ",
|
||||||
|
Arrays.<Object>asList(name, timeoutSetName), timeUnit.toMillis(timeToLive));
|
||||||
|
}
|
||||||
|
|
||||||
|
public Future<Boolean> expireAtAsync(long timestamp) {
|
||||||
|
return commandExecutor.evalWriteAsync(name, LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
|
||||||
|
"redis.call('zadd', KEYS[2], 92233720368547758, 'redisson__expiretag');" +
|
||||||
|
"local entries = redis.call('hgetall', KEYS[1]); " +
|
||||||
|
"for i, v in ipairs(entries) do " +
|
||||||
|
"if i % 2 == 0 then " +
|
||||||
|
"local name = '{' .. KEYS[1] .. '}:' .. v; " +
|
||||||
|
"redis.call('pexpireat', name, ARGV[1]); " +
|
||||||
|
"end;" +
|
||||||
|
"end; " +
|
||||||
|
"redis.call('pexpireat', KEYS[2], ARGV[1]); " +
|
||||||
|
"return redis.call('pexpireat', KEYS[1], ARGV[1]); ",
|
||||||
|
Arrays.<Object>asList(name, timeoutSetName), timestamp);
|
||||||
|
}
|
||||||
|
|
||||||
|
public Future<Boolean> clearExpireAsync() {
|
||||||
|
return commandExecutor.evalWriteAsync(name, LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
|
||||||
|
"redis.call('zrem', KEYS[2], 'redisson__expiretag'); " +
|
||||||
|
"local entries = redis.call('hgetall', KEYS[1]); " +
|
||||||
|
"for i, v in ipairs(entries) do " +
|
||||||
|
"if i % 2 == 0 then " +
|
||||||
|
"local name = '{' .. KEYS[1] .. '}:' .. v; " +
|
||||||
|
"redis.call('persist', name); " +
|
||||||
|
"end;" +
|
||||||
|
"end; " +
|
||||||
|
"redis.call('persist', KEYS[2]); " +
|
||||||
|
"return redis.call('persist', KEYS[1]); ",
|
||||||
|
Arrays.<Object>asList(name, timeoutSetName));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,165 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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;
|
||||||
|
|
||||||
|
import java.lang.reflect.InvocationHandler;
|
||||||
|
import java.lang.reflect.Method;
|
||||||
|
import java.lang.reflect.Proxy;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.CountDownLatch;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
||||||
|
import org.redisson.client.RedisException;
|
||||||
|
import org.redisson.client.RedisTimeoutException;
|
||||||
|
import org.redisson.core.MessageListener;
|
||||||
|
import org.redisson.core.RBlockingQueue;
|
||||||
|
import org.redisson.core.RRemoteService;
|
||||||
|
import org.redisson.core.RTopic;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
|
import io.netty.buffer.ByteBufUtil;
|
||||||
|
import io.netty.util.concurrent.Future;
|
||||||
|
import io.netty.util.concurrent.FutureListener;
|
||||||
|
import io.netty.util.internal.PlatformDependent;
|
||||||
|
import io.netty.util.internal.ThreadLocalRandom;
|
||||||
|
|
||||||
|
public class RedissonRemoteService implements RRemoteService {
|
||||||
|
|
||||||
|
private static final Logger log = LoggerFactory.getLogger(RedissonRemoteService.class);
|
||||||
|
|
||||||
|
private final Map<RemoteServiceKey, RemoteServiceMethod> beans = PlatformDependent.newConcurrentHashMap();
|
||||||
|
|
||||||
|
private final Redisson redisson;
|
||||||
|
|
||||||
|
public RedissonRemoteService(Redisson redisson) {
|
||||||
|
this.redisson = redisson;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public <T> void register(Class<T> remoteInterface, T object) {
|
||||||
|
register(remoteInterface, object, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public <T> void register(Class<T> remoteInterface, T object, int executorsAmount) {
|
||||||
|
if (executorsAmount < 1) {
|
||||||
|
throw new IllegalArgumentException("executorsAmount can't be lower than 1");
|
||||||
|
}
|
||||||
|
for (Method method : remoteInterface.getMethods()) {
|
||||||
|
RemoteServiceMethod value = new RemoteServiceMethod(method, object);
|
||||||
|
RemoteServiceKey key = new RemoteServiceKey(remoteInterface, method.getName());
|
||||||
|
if (beans.put(key, value) != null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for (int i = 0; i < executorsAmount; i++) {
|
||||||
|
String requestQueueName = "redisson_remote_service:{" + remoteInterface.getName() + "}";
|
||||||
|
RBlockingQueue<RemoteServiceRequest> requestQueue = redisson.getBlockingQueue(requestQueueName);
|
||||||
|
subscribe(remoteInterface, requestQueue);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private <T> void subscribe(final Class<T> remoteInterface, final RBlockingQueue<RemoteServiceRequest> requestQueue) {
|
||||||
|
Future<RemoteServiceRequest> take = requestQueue.takeAsync();
|
||||||
|
take.addListener(new FutureListener<RemoteServiceRequest>() {
|
||||||
|
@Override
|
||||||
|
public void operationComplete(Future<RemoteServiceRequest> future) throws Exception {
|
||||||
|
if (!future.isSuccess()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
RemoteServiceRequest request = future.getNow();
|
||||||
|
RemoteServiceMethod method = beans.get(new RemoteServiceKey(remoteInterface, request.getMethodName()));
|
||||||
|
String responseName = "redisson_remote_service:{" + remoteInterface.getName() + "}:" + request.getRequestId();
|
||||||
|
RTopic<RemoteServiceResponse> topic = redisson.getTopic(responseName);
|
||||||
|
RemoteServiceResponse response;
|
||||||
|
try {
|
||||||
|
Object result = method.getMethod().invoke(method.getBean(), request.getArgs());
|
||||||
|
response = new RemoteServiceResponse(result);
|
||||||
|
} catch (Exception e) {
|
||||||
|
response = new RemoteServiceResponse(e.getCause());
|
||||||
|
log.error("Can't execute: " + request, e);
|
||||||
|
}
|
||||||
|
|
||||||
|
long clients = topic.publish(response);
|
||||||
|
if (clients == 0) {
|
||||||
|
log.error("None of clients has not received a response: {} for request: {}", response, request);
|
||||||
|
}
|
||||||
|
|
||||||
|
subscribe(remoteInterface, requestQueue);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public <T> T get(Class<T> remoteInterface) {
|
||||||
|
return get(remoteInterface, -1, null);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public <T> T get(final Class<T> remoteInterface, final int timeout, final TimeUnit timeUnit) {
|
||||||
|
InvocationHandler handler = new InvocationHandler() {
|
||||||
|
@Override
|
||||||
|
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
|
||||||
|
String requestId = generateRequestId();
|
||||||
|
|
||||||
|
String requestQueueName = "redisson_remote_service:{" + remoteInterface.getName() + "}";
|
||||||
|
RBlockingQueue<RemoteServiceRequest> requestQueue = redisson.getBlockingQueue(requestQueueName);
|
||||||
|
RemoteServiceRequest request = new RemoteServiceRequest(requestId, method.getName(), args);
|
||||||
|
requestQueue.add(request);
|
||||||
|
|
||||||
|
String responseName = "redisson_remote_service:{" + remoteInterface.getName() + "}:" + requestId;
|
||||||
|
final RTopic<RemoteServiceResponse> topic = redisson.getTopic(responseName);
|
||||||
|
final CountDownLatch latch = new CountDownLatch(1);
|
||||||
|
final AtomicReference<RemoteServiceResponse> response = new AtomicReference<RemoteServiceResponse>();
|
||||||
|
int listenerId = topic.addListener(new MessageListener<RemoteServiceResponse>() {
|
||||||
|
@Override
|
||||||
|
public void onMessage(String channel, RemoteServiceResponse msg) {
|
||||||
|
response.set(msg);
|
||||||
|
latch.countDown();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
if (timeout == -1) {
|
||||||
|
latch.await();
|
||||||
|
} else {
|
||||||
|
if (!latch.await(timeout, timeUnit)) {
|
||||||
|
topic.removeListener(listenerId);
|
||||||
|
throw new RedisTimeoutException("No response after " + timeUnit.toMillis(timeout) + "ms for request: " + request);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
topic.removeListener(listenerId);
|
||||||
|
RemoteServiceResponse msg = response.get();
|
||||||
|
if (msg.getError() != null) {
|
||||||
|
throw msg.getError();
|
||||||
|
}
|
||||||
|
return msg.getResult();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
return (T) Proxy.newProxyInstance(remoteInterface.getClassLoader(), new Class[] {remoteInterface}, handler);
|
||||||
|
}
|
||||||
|
|
||||||
|
private String generateRequestId() {
|
||||||
|
byte[] id = new byte[16];
|
||||||
|
// TODO JDK UPGRADE replace to native ThreadLocalRandom
|
||||||
|
ThreadLocalRandom.current().nextBytes(id);
|
||||||
|
return ByteBufUtil.hexDump(id);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,68 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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;
|
||||||
|
|
||||||
|
public class RemoteServiceKey {
|
||||||
|
|
||||||
|
private final Class<?> serviceInterface;
|
||||||
|
private final String methodName;
|
||||||
|
|
||||||
|
public RemoteServiceKey(Class<?> serviceInterface, String methodName) {
|
||||||
|
super();
|
||||||
|
this.serviceInterface = serviceInterface;
|
||||||
|
this.methodName = methodName;
|
||||||
|
}
|
||||||
|
|
||||||
|
public String getMethodName() {
|
||||||
|
return methodName;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Class<?> getServiceInterface() {
|
||||||
|
return serviceInterface;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public int hashCode() {
|
||||||
|
final int prime = 31;
|
||||||
|
int result = 1;
|
||||||
|
result = prime * result + ((methodName == null) ? 0 : methodName.hashCode());
|
||||||
|
result = prime * result + ((serviceInterface == null) ? 0 : serviceInterface.getName().hashCode());
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public boolean equals(Object obj) {
|
||||||
|
if (this == obj)
|
||||||
|
return true;
|
||||||
|
if (obj == null)
|
||||||
|
return false;
|
||||||
|
if (getClass() != obj.getClass())
|
||||||
|
return false;
|
||||||
|
RemoteServiceKey other = (RemoteServiceKey) obj;
|
||||||
|
if (methodName == null) {
|
||||||
|
if (other.methodName != null)
|
||||||
|
return false;
|
||||||
|
} else if (!methodName.equals(other.methodName))
|
||||||
|
return false;
|
||||||
|
if (serviceInterface == null) {
|
||||||
|
if (other.serviceInterface != null)
|
||||||
|
return false;
|
||||||
|
} else if (!serviceInterface.equals(other.serviceInterface))
|
||||||
|
return false;
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,39 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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;
|
||||||
|
|
||||||
|
import java.lang.reflect.Method;
|
||||||
|
|
||||||
|
public class RemoteServiceMethod {
|
||||||
|
|
||||||
|
private final Object bean;
|
||||||
|
private final Method method;
|
||||||
|
|
||||||
|
public RemoteServiceMethod(Method method, Object bean) {
|
||||||
|
super();
|
||||||
|
this.method = method;
|
||||||
|
this.bean = bean;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Object getBean() {
|
||||||
|
return bean;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Method getMethod() {
|
||||||
|
return method;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,54 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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;
|
||||||
|
|
||||||
|
import java.util.Arrays;
|
||||||
|
|
||||||
|
public class RemoteServiceRequest {
|
||||||
|
|
||||||
|
private String requestId;
|
||||||
|
private String methodName;
|
||||||
|
private Object[] args;
|
||||||
|
|
||||||
|
public RemoteServiceRequest() {
|
||||||
|
}
|
||||||
|
|
||||||
|
public RemoteServiceRequest(String requestId, String methodName, Object[] args) {
|
||||||
|
super();
|
||||||
|
this.requestId = requestId;
|
||||||
|
this.methodName = methodName;
|
||||||
|
this.args = args;
|
||||||
|
}
|
||||||
|
|
||||||
|
public String getRequestId() {
|
||||||
|
return requestId;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Object[] getArgs() {
|
||||||
|
return args;
|
||||||
|
}
|
||||||
|
|
||||||
|
public String getMethodName() {
|
||||||
|
return methodName;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String toString() {
|
||||||
|
return "RemoteServiceRequest[requestId=" + requestId + ", methodName=" + methodName + ", args="
|
||||||
|
+ Arrays.toString(args) + "]";
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,47 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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;
|
||||||
|
|
||||||
|
public class RemoteServiceResponse {
|
||||||
|
|
||||||
|
private Object result;
|
||||||
|
private Throwable error;
|
||||||
|
|
||||||
|
public RemoteServiceResponse() {
|
||||||
|
}
|
||||||
|
|
||||||
|
public RemoteServiceResponse(Object result) {
|
||||||
|
this.result = result;
|
||||||
|
}
|
||||||
|
|
||||||
|
public RemoteServiceResponse(Throwable error) {
|
||||||
|
this.error = error;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Throwable getError() {
|
||||||
|
return error;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Object getResult() {
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String toString() {
|
||||||
|
return "RemoteServiceResponse [result=" + result + ", error=" + error + "]";
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,58 @@
|
|||||||
|
/**
|
||||||
|
* Copyright 2014 Nikita Koksharov, Nickolay Borbit
|
||||||
|
*
|
||||||
|
* 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.core;
|
||||||
|
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
public interface RRemoteService {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Register remote service with single executor
|
||||||
|
*
|
||||||
|
* @param remoteInterface
|
||||||
|
* @param object
|
||||||
|
*/
|
||||||
|
<T> void register(Class<T> remoteInterface, T object);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Register remote service with custom executors amount
|
||||||
|
*
|
||||||
|
* @param remoteInterface
|
||||||
|
* @param object
|
||||||
|
* @param executorsAmount
|
||||||
|
*/
|
||||||
|
<T> void register(Class<T> remoteInterface, T object, int executorsAmount);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get remote service object for remote invocations
|
||||||
|
*
|
||||||
|
* @param remoteInterface
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
|
<T> T get(Class<T> remoteInterface);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get remote service object for remote invocations
|
||||||
|
* with specified invocation timeout
|
||||||
|
*
|
||||||
|
* @param remoteInterface
|
||||||
|
* @param timeout - invocation timeout
|
||||||
|
* @param timeUnit
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
|
<T> T get(Class<T> remoteInterface, int timeout, TimeUnit timeUnit);
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,108 @@
|
|||||||
|
package org.redisson;
|
||||||
|
|
||||||
|
import org.junit.Assert;
|
||||||
|
import org.junit.Test;
|
||||||
|
import org.redisson.client.RedisTimeoutException;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.*;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
public class RedissonRemoteServiceTest extends BaseTest {
|
||||||
|
|
||||||
|
public interface RemoteInterface {
|
||||||
|
|
||||||
|
void voidMethod(String name, Long param);
|
||||||
|
|
||||||
|
Long resultMethod(Long value);
|
||||||
|
|
||||||
|
void errorMethod() throws IOException;
|
||||||
|
|
||||||
|
void errorMethodWithCause();
|
||||||
|
|
||||||
|
void timeoutMethod() throws InterruptedException;
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
public class RemoteImpl implements RemoteInterface {
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void voidMethod(String name, Long param) {
|
||||||
|
System.out.println(name + " " + param);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Long resultMethod(Long value) {
|
||||||
|
return value*2;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void errorMethod() throws IOException {
|
||||||
|
throw new IOException("Checking error throw");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void errorMethodWithCause() {
|
||||||
|
try {
|
||||||
|
int s = 2 / 0;
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new RuntimeException("Checking error throw", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void timeoutMethod() throws InterruptedException {
|
||||||
|
Thread.sleep(2000);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test(expected = RedisTimeoutException.class)
|
||||||
|
public void testTimeout() throws InterruptedException {
|
||||||
|
RedissonClient r1 = Redisson.create();
|
||||||
|
r1.getRemoteSerivce().register(RemoteInterface.class, new RemoteImpl());
|
||||||
|
|
||||||
|
RedissonClient r2 = Redisson.create();
|
||||||
|
RemoteInterface ri = r2.getRemoteSerivce().get(RemoteInterface.class, 1, TimeUnit.SECONDS);
|
||||||
|
|
||||||
|
try {
|
||||||
|
ri.timeoutMethod();
|
||||||
|
} finally {
|
||||||
|
r1.shutdown();
|
||||||
|
r2.shutdown();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testInvocations() {
|
||||||
|
RedissonClient r1 = Redisson.create();
|
||||||
|
r1.getRemoteSerivce().register(RemoteInterface.class, new RemoteImpl());
|
||||||
|
|
||||||
|
RedissonClient r2 = Redisson.create();
|
||||||
|
RemoteInterface ri = r2.getRemoteSerivce().get(RemoteInterface.class);
|
||||||
|
|
||||||
|
ri.voidMethod("someName", 100L);
|
||||||
|
assertThat(ri.resultMethod(100L)).isEqualTo(200);
|
||||||
|
|
||||||
|
try {
|
||||||
|
ri.errorMethod();
|
||||||
|
Assert.fail();
|
||||||
|
} catch (IOException e) {
|
||||||
|
assertThat(e.getMessage()).isEqualTo("Checking error throw");
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
ri.errorMethodWithCause();
|
||||||
|
Assert.fail();
|
||||||
|
} catch (Exception e) {
|
||||||
|
assertThat(e.getCause()).isInstanceOf(ArithmeticException.class);
|
||||||
|
assertThat(e.getCause().getMessage()).isEqualTo("/ by zero");
|
||||||
|
}
|
||||||
|
|
||||||
|
r1.shutdown();
|
||||||
|
r2.shutdown();
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue