Skip to content

Commit

Permalink
Ability to define custom ExecutorService for remoteService workers
Browse files Browse the repository at this point in the history
  • Loading branch information
Nikita committed Aug 8, 2016
1 parent c3c190e commit 6615293
Show file tree
Hide file tree
Showing 3 changed files with 90 additions and 28 deletions.
54 changes: 39 additions & 15 deletions src/main/java/org/redisson/RedissonRemoteService.java
View file
Expand Up @@ -23,6 +23,7 @@
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

Expand Down Expand Up @@ -98,8 +99,12 @@ public <T> void register(Class<T> remoteInterface, T object) {
}

@Override
public <T> void register(Class<T> remoteInterface, T object, int executorsAmount) {
if (executorsAmount < 1) {
public <T> void register(Class<T> remoteInterface, T object, int workersAmount) {
register(remoteInterface, object, workersAmount, null);
}

public <T> void register(Class<T> remoteInterface, T object, int workersAmount, Executor executor) {
if (workersAmount < 1) {
throw new IllegalArgumentException("executorsAmount can't be lower than 1");
}
for (Method method : remoteInterface.getMethods()) {
Expand All @@ -110,10 +115,10 @@ public <T> void register(Class<T> remoteInterface, T object, int executorsAmount
}
}

for (int i = 0; i < executorsAmount; i++) {
for (int i = 0; i < workersAmount; i++) {
String requestQueueName = getRequestQueueName(remoteInterface);
RBlockingQueue<RemoteServiceRequest> requestQueue = redisson.getBlockingQueue(requestQueueName, getCodec());
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
}
}

Expand Down Expand Up @@ -144,7 +149,7 @@ protected byte[] encode(Object obj) {
}
}

private <T> void subscribe(final Class<T> remoteInterface, final RBlockingQueue<RemoteServiceRequest> requestQueue) {
private <T> void subscribe(final Class<T> remoteInterface, final RBlockingQueue<RemoteServiceRequest> requestQueue, final Executor executor) {
Future<RemoteServiceRequest> take = requestQueue.takeAsync();
take.addListener(new FutureListener<RemoteServiceRequest>() {
@Override
Expand All @@ -154,7 +159,7 @@ public void operationComplete(Future<RemoteServiceRequest> future) throws Except
return;
}
// re-subscribe after a failed takeAsync
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
return;
}

Expand All @@ -166,7 +171,7 @@ public void operationComplete(Future<RemoteServiceRequest> future) throws Except
if (request.getOptions().isAckExpected() && System.currentTimeMillis() - request.getDate() > request.getOptions().getAckTimeoutInMillis()) {
log.debug("request: {} has been skipped due to ackTimeout");
// re-subscribe after a skipped ackTimeout
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
return;
}

Expand Down Expand Up @@ -196,27 +201,46 @@ public void operationComplete(Future<Boolean> future) throws Exception {
return;
}
// re-subscribe after a failed send (ack)
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
return;
}

if (!future.getNow()) {
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
return;
}

invokeMethod(remoteInterface, requestQueue, request, method, responseName);

if (executor != null) {
executor.execute(new Runnable() {
@Override
public void run() {
invokeMethod(remoteInterface, requestQueue, request, method, responseName, executor);
}
});
} else {
invokeMethod(remoteInterface, requestQueue, request, method, responseName, executor);
}
}
});
} else {
invokeMethod(remoteInterface, requestQueue, request, method, responseName);
if (executor != null) {
executor.execute(new Runnable() {
@Override
public void run() {
invokeMethod(remoteInterface, requestQueue, request, method, responseName, executor);
}
});
} else {
invokeMethod(remoteInterface, requestQueue, request, method, responseName, executor);
}
}
}

});
}

private <T> void invokeMethod(final Class<T> remoteInterface, final RBlockingQueue<RemoteServiceRequest> requestQueue, final RemoteServiceRequest request, RemoteServiceMethod method, String responseName) {
private <T> void invokeMethod(final Class<T> remoteInterface, final RBlockingQueue<RemoteServiceRequest> requestQueue,
final RemoteServiceRequest request, RemoteServiceMethod method, String responseName, final Executor executor) {
final AtomicReference<RemoteServiceResponse> responseHolder = new AtomicReference<RemoteServiceResponse>();
try {
Object result = method.getMethod().invoke(method.getBean(), request.getArgs());
Expand All @@ -241,12 +265,12 @@ public void operationComplete(Future<List<?>> future) throws Exception {
}
}
// re-subscribe anyways (fail or success) after the send (response)
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
}
});
} else {
// re-subscribe anyways after the method invocation
subscribe(remoteInterface, requestQueue);
subscribe(remoteInterface, requestQueue, executor);
}
}

Expand Down
19 changes: 15 additions & 4 deletions src/main/java/org/redisson/api/RRemoteService.java
View file
Expand Up @@ -15,6 +15,7 @@
*/
package org.redisson.api;

import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
Expand Down Expand Up @@ -58,21 +59,31 @@
public interface RRemoteService {

/**
* Register remote service with single executor
* Register remote service with single worker
*
* @param remoteInterface
* @param object
*/
<T> void register(Class<T> remoteInterface, T object);

/**
* Register remote service with custom executors amount
* Register remote service with custom workers amount
*
* @param remoteInterface
* @param object
* @param executorsAmount
* @param workersAmount
*/
<T> void register(Class<T> remoteInterface, T object, int executorsAmount);
<T> void register(Class<T> remoteInterface, T object, int workersAmount);

/**
* Register remote service with custom workers amount
* and executor for running them
*
* @param remoteInterface
* @param object
* @param workersAmount
*/
<T> void register(Class<T> remoteInterface, T object, int workersAmount, Executor executor);

/**
* Get remote service object for remote invocations.
Expand Down
45 changes: 36 additions & 9 deletions src/test/java/org/redisson/RedissonRemoteServiceTest.java
View file
@@ -1,6 +1,18 @@
package org.redisson;

import io.netty.handler.codec.EncoderException;
import static org.assertj.core.api.Assertions.assertThat;

import java.io.IOException;
import java.io.NotSerializableException;
import java.io.Serializable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Assert;
import org.junit.Test;
import org.redisson.api.RemoteInvocationOptions;
Expand All @@ -10,15 +22,7 @@
import org.redisson.remote.RemoteServiceAckTimeoutException;
import org.redisson.remote.RemoteServiceTimeoutException;

import java.io.IOException;
import java.io.NotSerializableException;
import java.io.Serializable;
import io.netty.util.concurrent.Future;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.assertj.core.api.Assertions.assertThat;

public class RedissonRemoteServiceTest extends BaseTest {

Expand Down Expand Up @@ -183,6 +187,29 @@ public void testAsync() throws InterruptedException {
r1.shutdown();
r2.shutdown();
}

@Test
public void testExecutorAsync() throws InterruptedException {
RedissonClient r1 = createInstance();
ExecutorService executor = Executors.newSingleThreadExecutor();
r1.getRemoteSerivce().register(RemoteInterface.class, new RemoteImpl(), 1, executor);

RedissonClient r2 = createInstance();
RemoteInterfaceAsync ri = r2.getRemoteSerivce().get(RemoteInterfaceAsync.class);

Future<Void> f = ri.voidMethod("someName", 100L);
f.sync();
Future<Long> resFuture = ri.resultMethod(100L);
resFuture.sync();
assertThat(resFuture.getNow()).isEqualTo(200);

r1.shutdown();
r2.shutdown();

executor.shutdown();
executor.awaitTermination(1, TimeUnit.MINUTES);
}


@Test
public void testExecutorsAmountConcurrency() throws InterruptedException {
Expand Down

0 comments on commit 6615293

Please sign in to comment.