diff --git a/README.md b/README.md index 2e4ee8a..fdfab86 100644 --- a/README.md +++ b/README.md @@ -1,20 +1,19 @@

-yarmi is yet anotehr RMI based on JSON. it's simple yet powerful when developing server & client based distributed application within a network of small scale +yarmi is yet-another remote method invocation framework for simple distributed service architecture which provides service discovery mechanism out of the box. [![Codacy Badge](https://api.codacy.com/project/badge/Grade/5c9f40d574c64e629af11f284c447bea)](https://www.codacy.com/app/innocentevil0914/yarmi?utm_source=github.com&utm_medium=referral&utm_content=fritzprix/yarmi&utm_campaign=Badge_Grade) ### Features -1. Support large blob as method parameter or response +1. Simple APIs +> discover and request service with just a few API calls +2. Support large blob as method parameter or response > yarmi supports blob exchange between client and server by default with BlobSession which exposes familiar read / write APIs -2. Provide service discovery out-of-the-box -> yarmi contains simple service discovery feature and also support another type of service discovery (e.g. DNS-SD) as module -3. Support various transport -> yarmi also provides abstraction over transport layer so it can over any kinds of transport like tcp / ip or bluetooth rfcomm. -4. Zero-cost migration to (from) RESTful application -> Provides conceptual similarity to popular RESTful application framework (like service / controller mapping). -> and that means not only the migration from / to RESTful implementation is easy -> but also implementing proxy for any RESTful service in heterogeneous network scenario (like typical IoT application) is also well supported +3. Provide service discovery out-of-the-box +> service discovery is provided out of the box which is not dependent on any other lookup service. +4. Extensible design +> yarmi core itself is agnostic to network / messaging / discovery / negotiation implementation. + ### How-To @@ -42,7 +41,7 @@ yarmi is yet anotehr RMI based on JSON. it's simple yet powerful when developing com.doodream yarmi-core - 0.0.4 + 0.0.5 ``` @@ -148,8 +147,8 @@ public static class SimpleClient { SimpleServiceDiscovery discovery = new SimpleServiceDiscovery(); discovery.startDiscovery(TestService.class, new ServiceDiscoveryListener() { @Override - public void onDiscovered(RMIServiceProxy proxy) { - discoveredService.add(proxy); + public void onDiscovered(RMIServiceInfo info) { + discoveredService.add(RMIServiceInfo.toServiceProxy(info)); } @Override diff --git a/pom.xml b/pom.xml index 351b44e..5babd44 100644 --- a/pom.xml +++ b/pom.xml @@ -153,19 +153,19 @@ 1.0 - org.mongodb - bson - 3.2.2 + com.fasterxml.jackson.core + jackson-core + 2.9.6 - ch.qos.logback - logback-classic - 1.3.0-alpha4 + de.undercouch + bson4jackson + 2.9.2 - ch.qos.logback - logback-access - 1.3.0-alpha4 + org.slf4j + slf4j-api + 1.7.25 com.google.code.gson diff --git a/src/main/java/com/doodream/rmovjs/annotation/RMIException.java b/src/main/java/com/doodream/rmovjs/annotation/RMIException.java new file mode 100644 index 0000000..a9f6bd4 --- /dev/null +++ b/src/main/java/com/doodream/rmovjs/annotation/RMIException.java @@ -0,0 +1,16 @@ +package com.doodream.rmovjs.annotation; + +import com.doodream.rmovjs.model.Response; + +public class RMIException extends RuntimeException { + private int code; + + public RMIException(Response response) { + super((String) response.getBody()); + code = response.getCode(); + } + + public int code() { + return code; + } +} diff --git a/src/main/java/com/doodream/rmovjs/annotation/server/Service.java b/src/main/java/com/doodream/rmovjs/annotation/server/Service.java index 0987de5..200c2ee 100644 --- a/src/main/java/com/doodream/rmovjs/annotation/server/Service.java +++ b/src/main/java/com/doodream/rmovjs/annotation/server/Service.java @@ -6,12 +6,13 @@ import com.doodream.rmovjs.net.SimpleNegotiator; import com.doodream.rmovjs.net.tcp.TcpServiceAdapter; import com.doodream.rmovjs.serde.Converter; -import com.doodream.rmovjs.serde.json.JsonConverter; +import com.doodream.rmovjs.serde.bson.BsonConverter; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; /** + * Service Annotation * Created by innocentevil on 18. 5. 4. */ @@ -34,8 +35,8 @@ Class adapter() default TcpServiceAdapter.class; /** - * parameters for constrcutor of network adapter class, will be passed as argument whenever the adapter class - * needs to be instanciated. + * parameters for constructor of network adapter class, will be passed as argument whenever the adapter class + * needs to be instantiated. * the order of parameters will be kept in the process of ser-der. * @return parameters to adapter constructor */ @@ -43,5 +44,5 @@ Class negotiator() default SimpleNegotiator.class; - Class converter() default JsonConverter.class; + Class converter() default BsonConverter.class; } diff --git a/src/main/java/com/doodream/rmovjs/client/HaRMIClient.java b/src/main/java/com/doodream/rmovjs/client/HaRMIClient.java index 390a2fa..075aa2e 100644 --- a/src/main/java/com/doodream/rmovjs/client/HaRMIClient.java +++ b/src/main/java/com/doodream/rmovjs/client/HaRMIClient.java @@ -1,78 +1,243 @@ package com.doodream.rmovjs.client; +import com.doodream.rmovjs.annotation.RMIException; +import com.doodream.rmovjs.annotation.server.Controller; import com.doodream.rmovjs.annotation.server.Service; -import com.doodream.rmovjs.model.Endpoint; +import com.doodream.rmovjs.method.RMIMethod; +import com.doodream.rmovjs.model.RMIError; import com.doodream.rmovjs.model.RMIServiceInfo; import com.doodream.rmovjs.net.RMIServiceProxy; import com.doodream.rmovjs.sdp.ServiceDiscovery; import com.doodream.rmovjs.sdp.ServiceDiscoveryListener; -import com.doodream.rmovjs.serde.Converter; -import com.doodream.rmovjs.util.LruCache; +import com.google.common.base.Preconditions; import io.reactivex.Observable; import io.reactivex.disposables.CompositeDisposable; import io.reactivex.schedulers.Schedulers; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Data; -import lombok.NoArgsConstructor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Consumer; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; /** * High-Available RMI Client */ -public class HaRMIClient implements InvocationHandler, Consumer { - +public class HaRMIClient implements InvocationHandler { + private interface Selectable { + RMIClient selectNext(List proxies, RMIClient lastSelected); + } /** - * policy + * types of policy used to select service every request */ - public enum RequestRoutePolicy { - RoundRobin, -s } + public enum RequestRoutePolicy implements Selectable { + RoundRobin { + @Override + public RMIClient selectNext(List proxies, RMIClient lastSelected) { + if(lastSelected == null) { + return proxies.get(0); + } + if(proxies.contains(lastSelected)) { + int idx = proxies.indexOf(lastSelected) + 1; + if(idx >= proxies.size()) { + idx = 0; + } + return proxies.get(idx); + } + return proxies.get(0); + } + }, + FastestFirst { + @Override + public RMIClient selectNext(List proxies, RMIClient lastSelected) { + return Observable.fromIterable(proxies) + .sorted(RMIClient::compareTo) + .blockingFirst(); + } + }, + Random { + @Override + public RMIClient selectNext(List proxies, RMIClient lastSelected) { + return null; + } + } + } - private static final Logger Log = LoggerFactory.getLogger(HaRMIClient.class); + public interface AvailabilityChangeListener { + void onAvailabilityChanged(int availableServices); + } - private String controllerPath; - private Map methodMap; - private RMIServiceProxy serviceProxy; - private RMIServiceInfo serviceInfo; - private Converter converter; + private static final long DEFAULT_QOS_UPDATE_PERIOD = 2000L; + private static final long AVAILABILITY_WAIT_TIMEOUT = 5000L; + private static final int MAX_TRIAL_COUNT = 3; + private static final Logger Log = LoggerFactory.getLogger(HaRMIClient.class); - private ServiceDiscovery discovery; + private AvailabilityChangeListener availabilityChangeListener; + private RequestRoutePolicy routePolicy; private CompositeDisposable compositeDisposable; - private LruCache proxies; + private RMIClient lastProxy; + private long qosFactor; + private ExecutorService listenerInvoker; + private HashSet discoveredProxySet; + private final ArrayList clients; + private Class controller; + private Class svc; + private long qosUpdateTime; + private TimeUnit qosUpdateTimeUnit; - public static T create(ServiceDiscovery discovery, Class svc, Class ctrl) throws IllegalAccessException, IOException, InstantiationException { + /** + * close call proxy and release its resources + * @param callProxy call proxy returned by {@link #create(ServiceDiscovery, long, Class, Class, RequestRoutePolicy, AvailabilityChangeListener)} + * @param force if false, caller wait until the on-going requests are finished. if true, close the network connection immediately without waiting + */ + public static void destroy(Object callProxy, boolean force) { + final HaRMIClient client = (HaRMIClient) Proxy.getInvocationHandler(callProxy); + if(client == null) { + return; + } + client.close(force); + } + + /** + * create call proxy for multiple services + * it tries to keep connection to services as many as possible + * @param discovery @{@link ServiceDiscovery} used to discover service + * @param qos latency in millisecond which service should meet to keep connection to this client + * @param svc service definition class annotated by {@link Service} + * @param ctrl controller interface + * @param policy policy used to select service + * @param listener listener to monitor the change of service availability + * @param controller type which proxy is created from + * @return call proxy + */ + public static T create(ServiceDiscovery discovery, long qos, Class svc, Class ctrl, RequestRoutePolicy policy, AvailabilityChangeListener listener) { Service service = (Service) svc.getAnnotation(Service.class); - if(service == null) { - return null; - } + Preconditions.checkNotNull(service); + + + Controller controller = Observable.fromArray(svc.getDeclaredFields()) + .filter(field -> field.getType().equals(ctrl)) + .map(field -> field.getAnnotation(Controller.class)) + .blockingFirst(null); + + + List validMethods = Observable.fromArray(ctrl.getMethods()) + .filter(RMIMethod::isValidMethod).toList().blockingGet(); + + final RMIServiceInfo serviceInfo = RMIServiceInfo.from(svc); + Preconditions.checkNotNull(serviceInfo, "Invalid Service Class %s", svc); + + Preconditions.checkArgument(validMethods.size() > 0); + Preconditions.checkNotNull(controller, "no matched controller"); + Preconditions.checkArgument(ctrl.isInterface()); + + HaRMIClient haRMIClient = new HaRMIClient<>(svc, ctrl, qos, DEFAULT_QOS_UPDATE_PERIOD, TimeUnit.MILLISECONDS, policy); + haRMIClient.availabilityChangeListener = listener; + - HaRMIClient haRMIClient = new HaRMIClient(); CompositeDisposable compositeDisposable = new CompositeDisposable(); compositeDisposable.add(startDiscovery(discovery, svc) .subscribeOn(Schedulers.newThread()) - .subscribe(haRMIClient::accept)); + .subscribe(haRMIClient::registerProxy, haRMIClient::onError)); + haRMIClient.setDisposable(compositeDisposable); - return null; + return (T) Proxy.newProxyInstance(ctrl.getClassLoader(),new Class[]{ ctrl }, haRMIClient); } + /** + * check the availability of the service + * @param callProxy call proxy returned by {@link #create(ServiceDiscovery, long, Class, Class, RequestRoutePolicy, AvailabilityChangeListener)} + * @param blockUntilAvailable if true, caller will block until at least a service available, otherwise return immediately + * @return true, if there is at least an available service, otherwise false + */ + public static boolean isAvailable(Object callProxy, boolean blockUntilAvailable) { + HaRMIClient haRMIClient = (HaRMIClient) Proxy.getInvocationHandler(callProxy); + if(haRMIClient == null) { + return false; + } + int availability; + synchronized (haRMIClient.clients) { + while (!((availability = haRMIClient.getUpdatedAvailability()) > 0) && blockUntilAvailable) { + try { + haRMIClient.clients.wait(AVAILABILITY_WAIT_TIMEOUT); + } catch (InterruptedException e) { + return false; + } + } + } + return availability > 0; + } + + private synchronized int getUpdatedAvailability() { + return clients.size(); + } + + /** + * private constructor + * @param svc service definition annotated with {@link Service} + * @param ctrl class of controller interface + * @param qos minimum Quality of Service (latency) used to determine the service is good or bad, if the service will be disconnected if it doesn't satisfy the QoS + * @param qosUpdatePeriod time value how frequently the QoS be measured. + * @param timeUnit time unit for qosUpdatePeriod + * @param policy policy used to select service for each request + */ + private HaRMIClient(Class svc, Class ctrl, long qos, long qosUpdatePeriod, TimeUnit timeUnit, RequestRoutePolicy policy) { + this.svc = svc; + this.controller = ctrl; + this.routePolicy = policy; + this.qosFactor = qos; + listenerInvoker = Executors.newSingleThreadExecutor(); + clients = new ArrayList<>(); + discoveredProxySet = new HashSet<>(); + qosUpdateTime = qosUpdatePeriod; + qosUpdateTimeUnit = timeUnit; + } + + private void onError(Throwable throwable) { + Log.error(throwable.getLocalizedMessage()); + close(true); + } + + + private void setDisposable(CompositeDisposable compositeDisposable) { + this.compositeDisposable = compositeDisposable; + } + + private synchronized void registerProxy(RMIServiceProxy serviceProxy) { + if (serviceProxy == null) { + return; + } + if (discoveredProxySet.add(serviceProxy.who())) { + clients.add(RMIClient.createClient(serviceProxy, svc, controller, qosFactor, qosUpdateTime, qosUpdateTimeUnit)); + } + + listenerInvoker.submit(() -> { + synchronized (clients) { + availabilityChangeListener.onAvailabilityChanged(clients.size()); + clients.notifyAll(); + } + }); + } + + private static Observable startDiscovery(ServiceDiscovery discovery, Class svc) { - return Observable.create(emitter -> discovery.startDiscovery(svc, new ServiceDiscoveryListener() { + return Observable.create(emitter -> discovery.startDiscovery(svc, false, new ServiceDiscoveryListener() { + @Override - public void onDiscovered(RMIServiceProxy proxy) { - emitter.onNext(proxy); + public void onDiscovered(RMIServiceInfo info) { + emitter.onNext(RMIServiceInfo.toServiceProxy(info)); } @Override @@ -81,23 +246,86 @@ public void onDiscoveryStarted() { } @Override - public void onDiscoveryFinished() throws IllegalAccessException { + public void onDiscoveryFinished() { emitter.onComplete(); } })); } - private HaRMIClient() { + + private void close(boolean force) { + compositeDisposable.dispose(); + listenerInvoker.shutdown(); + clients.forEach(client -> { + try { + client.close(force); + } catch (IOException e) { + e.printStackTrace(); + } + }); + } + + + private RMIClient selectNext() throws TimeoutException { + RMIClient selected; + int trial = MAX_TRIAL_COUNT; + try { + do { + selected = routePolicy.selectNext(clients, lastProxy); + if (selected != null) { + return selected; + } + synchronized (clients) { + clients.wait(); + } + } while (trial-- > 0); + } catch (InterruptedException ignore) { } + throw new TimeoutException(String.format("fail to select next client /w %s", routePolicy)); } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { - return null; + lastProxy = selectNext(); + Preconditions.checkNotNull(lastProxy, "invalid call proxy : null call proxy"); + try { + return lastProxy.invoke(lastProxy, method, args); + } catch (RMIException e) { + if(RMIError.isServiceBad(e.code())) { + purgeBadProxy(lastProxy); + } + // rethrow it + throw e; + } } @Override - public void accept(RMIServiceProxy serviceProxy) { + protected void finalize() throws Throwable { + close(true); + super.finalize(); + } + /** + * purge bad proxy + * @param client client which doesn't meet QoS requirement + */ + private synchronized void purgeBadProxy(RMIClient client) { + Log.debug("Purge client {}", client); + clients.remove(client); + if(!discoveredProxySet.remove(client.who())) { + Log.warn("client ({}) is not in the discovered set", client); + } + try { + client.close(true); + } catch (IOException e) { + Log.warn(e.getMessage()); + } + listenerInvoker.submit(() -> { + synchronized (client) { + availabilityChangeListener.onAvailabilityChanged(clients.size()); + client.notifyAll(); + } + }); } + } diff --git a/src/main/java/com/doodream/rmovjs/client/RMIClient.java b/src/main/java/com/doodream/rmovjs/client/RMIClient.java index 2504ca1..9b177dd 100644 --- a/src/main/java/com/doodream/rmovjs/client/RMIClient.java +++ b/src/main/java/com/doodream/rmovjs/client/RMIClient.java @@ -1,157 +1,272 @@ package com.doodream.rmovjs.client; +import com.doodream.rmovjs.annotation.RMIException; import com.doodream.rmovjs.annotation.server.Controller; import com.doodream.rmovjs.annotation.server.Service; import com.doodream.rmovjs.method.RMIMethod; import com.doodream.rmovjs.model.Endpoint; +import com.doodream.rmovjs.model.RMIError; import com.doodream.rmovjs.model.RMIServiceInfo; import com.doodream.rmovjs.model.Response; import com.doodream.rmovjs.net.RMIServiceProxy; import com.google.common.base.Preconditions; import io.reactivex.Observable; import io.reactivex.Single; +import io.reactivex.schedulers.Schedulers; +import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; -import java.lang.reflect.Field; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; /** * {@link RMIClient} build method invocation proxy from {@link RMIServiceProxy} which is discovered from SDP * */ -public class RMIClient implements InvocationHandler { +public class RMIClient implements InvocationHandler, Comparable { private static final Logger Log = LoggerFactory.getLogger(RMIClient.class); private Map methodMap; private RMIServiceProxy serviceProxy; + private AtomicInteger ongoingRequestCount; + private long timeout; + private Long measuredPing; + private volatile boolean markToClose; - private RMIClient(RMIServiceProxy serviceProxy) { + private RMIClient(RMIServiceProxy serviceProxy, long timeout,long pingUpdatePeriod, TimeUnit timeUnit) { this.serviceProxy = serviceProxy; + markToClose = false; + measuredPing = Long.MAX_VALUE; + this.timeout = timeout; + ongoingRequestCount = new AtomicInteger(0); + if(pingUpdatePeriod > 0L) { + serviceProxy.startPeriodicQosUpdate(timeout, pingUpdatePeriod, timeUnit); + } + } + + /** + * return QoS value measured updated last + * @param proxy proxy object + * @return QoS value (defined latency in millisecond from request to response) + */ + public static long getMeasuredQoS(Object proxy) { + return forEachClient(proxy) + .blockingFirst().getMeasuredPing(); } + /** + * check whether there are on-going requests for given RMI call proxy + * @param proxy RMI proxy which is create by {@link #create(RMIServiceProxy, Class, Class)} or {@link #create(RMIServiceProxy, Class, Class, long, long, TimeUnit)} + * @return true if there is no on-going request, otherwise false + */ + public static boolean isClosable(Object proxy) { + return forEachClient(proxy) + .blockingFirst().isClosable(); + } + + private void setMethodEndpointMap(Map map) { this.methodMap = map; } + private boolean isClosable() { + return ongoingRequestCount.get() == 0; + } - public static void destroy(Object proxy) throws IOException { - Class proxyClass = proxy.getClass(); - if(Proxy.isProxyClass(proxyClass)) { - RMIClient client = (RMIClient) Proxy.getInvocationHandler(proxy); - client.close(); - } else { - Service service = proxy.getClass().getAnnotation(Service.class); - if(service == null) { - throw new RuntimeException("Invalid Proxy"); - } - Observable.fromArray(proxyClass.getDeclaredFields()) - .filter(field -> field.getAnnotation(Controller.class) != null) - .map(field -> field.get(proxy)) - .blockingSubscribe(RMIClient::destroy); - } + /** + * get {@link RMIClient} for given RMI call proxy + * @param proxy + * @return + */ + static RMIClient access(Object proxy) { + return forEachClient(proxy) + .blockingFirst(); } + /** + * destroy RMI call proxy and release resources + * @param proxy RMI call proxy returned by {@link #create(RMIServiceProxy, Class, Class, long, long, TimeUnit)} or {@link #create(RMIServiceProxy, Class, Class)} + * @param force if true, close regardless its on-going request, otherwise, wait until the all the on-going requests is complete + */ + public static void destroy(Object proxy, boolean force) { + forEachClient(proxy) + .subscribeOn(Schedulers.io()) + .blockingSubscribe(client -> client.close(force)); + } - public static T createService(RMIServiceProxy serviceProxy, Class svc) throws IllegalAccessException, InstantiationException { - Object svcProxy = svc.newInstance(); + private static Observable forEachClient(Object proxy) { + return Observable.create(emitter -> { + Class proxyClass = proxy.getClass(); + if(Proxy.isProxyClass(proxyClass)) { + RMIClient client = (RMIClient) Proxy.getInvocationHandler(proxy); + emitter.onNext(client); + } else { + Service service = proxy.getClass().getAnnotation(Service.class); + if(service == null) { + throw new IllegalArgumentException("Invalid Proxy"); + } + Observable.fromArray(proxyClass.getDeclaredFields()) + .filter(field -> field.getAnnotation(Controller.class) != null) + .map(field -> field.get(proxy)) + .map(Proxy::getInvocationHandler) + .cast(RMIClient.class) + .blockingSubscribe(emitter::onNext); + } + emitter.onComplete(); + }); + } - Observable.fromArray(svc.getDeclaredFields()) - .filter(field -> field.getAnnotation(Controller.class) != null) - .blockingSubscribe(field -> { - Object controller = create(serviceProxy, svc, field.getType()); - field.set(svcProxy, controller); - }); - return (T) svcProxy; + /** + * close method invocation proxy created by {@link #create(RMIServiceProxy, Class, Class)} or {@link #createService(RMIServiceProxy, Class)} method + * @param proxy returned proxy instance from {@link #create(RMIServiceProxy, Class, Class)} or {@link #createService(RMIServiceProxy, Class)} + */ + public static void destroy(Object proxy) { + destroy(proxy, false); } /** * * @param serviceProxy * @param svc - * @param ctrl * @param * @return + * @throws IllegalAccessError + * @throws InstantiationException + * @throws IllegalAccessException */ - @Nullable - public static T create(RMIServiceProxy serviceProxy, Class svc, Class ctrl) { - Service service = (Service) svc.getAnnotation(Service.class); + public static T createService(RMIServiceProxy serviceProxy, Class svc) throws IllegalAccessError, InstantiationException, IllegalAccessException { + return createService(serviceProxy, svc, 0L, 0L, TimeUnit.MILLISECONDS); + } + + /** + * create service proxy instance which contains controller interface as its member fields. + * @param serviceProxy active service proxy which is obtained from the service discovery + * @param svc Service definition class + * @param Type of service class + * @return proxy instance for service, getter or direct field referencing can be used to access controller + * @throws IllegalAccessException there is no public constructor for service class + * @throws InstantiationException if this {@code Class} represents an abstract class, + * * an interface, an array class, a primitive type, or void; + * * or if the class has no nullary constructor; + * * or if the instantiation fails for some other reason. + */ + public static T createService(RMIServiceProxy serviceProxy, Class svc, long timeout, long pingInterval, TimeUnit timeUnit) throws IllegalAccessException, InstantiationException { + Object svcProxy = svc.newInstance(); + Observable.fromArray(svc.getDeclaredFields()) + .filter(field -> field.getAnnotation(Controller.class) != null) + .blockingSubscribe(field -> { + Object controller = create(serviceProxy, svc, field.getType(), pingInterval, timeout, timeUnit); + field.setAccessible(true); + field.set(svcProxy, controller); + }); + + return (T) svcProxy; + } + + static RMIClient createClient(RMIServiceProxy serviceProxy, Class svc, Class ctrl, long pingTimeout, long pingInterval, TimeUnit timeUnit) { + Service service = svc.getAnnotation(Service.class); + Preconditions.checkNotNull(service); if(!serviceProxy.provide(ctrl)) { + Log.warn("service is not supported"); return null; } + Controller controller = Observable.fromArray(svc.getDeclaredFields()) + .filter(field -> field.getType().equals(ctrl)) + .map(field -> field.getAnnotation(Controller.class)) + .blockingFirst(null); - try { - Preconditions.checkNotNull(service); - Controller controller = Observable.fromArray(svc.getDeclaredFields()) - .filter(field -> field.getType().equals(ctrl)) - .map(field -> field.getAnnotation(Controller.class)) - .blockingFirst(null); - - Preconditions.checkNotNull(controller, "no matched controller"); - Preconditions.checkArgument(ctrl.isInterface()); - + final RMIServiceInfo serviceInfo = RMIServiceInfo.from(svc); + List validMethods = Observable.fromArray(ctrl.getMethods()) + .filter(RMIMethod::isValidMethod).toList().blockingGet(); - final RMIServiceInfo serviceInfo = RMIServiceInfo.from(svc); - Preconditions.checkNotNull(serviceInfo, "Invalid Service Class %s", svc); + Preconditions.checkNotNull(controller, "no matched controller"); + Preconditions.checkArgument(ctrl.isInterface()); + Preconditions.checkNotNull(serviceInfo, "Invalid Service Class %s", svc); + Preconditions.checkArgument(validMethods.size() > 0); + try { if (!serviceProxy.isOpen()) { + // RMIServiceProxy is opened only once serviceProxy.open(); } - RMIClient rmiClient = new RMIClient(serviceProxy); - + RMIClient rmiClient = new RMIClient(serviceProxy, pingTimeout, pingInterval, timeUnit); - Observable methodObservable = Observable.fromArray(ctrl.getMethods()) - .filter(RMIMethod::isValidMethod); - - Observable endpointObservable = methodObservable + Observable endpointObservable = Observable.fromIterable(validMethods) .map(method -> Endpoint.create(controller, method)); - Single> hashMapSingle = methodObservable + Single> hashMapSingle = Observable.fromIterable(validMethods) .zipWith(endpointObservable, RMIClient::zipIntoMethodMap) .collectInto(new HashMap<>(), RMIClient::collectMethodMap); rmiClient.setMethodEndpointMap(hashMapSingle.blockingGet()); - // TODO: 18. 7. 31 consider give all the available controller interface to the call proxy + // 18. 7. 31 consider give all the available controller interface to the call proxy // main concern is... // what happen if there are two methods declared in different interfaces which is identical in parameter & return type, etc. - // TODO: 18. 7. 31 method collision is not properly handled at the moment, simple poc is performed to test + // method collision is not properly handled at the moment, simple poc is performed to test // https://gist.github.com/fritzprix/ca0ecc08fc3125cde529dd11185be0b9 - Object proxy = Proxy.newProxyInstance(ctrl.getClassLoader(), new Class[]{ctrl }, rmiClient); - - Log.debug("service proxy is created"); - return (T) proxy; + return rmiClient; } catch (Exception e) { - Log.error("{}", e); + Log.error("", e); + return null; + } + } + + public static T create(RMIServiceProxy serviceProxy, Class svc, Class ctrl, long pingTimeout, long pingInterval, TimeUnit timeUnit) { + RMIClient rmiClient = createClient(serviceProxy, svc, ctrl, pingTimeout, pingInterval, timeUnit); + if(rmiClient == null) { return null; } + return (T) Proxy.newProxyInstance(ctrl.getClassLoader(), new Class[] {ctrl}, rmiClient); + } + + /** + * create call proxy instance corresponding to given controller class + * @param serviceProxy active service proxy which is obtained from the service discovery + * @param svc Service definition class + * @param ctrl controller definition as interface + * @param type of controller class + * @return call proxy instance for controller + */ + @Nullable + public static T create(RMIServiceProxy serviceProxy, Class svc, Class ctrl) { + return create(serviceProxy, svc, ctrl, 0L, 0L, TimeUnit.MILLISECONDS); } @Override protected void finalize() throws Throwable { super.finalize(); - serviceProxy.close(); + try { + close(true); + } catch (IOException ignore) { + + } finally { + super.finalize(); + } } /** - * - * @param into - * @param methodEndpointMap + * collect maps between {@link Method} and @{@link Endpoint} into single map + * @param into map collecting fragmented (or partial) map of {@link Method} and {@link Endpoint} + * @param methodEndpointMap fragmented (or partial) map of {@link Method} and {@link Endpoint} */ - private static void collectMethodMap(Map into, Map methodEndpointMap) { + static void collectMethodMap(Map into, Map methodEndpointMap) { into.putAll(methodEndpointMap); } @@ -161,28 +276,60 @@ private static void collectMethodMap(Map into, Map zipIntoMethodMap(Method method, Endpoint endpoint) { + static Map zipIntoMethodMap(Method method, Endpoint endpoint) { Map methodEndpointMap = new HashMap<>(); methodEndpointMap.put(method, endpoint); return methodEndpointMap; } - private void close() throws IOException { + /** + * close service proxy used by this call proxy + * @param force if false, wait until on-going request complete, otherwise, close immediately + * @throws IOException proxy is already closed, + */ + void close(boolean force) throws IOException { + markToClose = true; + if(!force) { + try { + while (!isClosable()) { + // wait until proxy is closable + Thread.sleep(10L); + } + } catch (InterruptedException ignored) { } + } + serviceProxy.close(); } - @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if(markToClose) { + // prevent new request from being made + throw new RMIException(RMIError.CLOSED.getResponse()); + } Endpoint endpoint = methodMap.get(method); if(endpoint == null) { return null; } - try { - return serviceProxy.request(endpoint, args); - } catch (IOException e) { - serviceProxy.close(); - return Response.error(-1, e.getLocalizedMessage()); + ongoingRequestCount.getAndIncrement(); + Response response = serviceProxy.request(endpoint, timeout, args); + ongoingRequestCount.decrementAndGet(); + if(response.isSuccessful()) { + return response; } + throw new RMIException(response); + } + + @Override + public int compareTo(@NotNull RMIClient o) { + return Math.toIntExact(getMeasuredPing() - o.getMeasuredPing()); + } + + private long getMeasuredPing() { + return measuredPing; + } + + String who() { + return serviceProxy.who(); } } diff --git a/src/main/java/com/doodream/rmovjs/method/RMIMethod.java b/src/main/java/com/doodream/rmovjs/method/RMIMethod.java index 7bb0f33..19ecffd 100644 --- a/src/main/java/com/doodream/rmovjs/method/RMIMethod.java +++ b/src/main/java/com/doodream/rmovjs/method/RMIMethod.java @@ -57,6 +57,7 @@ public String extractPath(java.lang.reflect.Method method) { // but still inheritance of annotation is not supported (as I know) // TODO : fix if Java provides annotation inheritance feature + switch (this) { case GET: Get get = method.getAnnotation(Get.class); diff --git a/src/main/java/com/doodream/rmovjs/model/ControllerInfo.java b/src/main/java/com/doodream/rmovjs/model/ControllerInfo.java index 3344bc4..74cb265 100644 --- a/src/main/java/com/doodream/rmovjs/model/ControllerInfo.java +++ b/src/main/java/com/doodream/rmovjs/model/ControllerInfo.java @@ -16,8 +16,8 @@ public class ControllerInfo { public static ControllerInfo build(RMIController controller) { return ControllerInfo.builder() - .stubCls(controller.getItfcCls()) .version(controller.getController().version()) + .stubCls(controller.getStub()) .build(); } } diff --git a/src/main/java/com/doodream/rmovjs/model/Endpoint.java b/src/main/java/com/doodream/rmovjs/model/Endpoint.java index 49ee263..fa6ecd8 100644 --- a/src/main/java/com/doodream/rmovjs/model/Endpoint.java +++ b/src/main/java/com/doodream/rmovjs/model/Endpoint.java @@ -11,9 +11,12 @@ import com.doodream.rmovjs.parameter.Param; import com.doodream.rmovjs.util.Types; import com.google.common.base.Preconditions; -import com.google.gson.reflect.TypeToken; import io.reactivex.Observable; import io.reactivex.Single; +import io.reactivex.functions.BiFunction; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -23,10 +26,10 @@ import java.lang.annotation.Annotation; import java.lang.reflect.Method; -import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.List; import java.util.Locale; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -57,33 +60,83 @@ public static Endpoint create(Controller controller, Method method) { Annotation methodAnnotation = Observable .fromArray(method.getAnnotations()) - .filter(Endpoint::verifyMethod) + .filter(new Predicate() { + @Override + public boolean test(Annotation annotation) throws Exception { + return Endpoint.verifyMethod(annotation); + } + }) .blockingFirst(); final String parentPath = controller.path(); RMIMethod rmiMethod = RMIMethod.fromAnnotation(methodAnnotation); - Observable typeObservable = Observable.fromArray(method.getParameterTypes()); + Observable typeObservable = Observable.fromArray(method.getGenericParameterTypes()); Observable annotationsObservable = Observable.fromArray(method.getParameterAnnotations()); final String paramUnique = typeObservable .defaultIfEmpty(Void.class) - .map(Type::getTypeName) - .map("_"::concat) - .reduce(String::concat) + .map(new Function() { + @Override + public String apply(Type type) throws Exception { + return type.getTypeName(); + } + }) + .map(new Function() { + @Override + public String apply(String s) throws Exception { + return "_".concat(s); + } + }) + .reduce(new BiFunction() { + @Override + public String apply(String s, String s2) throws Exception { + return s.concat(s2); + } + }) .blockingGet(); Single respBlobObservable = Observable.fromArray(method.getGenericReturnType().getTypeName()) - .map(TYPE_PATTERN::matcher) - .filter(Matcher::matches) - .map(matcher -> matcher.group(1)) - .filter(s -> s.contains(BlobSession.class.getName())) + .map(new Function() { + @Override + public Matcher apply(String s) throws Exception { + return TYPE_PATTERN.matcher(s); + } + }) + .filter(new Predicate() { + @Override + public boolean test(Matcher matcher) throws Exception { + return matcher.matches(); + } + }) + .map(new Function() { + @Override + public String apply(Matcher matcher) throws Exception { + return matcher.group(1); + } + }) + .filter(new Predicate() { + @Override + public boolean test(String s) throws Exception { + return s.contains(BlobSession.class.getName()); + } + }) .count(); final Long blobCount = typeObservable - .filter(aClass -> aClass.equals(BlobSession.class)) - .count().zipWith(respBlobObservable, Math::addExact).blockingGet(); + .filter(new Predicate() { + @Override + public boolean test(Type type) throws Exception { + return type.equals(BlobSession.class); + } + }) + .count().zipWith(respBlobObservable, new BiFunction() { + @Override + public Long apply(Long aLong, Long aLong2) throws Exception { + return Math.addExact(aLong, aLong2); + } + }).blockingGet(); if(blobCount > 1) { throw new IllegalArgumentException(String.format("too many BlobSession in method @ %s", method.getName())); @@ -92,11 +145,21 @@ public static Endpoint create(Controller controller, Method method) { final String path = String.format(Locale.ENGLISH, "%s%s", parentPath, rmiMethod.extractPath(method)).replaceAll(DUPLICATE_PATH_SEPARATOR, "/"); - final int[] order = {0}; + final AtomicInteger order = new AtomicInteger(0); List params = typeObservable - .zipWith(annotationsObservable, Param::create) - .doOnNext(param -> param.setOrder(order[0]++)) + .zipWith(annotationsObservable, new BiFunction() { + @Override + public Param apply(Type type, Annotation[] annotations) throws Exception { + return Param.create(type, annotations); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Param param) throws Exception { + param.setOrder(order.getAndIncrement()); + } + }) .toList().blockingGet(); final String methodLookupKey = String.format("%x%x%x", rmiMethod.name().hashCode(), controller.path().hashCode(), paramUnique.hashCode()).toUpperCase(); diff --git a/src/main/java/com/doodream/rmovjs/model/RMIError.java b/src/main/java/com/doodream/rmovjs/model/RMIError.java index d88324a..ce6d63c 100644 --- a/src/main/java/com/doodream/rmovjs/model/RMIError.java +++ b/src/main/java/com/doodream/rmovjs/model/RMIError.java @@ -9,16 +9,28 @@ public enum RMIError { FORBIDDEN(403, Response.error(403, "ServiceInfo is not matched")), BAD_REQUEST(400, Response.error(400, "Bad Request")), UNHANDLED(400, Response.error(400, "Request Not Handled")), - INTERNAL_SERVER_ERROR(500, Response.error(500,"Internal Server Error")); + INTERNAL_SERVER_ERROR(500, Response.error(500,"Internal Server Error")), + TIMEOUT(500, Response.error(500,"Timeout")), + BAD_RESPONSE(501, Response.error(501,"Bad Response")), + CLOSED(510, Response.error(510, "RMI channel close")); private final int code; - private final Response response; - RMIError(int code, Response response) { + private final Response response; + RMIError(int code, Response response) { this.code = code; this.response = response; } + public static boolean isServiceBad(int code) { + return code >= 500; + } + public Response getResponse() { return response; } + + public int code() { + return code; + } + } diff --git a/src/main/java/com/doodream/rmovjs/model/RMIServiceInfo.java b/src/main/java/com/doodream/rmovjs/model/RMIServiceInfo.java index 4bf9354..d6e2263 100644 --- a/src/main/java/com/doodream/rmovjs/model/RMIServiceInfo.java +++ b/src/main/java/com/doodream/rmovjs/model/RMIServiceInfo.java @@ -2,17 +2,25 @@ import com.doodream.rmovjs.Properties; import com.doodream.rmovjs.annotation.server.Service; +import com.doodream.rmovjs.net.RMIServiceProxy; +import com.doodream.rmovjs.net.ServiceAdapter; +import com.doodream.rmovjs.net.ServiceProxyFactory; import com.doodream.rmovjs.server.RMIController; import com.google.gson.annotations.SerializedName; import io.reactivex.Observable; -import lombok.Builder; -import lombok.Data; -import lombok.EqualsAndHashCode; +import io.reactivex.Single; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; +import lombok.*; +import java.lang.reflect.Field; import java.util.Arrays; import java.util.List; @Builder +@NoArgsConstructor +@AllArgsConstructor @EqualsAndHashCode(exclude = {"proxyFactoryHint"}) @Data public class RMIServiceInfo { @@ -39,7 +47,7 @@ public class RMIServiceInfo { private List controllerInfos; /** - * remoteHint is used to guess conntion information (like address or bluetooth device name etc.,) + * remoteHint is used to guess connection information (like address or bluetooth device name etc.,) * */ @SerializedName("hint") @@ -48,7 +56,7 @@ public class RMIServiceInfo { public static RMIServiceInfo from(Class svc) { Service service = svc.getAnnotation(Service.class); - RMIServiceInfoBuilder builder = RMIServiceInfo.builder(); + final RMIServiceInfoBuilder builder = RMIServiceInfo.builder(); builder.version(Properties.getVersionString()) .adapter(service.adapter()) @@ -58,21 +66,83 @@ public static RMIServiceInfo from(Class svc) { .name(service.name()); Observable.fromArray(svc.getDeclaredFields()) - .filter(RMIController::isValidController) - .map(RMIController::create) - .map(ControllerInfo::build) + .filter(new Predicate() { + @Override + public boolean test(Field field) throws Exception { + return RMIController.isValidController(field); + } + }) + .map(new Function() { + @Override + public RMIController apply(Field field) throws Exception { + return RMIController.create(field); + } + }) + .map(new Function() { + @Override + public ControllerInfo apply(RMIController rmiController) throws Exception { + return ControllerInfo.build(rmiController); + } + }) .toList() - .doOnSuccess(builder::controllerInfos) + .doOnSuccess(new Consumer>() { + @Override + public void accept(List controllerInfos) throws Exception { + builder.controllerInfos(controllerInfos); + } + }) .subscribe(); return builder.build(); } - public static boolean isComplete(RMIServiceInfo info) { + public static boolean isValid(RMIServiceInfo info) { return (info.getProxyFactoryHint() != null) && (info.getControllerInfos() != null); } + public static RMIServiceProxy toServiceProxy(RMIServiceInfo info) { + return Single.just(info) + .map(new Function>() { + @Override + public Class apply(RMIServiceInfo rmiServiceInfo) throws Exception { + return rmiServiceInfo.getAdapter(); + } + }) + .map(new Function, Object>() { + @Override + public Object apply(Class cls) throws Exception { + return cls.newInstance(); + } + }) + .cast(ServiceAdapter.class) + .map(new Function() { + @Override + public ServiceProxyFactory apply(ServiceAdapter serviceAdapter) throws Exception { + return serviceAdapter.getProxyFactory(info); + } + }) + .map(new Function() { + @Override + public RMIServiceProxy apply(ServiceProxyFactory serviceProxyFactory) throws Exception { + return serviceProxyFactory.build(); + } + }) + .onErrorReturn(new Function() { + @Override + public RMIServiceProxy apply(Throwable throwable) throws Exception { + return RMIServiceProxy.NULL_PROXY; + } + }) + .filter(new Predicate() { + @Override + public boolean test(RMIServiceProxy proxy) throws Exception { + return !RMIServiceProxy.NULL_PROXY.equals(proxy); + } + }) + .blockingGet(RMIServiceProxy.NULL_PROXY); + } + public void copyFrom(RMIServiceInfo info) { setProxyFactoryHint(info.getProxyFactoryHint()); setParams(info.getParams()); diff --git a/src/main/java/com/doodream/rmovjs/model/Request.java b/src/main/java/com/doodream/rmovjs/model/Request.java index 1233860..03cac34 100644 --- a/src/main/java/com/doodream/rmovjs/model/Request.java +++ b/src/main/java/com/doodream/rmovjs/model/Request.java @@ -2,14 +2,13 @@ import com.doodream.rmovjs.net.ClientSocketAdapter; -import com.doodream.rmovjs.net.session.BlobSession; -import com.doodream.rmovjs.net.session.SessionControlMessage; -import com.doodream.rmovjs.net.session.SessionControlMessageWriter; +import com.doodream.rmovjs.net.session.*; import com.doodream.rmovjs.parameter.Param; -import com.doodream.rmovjs.serde.Converter; import com.doodream.rmovjs.serde.Writer; import com.google.gson.annotations.SerializedName; import io.reactivex.Observable; +import io.reactivex.functions.BiConsumer; +import io.reactivex.functions.BiFunction; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -17,10 +16,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.Optional; /** * Request contains information for client method invocation consisted with below @@ -61,11 +60,14 @@ public static boolean isValid(Request request) { (request.getParams() != null); } - public static SessionControlMessageWriter buildSessionMessageWriter(Writer writer) { - return (controlMessage) -> { - writer.write(Request.builder() - .scm(controlMessage) - .build()); + public static SessionControlMessageWriter buildSessionMessageWriter(final Writer writer) { + return new SessionControlMessageWriter() { + @Override + public void write(SessionControlMessage controlMessage) throws IOException { + writer.write(Request.builder() + .scm(controlMessage) + .build()); + } }; } @@ -80,29 +82,40 @@ public static Request fromEndpoint(Endpoint endpoint, Object ...args) { .endpoint(endpoint.getUnique()) .build(); } else { - Optional optionalSession = BlobSession.findOne(args); - Request.RequestBuilder builder = Request.builder() + BlobSession session = BlobSession.findOne(args); + final Request.RequestBuilder builder = Request.builder() .params(convertParams(endpoint, args)) .endpoint(endpoint.getUnique()); - optionalSession.ifPresent(builder::session); + + if(!session.equals(BlobSession.NULL)) { + builder.session(session); + } + return builder.build(); } } - private static List convertParams(Endpoint endpoint, Object[] objects) { + private static List convertParams(final Endpoint endpoint, Object[] objects) { if(objects == null) { return Collections.EMPTY_LIST; } - return Observable.fromIterable(endpoint.getParams()).zipWith(Observable.fromArray(objects), (param, o) -> { - param.apply(o); - if(param.isInstanceOf(BlobSession.class)) { - Log.debug("Endpoint {} has session param : {}", endpoint, param); - if(o != null) { - endpoint.session = (BlobSession) o; + return Observable.fromIterable(endpoint.getParams()).zipWith(Observable.fromArray(objects), new BiFunction() { + @Override + public Param apply(Param param, Object o) throws Exception { + param.apply(o); + if(param.isInstanceOf(BlobSession.class)) { + if(o != null) { + endpoint.session = (BlobSession) o; + } } + return param; + } + }).collectInto(new ArrayList(), new BiConsumer, Param>() { + @Override + public void accept(ArrayList params, Param param) throws Exception { + params.add(param); } - return param; - }).collectInto(new ArrayList(), List::add).blockingGet(); + }).blockingGet(); } } diff --git a/src/main/java/com/doodream/rmovjs/model/Response.java b/src/main/java/com/doodream/rmovjs/model/Response.java index fd564ad..08cc04c 100644 --- a/src/main/java/com/doodream/rmovjs/model/Response.java +++ b/src/main/java/com/doodream/rmovjs/model/Response.java @@ -8,6 +8,7 @@ import com.doodream.rmovjs.serde.Converter; import com.doodream.rmovjs.serde.Writer; import com.doodream.rmovjs.serde.json.JsonConverter; +import com.doodream.rmovjs.util.Types; import com.google.common.base.Preconditions; import com.google.gson.annotations.SerializedName; import lombok.AllArgsConstructor; @@ -17,7 +18,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; +import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; +import java.util.Collection; /** * Created by innocentevil on 18. 5. 4. @@ -34,7 +38,6 @@ public class Response { private T body; private boolean isSuccessful; private boolean hasSessionSwitch; - private ResponseBody errorBody; private int code; private int nonce; @SerializedName("scm") @@ -44,19 +47,11 @@ public static Response success(T body) { return Response.builder() .body(body) .code(SUCCESS) + .hasSessionSwitch(body instanceof BlobSession) .isSuccessful(true) .build(); } - public static Response success(BlobSession session) { - return Response.builder() - .body(session) - .code(SUCCESS) - .isSuccessful(true) - .hasSessionSwitch(true) - .build(); - } - public T getBody() { return body; } @@ -66,10 +61,10 @@ public void setBody(T body) { } public static Response error(int code, String msg) { - return Response.builder() - .code(code) + return Response.builder() .isSuccessful(false) - .errorBody(new ResponseBody(msg)) + .code(code) + .body(msg) .build(); } @@ -102,19 +97,18 @@ public static Response from(RMIError error) { } public static void validate(Response res) { - if(res.code == Response.SUCCESS) { - Preconditions.checkNotNull(res.getBody(), "Successful response must have non-null body"); - } else { - Preconditions.checkNotNull(res.getErrorBody(), "Error response must have non-null error body"); + Preconditions.checkNotNull(res.getBody(), "Successful response must have non-null body"); + if(!res.isSuccessful) { + Preconditions.checkArgument(res.getBody() instanceof String, "Error response must have non-null error body"); } } - public static SessionControlMessageWriter buildSessionMessageWriter(Writer writer) { - return (controlMessage) -> { - writer.write(Response.builder() - .scm(controlMessage) - .build()); - + public static SessionControlMessageWriter buildSessionMessageWriter(final Writer writer) { + return new SessionControlMessageWriter() { + @Override + public void write(SessionControlMessage controlMessage) throws IOException { + writer.write(Response.builder().scm(controlMessage).build()); + } }; } @@ -134,7 +128,21 @@ public boolean hasScm() { * @param converter converter implementation * @param type {@link Type} for body content */ - public void resolve(Converter converter, Type type) { - setBody(converter.resolve(getBody(), type)); + public void resolve(Converter converter, Type type) throws IllegalAccessException, InstantiationException, ClassNotFoundException { + if(type instanceof ParameterizedType) { + Class rawCls = Class.forName(((ParameterizedType) type).getRawType().getTypeName()); + if(Types.isCastable(body, rawCls)) { + // ex > Bson4Jackson parsed as collections like ArrayList + // however, if there is recursive type parameters like ArrayList> + return; + } + } else { + Class rawCls = Class.forName(type.getTypeName()); + if(Types.isCastable(body, rawCls)) { + return; + } + } + setBody((T) converter.resolve(getBody(), type)); } + } diff --git a/src/main/java/com/doodream/rmovjs/net/BaseServiceAdapter.java b/src/main/java/com/doodream/rmovjs/net/BaseServiceAdapter.java index bb2e104..18ff32f 100644 --- a/src/main/java/com/doodream/rmovjs/net/BaseServiceAdapter.java +++ b/src/main/java/com/doodream/rmovjs/net/BaseServiceAdapter.java @@ -7,15 +7,23 @@ import com.doodream.rmovjs.serde.Converter; import com.google.common.base.Preconditions; import io.reactivex.Observable; +import io.reactivex.ObservableEmitter; +import io.reactivex.ObservableOnSubscribe; +import io.reactivex.ObservableSource; import io.reactivex.disposables.CompositeDisposable; -import io.reactivex.functions.Function; +import io.reactivex.functions.*; +import io.reactivex.observables.GroupedObservable; import io.reactivex.schedulers.Schedulers; import lombok.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; -import java.util.Optional; +import java.net.InetAddress; +import java.net.InterfaceAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; public abstract class BaseServiceAdapter implements ServiceAdapter { @@ -24,58 +32,124 @@ public abstract class BaseServiceAdapter implements ServiceAdapter { private volatile boolean listen = false; @Override - public String listen(RMIServiceInfo serviceInfo, Converter converter, @NonNull Function handleRequest) throws IllegalAccessException, InstantiationException, IOException { + public String listen(final RMIServiceInfo serviceInfo, final Converter converter, final InetAddress network, @NonNull final Function handleRequest) throws IllegalAccessException, InstantiationException, IOException { if(listen) { throw new IllegalStateException("service already listening"); } final RMINegotiator negotiator = (RMINegotiator) serviceInfo.getNegotiator().newInstance(); Preconditions.checkNotNull(negotiator, "fail to resolve %s", serviceInfo.getNegotiator()); Preconditions.checkNotNull(converter, "fail to resolve %s", serviceInfo.getConverter()); - onStart(); + Preconditions.checkNotNull(network, "no network interface given"); + + onStart(network); listen = true; compositeDisposable.add(Observable.just(converter) - .map(c -> acceptClient()) - .doOnNext(socket -> Log.debug("{} connected", socket.getRemoteName())) - .repeatUntil(() -> !listen) - .map(client -> negotiator.handshake(client, serviceInfo, converter, false)) - .map(socket -> new ClientSocketAdapter(socket, converter)) + .map(new Function() { + @Override + public RMISocket apply(Converter converter) throws Exception { + return acceptClient(); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(RMISocket socket) throws Exception { + Log.debug("{} connected", socket.getRemoteName()); + } + }) + .repeatUntil(new BooleanSupplier() { + @Override + public boolean getAsBoolean() throws Exception { + return !listen; + } + }) + .map(new Function() { + @Override + public RMISocket apply(RMISocket rmiSocket) throws Exception { + return negotiator.handshake(rmiSocket, serviceInfo, converter, false); + } + }) + .map(new Function() { + @Override + public ClientSocketAdapter apply(RMISocket rmiSocket) throws Exception { + return new ClientSocketAdapter(rmiSocket, converter); + } + }) .subscribeOn(Schedulers.newThread()) - .subscribe(adapter-> onHandshakeSuccess(adapter, handleRequest),this::onError)); + .subscribe(new Consumer() { + @Override + public void accept(ClientSocketAdapter clientSocketAdapter) throws Exception { + onHandshakeSuccess(clientSocketAdapter, handleRequest); + } + }, new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + onError(throwable); + } + })); return getProxyConnectionHint(serviceInfo); } - private void onHandshakeSuccess(ClientSocketAdapter adapter, Function handleRequest) { - + private void onHandshakeSuccess(final ClientSocketAdapter adapter, final Function handleRequest) { compositeDisposable.add(adapter.listen() - .groupBy(Request::isValid) - .flatMap(booleanRequestGroupedObservable -> Observable.>create(emitter -> { - if(booleanRequestGroupedObservable.getKey()) { - emitter.setDisposable(booleanRequestGroupedObservable.subscribe(request -> emitter.onNext(Optional.of(request)))); - } else { - // bad request handle added - emitter.setDisposable(booleanRequestGroupedObservable.subscribe(request -> { - adapter.write(Response.from(RMIError.BAD_REQUEST)); - emitter.onNext(Optional.empty()); - })); + .groupBy(new Function() { + @Override + public Boolean apply(Request request) throws Exception { + return Request.isValid(request); } - })) - .filter(Optional::isPresent) - .map(Optional::get) - .doOnNext(request -> request.setClient(adapter)) - .doOnNext(request -> Log.trace("Request <= {}", request)) + }) + .flatMap(new Function, ObservableSource>() { + @Override + public ObservableSource apply(final GroupedObservable booleanRequestGroupedObservable) throws Exception { + return Observable.create(new ObservableOnSubscribe() { + @Override + public void subscribe(final ObservableEmitter emitter) throws Exception { + if(booleanRequestGroupedObservable.getKey()) { + emitter.setDisposable(booleanRequestGroupedObservable.subscribe(new Consumer() { + @Override + public void accept(Request request) throws Exception { + emitter.onNext(request); + } + })); + } else { + // bad request handle added + emitter.setDisposable(booleanRequestGroupedObservable.subscribe(new Consumer() { + @Override + public void accept(Request request) throws Exception { + adapter.write(Response.from(RMIError.BAD_REQUEST)); + } + })); + } + } + }); + } + }) .observeOn(Schedulers.io()) - .subscribe(request -> { - final Response response = handleRequest.apply(request); - Log.trace("Response => {}", response); - adapter.write(response); - },this::onError)); + .subscribe(new Consumer() { + @Override + public void accept(Request request) throws Exception { + if(Log.isTraceEnabled()) { + Log.trace("Request <= {}", request); + } + request.setClient(adapter); + final Response response = handleRequest.apply(request); + if(Log.isTraceEnabled()) { + Log.trace("Response => {}", response); + } + adapter.write(response); + } + }, new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + onError(throwable); + } + })); } private void onError(Throwable throwable) { - Log.error("Error : {}", throwable); + Log.error("Error : ", throwable); close(); } @@ -86,7 +160,7 @@ public void close() { try { onClose(); } catch (IOException e) { - Log.warn("{}", e); + Log.warn("", e); } } compositeDisposable.dispose(); @@ -94,7 +168,7 @@ public void close() { } - protected abstract void onStart() throws IOException; + protected abstract void onStart(InetAddress bindAddress) throws IOException; protected abstract boolean isClosed(); protected abstract String getProxyConnectionHint(RMIServiceInfo serviceInfo); protected abstract RMISocket acceptClient() throws IOException; diff --git a/src/main/java/com/doodream/rmovjs/net/BaseServiceProxy.java b/src/main/java/com/doodream/rmovjs/net/BaseServiceProxy.java index 8027ec4..ce395f5 100644 --- a/src/main/java/com/doodream/rmovjs/net/BaseServiceProxy.java +++ b/src/main/java/com/doodream/rmovjs/net/BaseServiceProxy.java @@ -12,19 +12,24 @@ import com.doodream.rmovjs.server.BasicService; import com.doodream.rmovjs.server.svc.HealthCheckController; import io.reactivex.Observable; +import io.reactivex.ObservableEmitter; +import io.reactivex.ObservableOnSubscribe; import io.reactivex.Scheduler; import io.reactivex.disposables.CompositeDisposable; +import io.reactivex.disposables.Disposable; +import io.reactivex.functions.BiFunction; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; import io.reactivex.schedulers.Schedulers; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.lang.reflect.Method; -import java.nio.charset.StandardCharsets; import java.util.Base64; -import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -51,12 +56,15 @@ public class BaseServiceProxy implements RMIServiceProxy { Log.debug("healthCheck {}", HEALTH_CHECK_ENDPOINT); } } - private AtomicInteger semaphore; + private AtomicInteger openSemaphore; + private AtomicInteger pingSemaphore; + private long measuredQos; private volatile boolean isOpened; private final ConcurrentHashMap sessionRegistry; private ConcurrentHashMap requestWaitQueue; private CompositeDisposable compositeDisposable; - private int requestNonce; + private Disposable pingDisposable; + private AtomicInteger requestId; private RMIServiceInfo serviceInfo; private Converter converter; private RMISocket socket; @@ -64,7 +72,6 @@ public class BaseServiceProxy implements RMIServiceProxy { private Writer writer; private Scheduler mListener = Schedulers.from(Executors.newWorkStealingPool(10)); - public static BaseServiceProxy create(RMIServiceInfo info, RMISocket socket) { return new BaseServiceProxy(info, socket); } @@ -73,10 +80,13 @@ private BaseServiceProxy(RMIServiceInfo info, RMISocket socket) { sessionRegistry = new ConcurrentHashMap<>(); requestWaitQueue = new ConcurrentHashMap<>(); compositeDisposable = new CompositeDisposable(); + pingDisposable = null; + // set Qos as bad as possible + measuredQos = Long.MAX_VALUE; - semaphore = new AtomicInteger(); - semaphore.getAndSet(0); - requestNonce = 0; + openSemaphore = new AtomicInteger(0); + pingSemaphore = new AtomicInteger(0); + requestId = new AtomicInteger(0); serviceInfo = info; isOpened = false; this.socket = socket; @@ -84,43 +94,63 @@ private BaseServiceProxy(RMIServiceInfo info, RMISocket socket) { @Override public synchronized void open() throws IOException, IllegalAccessException, InstantiationException { - if(semaphore.getAndAdd(1) != 0) { + if(!markAsUse(openSemaphore)) { + Log.debug("already opened"); return; } - Log.debug("Initialized"); + RMINegotiator negotiator = (RMINegotiator) serviceInfo.getNegotiator().newInstance(); converter = (Converter) serviceInfo.getConverter().newInstance(); + socket.open(); reader = converter.reader(socket.getInputStream()); writer = converter.writer(socket.getOutputStream()); socket = negotiator.handshake(socket, serviceInfo, converter, true); + Log.trace("open proxy for {} : success", serviceInfo.getName()); isOpened = true; - compositeDisposable.add(Observable.create(emitter -> { - while (isOpened) { - Response response = reader.read(Response.class); - if(response == null) { + compositeDisposable.add(Observable.create(new ObservableOnSubscribe() { + @Override + public void subscribe(ObservableEmitter emitter) throws Exception { + try { + while (isOpened) { + Response response = reader.read(Response.class); + if (response == null) { + return; + } + if (response.hasScm()) { + handleSessionControlMessage(response); + continue; + } + emitter.onNext(response); + } + } catch (IOException ignore) { + isOpened = false; + } finally { + emitter.onComplete(); + } + } + }).subscribeOn(mListener).subscribe(new Consumer() { + @Override + public void accept(Response response) throws Exception { + Request request = requestWaitQueue.remove(response.getNonce()); + if (request == null) { + Log.warn("no mapped request exists : {}", response); return; } - if(response.hasScm()) { - handleSessionControlMessage(response); - continue; + synchronized (request) { + Log.debug("request({}) is response({})", request, response); + request.setResponse(response); + request.notifyAll(); // wakeup waiting thread } - emitter.onNext(response); } - emitter.onComplete(); - }).subscribeOn(mListener).subscribe(response -> { - Request request = requestWaitQueue.remove(response.getNonce()); - if(request == null) { - Log.warn("no mapped request exists : {}", response); - return; + }, new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + onError(throwable); } - synchronized (request) { - request.setResponse(response); - request.notifyAll(); // wakeup waiting thread - } - }, this::onError)); + })); } @Override @@ -129,66 +159,107 @@ public boolean isOpen() { } @Override - public Response request(Endpoint endpoint, Object ...args) { + public Response request(Endpoint endpoint, long timeoutInMillisec, Object ...args) { return Observable.just(Request.fromEndpoint(endpoint, args)) - .doOnNext(request -> request.setNonce(++requestNonce)) - .doOnNext(request -> { - final BlobSession session = request.getSession(); - if(session != null) { - registerSession(session); + .doOnNext(new Consumer() { + @Override + public void accept(Request request) throws Exception { + request.setNonce(requestId.incrementAndGet()); + final BlobSession session = request.getSession(); + if(session != null) { + registerSession(session); + } + if(Log.isTraceEnabled()) { + Log.trace("Request => {}", request); + } } }) - .doOnNext(request -> Log.trace("Request => {}", request)) - .map(request -> { - requestWaitQueue.put(request.getNonce(), request); - synchronized (request) { - writer.write(request); - // caller block here, until the response is ready - request.wait(); + .map(new Function() { + @Override + public Response apply(Request request) throws Exception { + requestWaitQueue.put(request.getNonce(), request); + try { + synchronized (request) { + writer.write(request); + if (timeoutInMillisec > 0) { + request.wait(timeoutInMillisec); + } else { + request.wait(); + } + } + } catch (InterruptedException e) { + return RMIError.CLOSED.getResponse(); + } + if (request.getResponse() == null) { + return RMIError.TIMEOUT.getResponse(); + } + return request.getResponse(); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Response response) throws Exception { + if (response.isSuccessful()) { + response.resolve(converter, endpoint.getUnwrappedRetType()); + } } - return Optional.of(request.getResponse()); }) - .filter(Optional::isPresent) - .map(Optional::get) - .defaultIfEmpty(RMIError.UNHANDLED.getResponse()) - .doOnError(this::onError) - .doOnNext(response -> response.resolve(converter, endpoint.getUnwrappedRetType())) - .doOnNext(response -> { - Log.trace("Response <= {}", response); - if(response.isHasSessionSwitch() && - response.isSuccessful()) { - final BlobSession session = (BlobSession) response.getBody(); - if(session != null) { - session.init(); - if (sessionRegistry.put(session.getKey(), session) != null) { - Log.warn("session conflict for {}", session.getKey()); - return; - } - session.start(reader, writer, Request::buildSessionMessageWriter, () -> unregisterSession(session)); + .map(new Function() { + @Override + public Response apply(Response response) throws Exception { + if (response.isHasSessionSwitch()) { + return handleBlobResponse(response); + } else { + return response; } } }) + .defaultIfEmpty(RMIError.UNHANDLED.getResponse()) + .doOnError(new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + onError(throwable); + } + }) .blockingSingle(); } - private void handleSessionControlMessage(Response response) throws IOException { + private Response handleBlobResponse(Response response) { + if(response.getBody() != null) { + final BlobSession session = (BlobSession) response.getBody(); + session.init(); + if(sessionRegistry.put(session.getKey(), session) != null) { + Log.warn("session conflict for {}", session.getKey()); + } else { + session.start(reader, writer, converter, Request::buildSessionMessageWriter, () -> unregisterSession(session)); + } + return response; + } + return RMIError.BAD_RESPONSE.getResponse(); + } + + private void handleSessionControlMessage(Response response) throws IOException, IllegalAccessException, InstantiationException, ClassNotFoundException { SessionControlMessage scm = response.getScm(); BlobSession session; session = sessionRegistry.get(scm.getKey()); if (session == null) { - Log.warn("Session not available for {}", scm); + Log.warn("session not available for {} @ {}", scm.getCommand(), scm.getKey()); return; } session.handle(scm); } + /** + * register session + * @param session + */ private void registerSession(BlobSession session) { if (sessionRegistry.put(session.getKey(), session) != null) { Log.warn("session : {} collision in registry", session.getKey()); } Log.trace("session registered {}", session); - session.start(reader, writer, Request::buildSessionMessageWriter, () -> unregisterSession(session)); + session.start(reader, writer, converter, Request::buildSessionMessageWriter, () -> unregisterSession(session)); } private void unregisterSession(BlobSession session ) { @@ -200,43 +271,117 @@ private void unregisterSession(BlobSession session ) { } private void onError(Throwable throwable) { - Log.error("{}", throwable); + Log.error("proxy closed {}({})", socket.getRemoteName() ,serviceInfo.getName(), throwable); try { - close(); + actualClose(); } catch (IOException ignored) { } } public void close() throws IOException { - if(semaphore.getAndDecrement() != 0) { + if(!markAsUnuse(openSemaphore)) { return; } - if(!socket.isClosed()) { - socket.close(); - } + actualClose(); + } + + /** + * close socket for this service proxy and stop listening to response from the remote service + * @throws IOException try to close socket when it is already closed by peer or has not been opened at all + */ + private synchronized void actualClose() throws IOException { if(!compositeDisposable.isDisposed()) { compositeDisposable.dispose(); } compositeDisposable.clear(); - // wake blocked thread from wait queue - requestWaitQueue.values().forEach(request -> { + if(!pingDisposable.isDisposed()) { + pingDisposable.dispose(); + } + if(!socket.isClosed()) { + socket.close(); + } + for (Request request : requestWaitQueue.values()) { synchronized (request) { + // put error response on the request + request.setResponse(RMIError.CLOSED.getResponse()); + // wake blocked threads from wait queue request.notifyAll(); } - }); + } isOpened = false; Log.debug("proxy for {} closed", serviceInfo.getName()); } + @Override - public Optional ping() { - final Response response = request(HEALTH_CHECK_ENDPOINT); - if(response.isSuccessful() && (response.getCode() == Response.SUCCESS)) { - Log.debug("body {} /w cls {}", response.getBody(), response.getBody().getClass()); - return Optional.of(response.getBody()); + public void startPeriodicQosUpdate(long timeout, long interval, TimeUnit timeUnit) { + if(!markAsUse(pingSemaphore)) { + return; } - return Optional.empty(); + pingDisposable = Observable.interval(interval, timeUnit) + .subscribeOn(Schedulers.io()) + .subscribe(new Consumer() { + @Override + public void accept(Long aLong) throws Exception { + getQosUpdate(timeout); + } + }); + } + + @Override + public void stopPeriodicQosUpdate() { + if(!markAsUnuse(pingSemaphore)) { + return; + } + if(!pingDisposable.isDisposed()) { + pingDisposable.dispose(); + } + } + + + /** + * check whether the resource has been used previously or not + * @param semaphore {@link AtomicInteger} to be used as semaphore for resource + * @return true, if the resource has been unused previously, otherwise false + */ + private boolean markAsUse(AtomicInteger semaphore) { + return !(semaphore.getAndIncrement() > 0); + } + + /** + * check whether the resource is still used or not + * @param semaphore {@link AtomicInteger} to be used as semaphore for resource + * @return true, resource becomes unused, otherwise false + */ + private boolean markAsUnuse(AtomicInteger semaphore) { + return !(semaphore.updateAndGet(v -> { + if(v > 0) { + return --v; + } + return v; + }) > 0); } + @Override + public Long getQosUpdate(long timeout) { + if(!isOpen()) { + return Long.MAX_VALUE; + } + long sTime = System.currentTimeMillis(); + final Response response = request(HEALTH_CHECK_ENDPOINT, timeout); + if (response.isSuccessful() && (response.getCode() == Response.SUCCESS)) { + measuredQos = System.currentTimeMillis() - sTime; + } else { + measuredQos = Long.MAX_VALUE; + } + return measuredQos; + } + + @Override + public Long getQosMeasured() { + return measuredQos; + } + + @Override public String who() { return Base64.getEncoder().encodeToString(socket.getRemoteName().concat(serviceInfo.toString()).getBytes()); @@ -245,9 +390,24 @@ public String who() { @Override public boolean provide(Class controller) { return Observable.fromIterable(serviceInfo.getControllerInfos()) - .map(ControllerInfo::getStubCls) - .map(controller::equals) - .reduce((isThere1, isThere2) -> isThere1 || isThere2) + .map(new Function>() { + @Override + public Class apply(ControllerInfo controllerInfo) throws Exception { + return controllerInfo.getStubCls(); + } + }) + .map(new Function, Boolean>() { + @Override + public Boolean apply(Class stubCls) throws Exception { + return controller.equals(stubCls); + } + }) + .reduce(new BiFunction() { + @Override + public Boolean apply(Boolean match1, Boolean match2) throws Exception { + return match1 || match2; + } + }) .blockingGet(false); } } diff --git a/src/main/java/com/doodream/rmovjs/net/ClientSocketAdapter.java b/src/main/java/com/doodream/rmovjs/net/ClientSocketAdapter.java index ed313e6..ea94844 100644 --- a/src/main/java/com/doodream/rmovjs/net/ClientSocketAdapter.java +++ b/src/main/java/com/doodream/rmovjs/net/ClientSocketAdapter.java @@ -25,6 +25,7 @@ public class ClientSocketAdapter { private RMISocket client; private Reader reader; private Writer writer; + private Converter converter; private final ConcurrentHashMap sessionRegistry; ClientSocketAdapter(RMISocket socket, Converter converter) throws IOException { @@ -32,14 +33,15 @@ public class ClientSocketAdapter { sessionRegistry = new ConcurrentHashMap<>(); reader = converter.reader(socket.getInputStream()); writer = converter.writer(socket.getOutputStream()); + this.converter = converter; } - public void write(Response response) throws IOException { + public void write(Response response) throws Exception { if(response.isHasSessionSwitch()) { BlobSession session = (BlobSession) response.getBody(); sessionRegistry.put(session.getKey(), session); - session.start(reader, writer, Response::buildSessionMessageWriter, () -> unregisterSession(session)); + session.start(reader, writer, converter, Response::buildSessionMessageWriter, () -> unregisterSession(session)); } writer.write(response); } @@ -68,14 +70,14 @@ Observable listen() { return; } Log.debug("session registered {}", session); - session.start(reader, writer, Response::buildSessionMessageWriter, () -> unregisterSession(session)); + session.start(reader, writer, converter, Response::buildSessionMessageWriter, () -> unregisterSession(session)); // forward request to transfer session object to application } emitter.onNext(request); } emitter.onComplete(); } catch (IOException e) { - emitter.onError(e); + client.close(); } }); return requestObservable.subscribeOn(Schedulers.io()); @@ -89,7 +91,7 @@ private void unregisterSession(BlobSession session ) { Log.trace("remove session : {}", session.getKey()); } - private void handleSessionControlMessage(Request request) throws SessionControlException, IllegalStateException, IOException { + private void handleSessionControlMessage(Request request) throws SessionControlException, IllegalStateException, IOException, IllegalAccessException, InstantiationException, ClassNotFoundException { final SessionControlMessage scm = request.getScm(); BlobSession session; session = sessionRegistry.get(scm.getKey()); @@ -104,4 +106,8 @@ String who() { return client.getRemoteName(); } + public void close() throws IOException { + client.close(); + Log.debug("client closed"); + } } diff --git a/src/main/java/com/doodream/rmovjs/net/RMIServiceProxy.java b/src/main/java/com/doodream/rmovjs/net/RMIServiceProxy.java index ef2cfa5..2708be3 100644 --- a/src/main/java/com/doodream/rmovjs/net/RMIServiceProxy.java +++ b/src/main/java/com/doodream/rmovjs/net/RMIServiceProxy.java @@ -3,17 +3,15 @@ import com.doodream.rmovjs.model.Endpoint; import com.doodream.rmovjs.model.RMIError; import com.doodream.rmovjs.model.Response; -import jdk.nashorn.internal.runtime.options.Option; import java.io.IOException; -import java.util.Optional; - -import static java.util.Optional.empty; +import java.util.concurrent.TimeUnit; public interface RMIServiceProxy { RMIServiceProxy NULL_PROXY = new RMIServiceProxy() { @Override public void open() { + // NO OP } @Override @@ -22,17 +20,33 @@ public boolean isOpen() { } @Override - public Response request(Endpoint endpoint, Object ...args) { + public Response request(Endpoint endpoint, long timeoutInMilliSec, Object ...args) { return Response.from(RMIError.NOT_FOUND); } @Override public void close() { + // NO OP + } + + @Override + public void startPeriodicQosUpdate(long timeout, long interval, TimeUnit timeUnit) { + // NO OP + } + + @Override + public void stopPeriodicQosUpdate() { + // NO OP + } + + @Override + public Long getQosUpdate(long timeout) { + return Long.MAX_VALUE; } @Override - public Optional ping() { - return Optional.empty(); + public Long getQosMeasured() { + return Long.MAX_VALUE; } @Override @@ -55,9 +69,12 @@ public boolean provide(Class controller) { */ void open() throws IOException, IllegalAccessException, InstantiationException; boolean isOpen(); - Response request(Endpoint endpoint, Object ...args) throws IOException; + Response request(Endpoint endpoint, long timeoutMilliSec, Object ...args) throws IOException; void close() throws IOException; - Optional ping(); + void startPeriodicQosUpdate(long timeout, long interval, TimeUnit timeUnit); + void stopPeriodicQosUpdate(); + Long getQosUpdate(long timeout); + Long getQosMeasured(); String who(); boolean provide(Class controller); } diff --git a/src/main/java/com/doodream/rmovjs/net/ServiceAdapter.java b/src/main/java/com/doodream/rmovjs/net/ServiceAdapter.java index 94f4612..abc76d2 100644 --- a/src/main/java/com/doodream/rmovjs/net/ServiceAdapter.java +++ b/src/main/java/com/doodream/rmovjs/net/ServiceAdapter.java @@ -8,6 +8,7 @@ import io.reactivex.functions.Function; import java.io.IOException; +import java.net.InetAddress; /** * {@link ServiceAdapter} provides abstraction layer for network dependency including listed below @@ -29,13 +30,14 @@ public interface ServiceAdapter { * client로 부터의 네트워크 연결을 대기하며 client로 부터의 {@link Request}를 처리하기 위한 handler를 등록한다. * @param serviceInfo 서비스 정의 instance {@link RMIServiceInfo} * @param converter {@link Request} 및 {@link Response} instance에 대한 deserialization / serialization module + * @param network network interface which the service adapter listen to * @param requestHandler {@link Request}의 수신 및 {@link Response}의 응답을 처리하기 위한 handler로 {@link com.doodream.rmovjs.server.RMIService} * @return proxyFactoryHint as string * @throws IOException server 측 네트워크 endpoint 생성의 실패 혹은 I/O 오류 * @throws IllegalAccessError the error thrown when {@link ServiceAdapter} fails to resolve dependency object (e.g. negotiator, * @throws InstantiationException if dependent class represents an abstract class,an interface, an array class, a primitive type, or void;or if the class has no nullary constructor; */ - String listen(RMIServiceInfo serviceInfo, Converter converter, Function requestHandler) throws IOException, IllegalAccessException, InstantiationException; + String listen(RMIServiceInfo serviceInfo, Converter converter, InetAddress network, Function requestHandler) throws IOException, IllegalAccessException, InstantiationException; /** * return {@link ServiceProxyFactory} which is capable of building {@link RMIServiceProxy} able to connect to current service adapter diff --git a/src/main/java/com/doodream/rmovjs/net/SimpleNegotiator.java b/src/main/java/com/doodream/rmovjs/net/SimpleNegotiator.java index 44b9357..a56a2e0 100644 --- a/src/main/java/com/doodream/rmovjs/net/SimpleNegotiator.java +++ b/src/main/java/com/doodream/rmovjs/net/SimpleNegotiator.java @@ -8,6 +8,9 @@ import com.doodream.rmovjs.serde.Writer; import com.google.common.base.Preconditions; import io.reactivex.Observable; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,7 +21,7 @@ public class SimpleNegotiator implements RMINegotiator { @Override public RMISocket handshake(RMISocket socket, RMIServiceInfo service, Converter converter, boolean isClient) throws HandshakeFailException { - Log.info("Handshake start @ {}", isClient? "CLIENT" : "SERVER"); + Log.info("Handshake start as {} @ {}", isClient? "CLIENT" : "SERVER", socket.getRemoteName()); try { Reader reader = converter.reader(socket.getInputStream()); Writer writer = converter.writer(socket.getOutputStream()); @@ -33,43 +36,88 @@ public RMISocket handshake(RMISocket socket, RMIServiceInfo service, Converter c return socket; } - private void handshakeFromClient(RMIServiceInfo service, Reader reader, Writer writer) throws HandshakeFailException { + private void handshakeFromClient(final RMIServiceInfo service, Reader reader, Writer writer) throws HandshakeFailException { try { writer.write(service); + Log.debug("write {}", service); Response response = reader.read(Response.class); if ((response != null) && response.isSuccessful()) { - Log.info("Handshake Success {} (Ver. {})", service.getName(), service.getVersion()); + Log.debug("Handshake Success {} (Ver. {})", service.getName(), service.getVersion()); return; } Preconditions.checkNotNull(response, "Response is null"); - Log.error("Handshake Fail ({}) {}",response.getCode(), response.getErrorBody()); + Log.error("Handshake Fail ({}) {}",response.getCode(), response.getBody()); } catch (IOException ignore) { } throw new HandshakeFailException(); } - private void handshakeFromServer(RMIServiceInfo service, Reader reader, Writer writer) throws HandshakeFailException { + private void handshakeFromServer(final RMIServiceInfo service, Reader reader, final Writer writer) throws HandshakeFailException { try { Observable handshakeRequestSingle = Observable.just(reader.read(RMIServiceInfo.class)); Observable serviceInfoMatchedObservable = handshakeRequestSingle - .filter(info -> info.hashCode() == service.hashCode()) - .map(info -> Response.success("OK")); + .filter(new Predicate() { + @Override + public boolean test(RMIServiceInfo info) throws Exception { + return info.hashCode() == service.hashCode(); + } + }) + .map(new Function() { + @Override + public Response apply(RMIServiceInfo rmiServiceInfo) throws Exception { + return Response.success("OK"); + } + }); Observable serviceInfoMismatchObservable = handshakeRequestSingle - .filter(info -> info.hashCode() != service.hashCode()) - .map(info -> Response.from(RMIError.BAD_REQUEST)); + .filter(new Predicate() { + @Override + public boolean test(RMIServiceInfo info) throws Exception { + return info.hashCode() != service.hashCode(); + } + }) + .map(new Function() { + @Override + public Response apply(RMIServiceInfo rmiServiceInfo) throws Exception { + Log.debug("{} != {}", rmiServiceInfo, service); + return Response.from(RMIError.BAD_REQUEST); + } + }); boolean success = serviceInfoMatchedObservable.mergeWith(serviceInfoMismatchObservable) - .doOnNext(response -> Log.info("Handshake Response : ({}) {}", response.getCode(), response.getBody())) - .doOnNext(writer::write) - .map(Response::isSuccessful) - .filter(Boolean::booleanValue) + .doOnNext(new Consumer() { + @Override + public void accept(Response response) throws Exception { + Log.trace("Handshake Response : ({}) {}", response.getCode(), response.getBody()); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Response response) throws Exception { + Log.debug("write response {}", response); + writer.write(response); + } + }) + .map(new Function() { + @Override + public Boolean apply(Response response) throws Exception { + return response.isSuccessful(); + } + }) + .filter(new Predicate() { + @Override + public boolean test(Boolean aBoolean) throws Exception { + return aBoolean; + } + }) .blockingSingle(false); if (success) { return; } - } catch (IOException ignore) { } + } catch (IOException e) { + Log.error("", e); + } throw new HandshakeFailException(); } diff --git a/src/main/java/com/doodream/rmovjs/net/session/BlobSession.java b/src/main/java/com/doodream/rmovjs/net/session/BlobSession.java index 5269916..55f0dd9 100644 --- a/src/main/java/com/doodream/rmovjs/net/session/BlobSession.java +++ b/src/main/java/com/doodream/rmovjs/net/session/BlobSession.java @@ -1,17 +1,18 @@ package com.doodream.rmovjs.net.session; +import com.doodream.rmovjs.serde.Converter; import com.doodream.rmovjs.serde.Reader; import com.doodream.rmovjs.serde.Writer; import com.google.gson.annotations.SerializedName; import io.reactivex.Observable; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; import lombok.Data; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; -import java.util.Optional; import java.util.Random; -import java.util.function.Consumer; @Data public class BlobSession implements SessionHandler { @@ -29,6 +30,7 @@ public class BlobSession implements SessionHandler { // read error start with -2000 public static final int SIZE_NOT_MATCHED = -2001; public static final int INVALID_EOS_CHAR = -2002; + public static final BlobSession NULL = new BlobSession(null); private static int GLOBAL_KEY = 0; private static String DEFAULT_TYPE = "application/octet-stream"; @@ -69,18 +71,31 @@ public BlobSession() { * @param args arguments * @return */ - public static Optional findOne(Object[] args) { - return Observable.fromArray(args).filter(o -> o instanceof BlobSession).cast(BlobSession.class).map(Optional::ofNullable).blockingFirst(Optional.empty()); + public static BlobSession findOne(Object[] args) { + return Observable.fromArray(args) + .filter(new Predicate() { + @Override + public boolean test(Object o) throws Exception { + return o instanceof BlobSession; + } + }) + .cast(BlobSession.class) + .map(new Function() { + @Override + public BlobSession apply(BlobSession blobSession) throws Exception { + return blobSession; + } + }).blockingFirst(BlobSession.NULL); } @Override - public void handle(SessionControlMessage scm) throws SessionControlException, IOException { + public void handle(SessionControlMessage scm) throws SessionControlException, IOException, IllegalAccessException, ClassNotFoundException, InstantiationException { sessionHandler.handle(scm); } - public void start(Reader reader, Writer writer, SessionControlMessageWriter.Builder builder, Runnable onTeardown) { - sessionHandler.start(reader, writer, builder, onTeardown); + public void start(Reader reader, Writer writer, Converter converter, SessionControlMessageWriter.Builder builder, Runnable onTeardown) { + sessionHandler.start(reader, writer, converter, builder, onTeardown); } /** diff --git a/src/main/java/com/doodream/rmovjs/net/session/Consumer.java b/src/main/java/com/doodream/rmovjs/net/session/Consumer.java new file mode 100644 index 0000000..0b2f87e --- /dev/null +++ b/src/main/java/com/doodream/rmovjs/net/session/Consumer.java @@ -0,0 +1,5 @@ +package com.doodream.rmovjs.net.session; + +public interface Consumer { + void accept(T t); +} diff --git a/src/main/java/com/doodream/rmovjs/net/session/ReceiverSession.java b/src/main/java/com/doodream/rmovjs/net/session/ReceiverSession.java index ad41751..8e869c2 100644 --- a/src/main/java/com/doodream/rmovjs/net/session/ReceiverSession.java +++ b/src/main/java/com/doodream/rmovjs/net/session/ReceiverSession.java @@ -2,32 +2,63 @@ import com.doodream.rmovjs.net.session.param.SCMChunkParam; import com.doodream.rmovjs.net.session.param.SCMErrorParam; +import com.doodream.rmovjs.serde.Converter; import com.doodream.rmovjs.serde.Reader; import com.doodream.rmovjs.serde.Writer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.*; +import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.channels.Channels; -import java.nio.channels.WritableByteChannel; +import java.util.concurrent.ConcurrentLinkedQueue; public class ReceiverSession implements Session, SessionHandler { private static final Logger Log = LoggerFactory.getLogger(ReceiverSession.class); private String key; - private InputStream chunkInStream; - private WritableByteChannel chunkOutChannel; +// private InputStream chunkInStream; +// private OutputStream chunkOutStream; + private final ConcurrentLinkedQueue dataQueue = new ConcurrentLinkedQueue<>(); + private volatile boolean EOS; + private ByteBuffer cBuffer; + private Converter converter; private Runnable onTeardown; + private long overallRcvSize; private SessionControlMessageWriter scmWriter; ReceiverSession() { + overallRcvSize = 0L; } @Override - public int read(byte[] b, int offset, int len) throws IOException { - return chunkInStream.read(b, offset, len); + public int read(byte[] b, int offset, int len) { + if(cBuffer == null) { + byte[] data; + while((data = dataQueue.poll()) == null) { + if(EOS) { + return -1; + } + synchronized (dataQueue) { + try { + dataQueue.wait(100L); + } catch (InterruptedException e) { + return -1; + } + } + } + cBuffer = ByteBuffer.wrap(data); + } + final int rsz = cBuffer.remaining(); + if(rsz >= len) { + cBuffer.get(b, offset, len); + return len; + } else { + cBuffer.get(b, offset,rsz); + cBuffer = null; + int subRsz = read(b, offset + rsz, len - rsz); + return subRsz > 0 ? (rsz + subRsz) : rsz; + } } @Override @@ -37,10 +68,7 @@ public void write(byte[] b, int len) throws IOException { @Override public void open() throws IOException { - final PipedInputStream pipedInputStream = new PipedInputStream(); - chunkInStream = new BufferedInputStream(pipedInputStream, 64 * 1024); - chunkOutChannel = Channels.newChannel(new PipedOutputStream(pipedInputStream)); - + EOS = false; scmWriter.write(SessionControlMessage.builder() .command(SessionCommand.ACK) .key(key) @@ -62,24 +90,30 @@ void setSessionKey(String key) { } @Override - public void handle(SessionControlMessage scm) throws SessionControlException, IOException { - Log.debug("scm <= {} @ {}", scm.getCommand() ,scm.getKey()); + public void handle(SessionControlMessage scm) throws SessionControlException, IOException, IllegalAccessException, InstantiationException, ClassNotFoundException { final SessionCommand command = scm.getCommand(); + Object param = converter.resolve(scm.getParam(), command.getParamClass()); + if(Log.isTraceEnabled()) { + Log.trace("{} {}", command, param); + } switch (command) { case CHUNK: - SCMChunkParam chunkParam = (SCMChunkParam) scm.getParam(); - ByteBuffer buffer = ByteBuffer.wrap(chunkParam.getData()); - chunkOutChannel.write(buffer); - if(chunkParam.getType() == SCMChunkParam.TYPE_LAST) { - chunkOutChannel.close(); + SCMChunkParam chunkParam = (SCMChunkParam) param; + synchronized (dataQueue) { + dataQueue.offer(((SCMChunkParam) param).getData()); + dataQueue.notifyAll(); + if(chunkParam.getType() == SCMChunkParam.TYPE_LAST) { + EOS = true; + } } + overallRcvSize += chunkParam.getSizeInBytes(); break; case RESET: Log.debug("reset from peer"); onClose(); break; case ERR: - SCMErrorParam errorParam = (SCMErrorParam) scm.getParam(); + SCMErrorParam errorParam = (SCMErrorParam) param; throw new SessionControlException(errorParam); default: sendErrorMessage(key, SCMErrorParam.buildUnsupportedOperation(command)); @@ -87,16 +121,18 @@ public void handle(SessionControlMessage scm) throws SessionControlException, IO } @Override - public void start(Reader reader, Writer writer, SessionControlMessageWriter.Builder builder, Runnable onTeardown) { + public void start(Reader reader, Writer writer, Converter converter, SessionControlMessageWriter.Builder builder, Runnable onTeardown) { this.onTeardown = onTeardown; scmWriter = builder.build(writer); + this.converter = converter; } private void onClose() throws IOException { if(onTeardown != null) { onTeardown.run(); } - chunkInStream.close(); + EOS = true; + Log.debug("overall rcv size {}", overallRcvSize); } private void sendErrorMessage(String key, SCMErrorParam errorParam) throws IOException { diff --git a/src/main/java/com/doodream/rmovjs/net/session/SenderSession.java b/src/main/java/com/doodream/rmovjs/net/session/SenderSession.java index 3100ce0..6e23e03 100644 --- a/src/main/java/com/doodream/rmovjs/net/session/SenderSession.java +++ b/src/main/java/com/doodream/rmovjs/net/session/SenderSession.java @@ -2,6 +2,7 @@ import com.doodream.rmovjs.net.session.param.SCMChunkParam; import com.doodream.rmovjs.net.session.param.SCMErrorParam; +import com.doodream.rmovjs.serde.Converter; import com.doodream.rmovjs.serde.Reader; import com.doodream.rmovjs.serde.Writer; import org.slf4j.Logger; @@ -11,9 +12,7 @@ import java.nio.ByteBuffer; import java.util.LinkedHashMap; import java.util.Map; -import java.util.OptionalLong; import java.util.Random; -import java.util.function.Consumer; public class SenderSession implements Session, SessionHandler { @@ -26,6 +25,7 @@ public class SenderSession implements Session, SessionHandler { private String key; private Consumer onReady; private Runnable onTeardown; + private Converter converter; private byte[] bufferSource = new byte[BlobSession.CHUNK_MAX_SIZE_IN_BYTE]; private ByteBuffer writeBuffer; private SessionControlMessageWriter scmWriter; @@ -35,19 +35,15 @@ public class SenderSession implements Session, SessionHandler { SenderSession(Consumer onReady) { // create unique key for session - OptionalLong lo = RANDOM.longs(4).reduce((left, right) -> left + right); - if(!lo.isPresent()) { - throw new IllegalStateException("fail to generate random"); - } - key = Integer.toHexString(String.format("%d%d%d",GLOBAL_KEY++, System.currentTimeMillis(), lo.getAsLong()).hashCode()); + key = String.format("%8x%8x%8x",GLOBAL_KEY++, System.currentTimeMillis(), RANDOM.nextLong()).trim(); chunkSeqNumber = 0; chunkLruCache = SenderSession.getLruCache(MAX_CNWD_SIZE * 2); writeBuffer = ByteBuffer.wrap(bufferSource); this.onReady = onReady; } - private static Map getLruCache(int size) { + private static Map getLruCache(final int size) { return new LinkedHashMap (size * 4/3, 0.75f, true) { @Override protected boolean removeEldestEntry(Map.Entry eldest) { @@ -104,9 +100,12 @@ public synchronized void write(byte[] b, int len) throws IOException { } @Override - public void handle(SessionControlMessage scm) throws IllegalStateException, IOException { + public void handle(SessionControlMessage scm) throws IllegalStateException, IOException, IllegalAccessException, InstantiationException, ClassNotFoundException { final SessionCommand command = scm.getCommand(); - Log.debug("scm <= {} @ {}" , command, scm.getKey()); + Object param = converter.resolve(scm.getParam(), command.getParamClass()); + if(Log.isTraceEnabled()) { + Log.trace("{} {}", command, param); + } switch (command) { case ACK: // ready-to-receive from receiver @@ -119,7 +118,7 @@ public void handle(SessionControlMessage scm) throws IllegalStateException, IOEx onReady.accept(this); break; case ERR: - SCMErrorParam errorParam = (SCMErrorParam) scm.getParam(); + SCMErrorParam errorParam = (SCMErrorParam) param; handleErrorMessage(errorParam); break; case RESET: @@ -136,9 +135,10 @@ public void handle(SessionControlMessage scm) throws IllegalStateException, IOEx } @Override - public void start(Reader reader, Writer writer, SessionControlMessageWriter.Builder builder, Runnable onTeardown) { + public void start(Reader reader, Writer writer, Converter converter, SessionControlMessageWriter.Builder builder, Runnable onTeardown) { this.scmWriter = builder.build(writer); this.onTeardown = onTeardown; + this.converter = converter; } private void handleErrorMessage(SCMErrorParam errorParam) { diff --git a/src/main/java/com/doodream/rmovjs/net/session/SessionHandler.java b/src/main/java/com/doodream/rmovjs/net/session/SessionHandler.java index 9297dce..79fbdcc 100644 --- a/src/main/java/com/doodream/rmovjs/net/session/SessionHandler.java +++ b/src/main/java/com/doodream/rmovjs/net/session/SessionHandler.java @@ -1,12 +1,13 @@ package com.doodream.rmovjs.net.session; +import com.doodream.rmovjs.serde.Converter; import com.doodream.rmovjs.serde.Reader; import com.doodream.rmovjs.serde.Writer; import java.io.IOException; public interface SessionHandler { - void handle(SessionControlMessage scm) throws SessionControlException, IOException; - void start(Reader reader, Writer writer, SessionControlMessageWriter.Builder builder, Runnable onTeardown); + void handle(SessionControlMessage scm) throws SessionControlException, IOException, IllegalAccessException, InstantiationException, ClassNotFoundException; + void start(Reader reader, Writer writer, Converter converter, SessionControlMessageWriter.Builder builder, Runnable onTeardown); } diff --git a/src/main/java/com/doodream/rmovjs/net/tcp/TcpRMISocket.java b/src/main/java/com/doodream/rmovjs/net/tcp/TcpRMISocket.java index d5c7cae..4ce513a 100644 --- a/src/main/java/com/doodream/rmovjs/net/tcp/TcpRMISocket.java +++ b/src/main/java/com/doodream/rmovjs/net/tcp/TcpRMISocket.java @@ -20,7 +20,8 @@ public TcpRMISocket(Socket client) { } public TcpRMISocket(String host, int port) { - remoteAddress = InetSocketAddress.createUnresolved(host, port); + + remoteAddress = new InetSocketAddress(host, port); socket = null; } diff --git a/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceAdapter.java b/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceAdapter.java index 7757d5a..0b5acf9 100644 --- a/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceAdapter.java +++ b/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceAdapter.java @@ -6,10 +6,14 @@ import com.doodream.rmovjs.net.RMISocket; import com.doodream.rmovjs.net.ServiceProxyFactory; import io.reactivex.Observable; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.lang.reflect.Constructor; import java.net.*; public class TcpServiceAdapter extends BaseServiceAdapter { @@ -17,48 +21,56 @@ public class TcpServiceAdapter extends BaseServiceAdapter { protected static final Logger Log = LoggerFactory.getLogger(TcpServiceAdapter.class); private ServerSocket serverSocket; - private InetSocketAddress mAddress; - public static final String DEFAULT_PORT = "6644"; + private int port; + static final int DEFAULT_PORT = 6644; - public TcpServiceAdapter(String host, String port) throws UnknownHostException { - Log.debug("ServiceAdapter On {}/{}", host, port); - int p = Integer.valueOf(port); - mAddress = new InetSocketAddress(InetAddress.getByName(host), p); + public TcpServiceAdapter() { + port = DEFAULT_PORT; } - public TcpServiceAdapter(String port) throws UnknownHostException { - this(Inet4Address.getLocalHost().getHostAddress(), port); - } - - public TcpServiceAdapter() throws UnknownHostException { - this(DEFAULT_PORT); + public TcpServiceAdapter(String port) { + this.port = Integer.valueOf(port); } @Override - public ServiceProxyFactory getProxyFactory(RMIServiceInfo info) { - if(!RMIServiceInfo.isComplete(info)) { + public ServiceProxyFactory getProxyFactory(final RMIServiceInfo info) { + if(!RMIServiceInfo.isValid(info)) { throw new IllegalArgumentException("Incomplete service info"); } - String[] params = info.getParams().toArray(new String[0]); + final String[] params = info.getParams().toArray(new String[0]); return Observable.fromArray(TcpServiceProxyFactory.class.getConstructors()) - .filter(constructor -> constructor.getParameterCount() == params.length) - .map(constructor -> constructor.newInstance(params)) + .filter(new Predicate>() { + @Override + public boolean test(Constructor constructor) throws Exception { + return constructor.getParameterCount() == params.length; + } + }) + .map(new Function, Object>() { + @Override + public Object apply(Constructor constructor) throws Exception { + return constructor.newInstance(params); + } + }) .cast(ServiceProxyFactory.class) - .doOnNext(serviceProxyFactory -> serviceProxyFactory.setTargetService(info)) + .doOnNext(new Consumer() { + @Override + public void accept(ServiceProxyFactory serviceProxyFactory) throws Exception { + serviceProxyFactory.setTargetService(info); + } + }) .blockingFirst(); } @Override - protected void onStart() throws IOException { + protected void onStart(InetAddress bindAddress) throws IOException { serverSocket = new ServerSocket(); - Log.debug("service address {}", mAddress.getAddress().getHostAddress()); - serverSocket.bind(mAddress); - Log.debug("service started @ {}", mAddress.getAddress().getHostAddress()); + serverSocket.bind(new InetSocketAddress(bindAddress.getHostAddress(), port)); + Log.debug("service started @ {} : {}", serverSocket.getLocalSocketAddress(), serverSocket.getInetAddress().getHostAddress()); } @Override protected void onClose() throws IOException { - Log.debug("close() @ {}", mAddress.getAddress()); + Log.debug("close() @ {}", serverSocket.getInetAddress().getAddress()); if(serverSocket != null && !serverSocket.isClosed()) { serverSocket.close(); @@ -72,7 +84,7 @@ protected boolean isClosed() { @Override protected String getProxyConnectionHint(RMIServiceInfo serviceInfo) { - return mAddress.getAddress().getHostAddress(); + return serverSocket.getInetAddress().getHostAddress(); } @Override diff --git a/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceProxyFactory.java b/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceProxyFactory.java index 96ca1aa..4c7ea23 100644 --- a/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceProxyFactory.java +++ b/src/main/java/com/doodream/rmovjs/net/tcp/TcpServiceProxyFactory.java @@ -20,7 +20,7 @@ public TcpServiceProxyFactory(String port) { } public TcpServiceProxyFactory() { - this(TcpServiceAdapter.DEFAULT_PORT); + this(String.valueOf(TcpServiceAdapter.DEFAULT_PORT)); } @Override diff --git a/src/main/java/com/doodream/rmovjs/parameter/Param.java b/src/main/java/com/doodream/rmovjs/parameter/Param.java index bf4ed10..3b3b2ca 100644 --- a/src/main/java/com/doodream/rmovjs/parameter/Param.java +++ b/src/main/java/com/doodream/rmovjs/parameter/Param.java @@ -1,10 +1,12 @@ package com.doodream.rmovjs.parameter; +import com.doodream.rmovjs.model.Response; import com.doodream.rmovjs.serde.Converter; -import com.google.common.reflect.TypeToken; +import com.doodream.rmovjs.util.Types; import com.google.gson.Gson; import com.google.gson.GsonBuilder; import io.reactivex.Observable; +import io.reactivex.functions.Predicate; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -12,7 +14,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.lang.model.element.TypeElement; import java.lang.annotation.Annotation; import java.lang.reflect.Type; import java.util.List; @@ -32,15 +33,20 @@ public class Param { private static final Logger Log = LoggerFactory.getLogger(Param.class); private int order; - private ParamType type; + private ParamType location; private boolean required; private String name; private T value; - private Class cls; + private transient Type type; - public static Param create(Class cls, Annotation[] annotations) { + public static Param create(Type cls, Annotation[] annotations) { List filteredAnnotations = Observable.fromArray(annotations) - .filter(ParamType::isSupportedAnnotation) + .filter(new Predicate() { + @Override + public boolean test(Annotation annotation) throws Exception { + return ParamType.isSupportedAnnotation(annotation); + } + }) .toList() .blockingGet(); @@ -48,8 +54,8 @@ public static Param create(Class cls, Annotation[] annotations) { ParamType type = ParamType.fromAnnotation(annotation); return Param.builder() - .type(type) - .cls(cls) + .type(cls) + .location(type) .name(type.getName(annotation)) .required(type.isRequired(annotation)) .build(); @@ -63,13 +69,14 @@ public void apply(T value) { this.value = value; } - public T resolve(Converter converter, Type type) { + public Object resolve(Converter converter, Type type) throws IllegalAccessException, InstantiationException, ClassNotFoundException { + if(Types.isCastable(value, type)) { + return value; + } return converter.resolve(value, type); } - public boolean isInstanceOf(Class itfc) { - return !Observable.fromArray(this.cls.getInterfaces()) - .filter(aClass -> aClass.equals(itfc)) - .isEmpty().blockingGet(); + public boolean isInstanceOf(Type itfc) { + return this.type == itfc; } } diff --git a/src/main/java/com/doodream/rmovjs/sdp/BaseServiceDiscovery.java b/src/main/java/com/doodream/rmovjs/sdp/BaseServiceDiscovery.java index aadd203..68598ec 100644 --- a/src/main/java/com/doodream/rmovjs/sdp/BaseServiceDiscovery.java +++ b/src/main/java/com/doodream/rmovjs/sdp/BaseServiceDiscovery.java @@ -1,18 +1,21 @@ package com.doodream.rmovjs.sdp; import com.doodream.rmovjs.model.RMIServiceInfo; -import com.doodream.rmovjs.net.RMIServiceProxy; -import com.doodream.rmovjs.net.ServiceAdapter; -import com.doodream.rmovjs.net.ServiceProxyFactory; import com.doodream.rmovjs.serde.Converter; import com.google.common.base.Preconditions; -import io.reactivex.Observable; +import io.reactivex.*; import io.reactivex.disposables.Disposable; +import io.reactivex.functions.Action; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; +import io.reactivex.schedulers.Schedulers; import lombok.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; +import java.net.InetAddress; +import java.net.UnknownHostException; import java.util.HashMap; import java.util.HashSet; import java.util.concurrent.TimeUnit; @@ -22,83 +25,153 @@ public abstract class BaseServiceDiscovery implements ServiceDiscovery { private static final Logger Log = LoggerFactory.getLogger(BaseServiceDiscovery.class); + protected interface DiscoveryEventListener { + void onStart(); + void onDiscovered(RMIServiceInfo info); + void onError(Throwable e); + void onStop(); + } + + + private static class ServiceInfoSource implements ObservableOnSubscribe, DiscoveryEventListener { + private ObservableEmitter emitter; + + @Override + public void onStart() { + } + + @Override + public void onDiscovered(RMIServiceInfo info) { + emitter.onNext(info); + } + + @Override + public void onError(Throwable e) { + emitter.onError(e); + } + + @Override + public void onStop() { + emitter.onComplete(); + } + + @Override + public void subscribe(ObservableEmitter observableEmitter) throws Exception { + this.emitter = observableEmitter; + } + } + + private static final long TIMEOUT_IN_SEC = 5L; - private long tickIntervalInMilliSec; - private HashMap disposableMap; + private final HashMap disposableMap; - public BaseServiceDiscovery(long interval, TimeUnit unit) { - tickIntervalInMilliSec = unit.toMillis(interval); + public BaseServiceDiscovery() { disposableMap = new HashMap<>(); } @Override - public void startDiscovery(@NonNull Class service, @NonNull ServiceDiscoveryListener listener) throws InstantiationException, IllegalAccessException { - startDiscovery(service, listener, TIMEOUT_IN_SEC, TimeUnit.SECONDS); - } - - - private Observable observeTick() { - return Observable.interval(0L, tickIntervalInMilliSec, TimeUnit.MILLISECONDS); + public void startDiscovery(Class service, boolean once, InetAddress network, ServiceDiscoveryListener listener) throws InstantiationException, IllegalAccessException { + startDiscovery(service, once, TIMEOUT_IN_SEC, TimeUnit.SECONDS, network, listener); } @Override - public void startDiscovery(@NonNull Class service, @NonNull ServiceDiscoveryListener listener, long timeout, @NonNull TimeUnit unit) throws IllegalAccessException, InstantiationException { + public void startDiscovery(Class service, boolean once, long timeout, TimeUnit unit, InetAddress network, ServiceDiscoveryListener listener) throws IllegalAccessException, InstantiationException { if(disposableMap.containsKey(service)) { + Log.warn("discovery is already running for service {}", service); return; } final RMIServiceInfo info = RMIServiceInfo.from(service); - Converter converter = (Converter) info.getConverter().newInstance(); + final Converter converter = (Converter) info.getConverter().newInstance(); Preconditions.checkNotNull(info, "Invalid service type %s", service); Preconditions.checkNotNull(converter, "converter is not declared"); - HashSet discoveryCache = new HashSet<>(); - listener.onDiscoveryStarted(); + final HashSet discoveryCache = new HashSet<>(); + final ServiceInfoSource serviceInfoSource = new ServiceInfoSource(); - Observable serviceInfoObservable = observeTick() - .map(seq -> receiveServiceInfo(converter)) - .doOnNext(svcInfo -> Log.trace("received info : {}", svcInfo)) - .onErrorReturn(throwable -> RMIServiceInfo.builder().build()) - .filter(discoveryCache::add) - .filter(info::equals) - .doOnNext(discovered -> Log.debug("Discovered New Service : {} @ {}", discovered.getName(), discovered.getProxyFactoryHint())) - .doOnNext(info::copyFrom) - .timeout(timeout, unit); - - - disposableMap.put(service, serviceInfoObservable - .map(RMIServiceInfo::getAdapter) - .map(Class::newInstance) - .cast(ServiceAdapter.class) - .map(serviceAdapter -> serviceAdapter.getProxyFactory(info)) - .map(ServiceProxyFactory::build) - .doOnDispose(() -> { - close(); - disposableMap.remove(service); - listener.onDiscoveryFinished(); + onStartDiscovery(serviceInfoSource, network); + listener.onDiscoveryStarted(); + disposableMap.put(service, Observable.create(serviceInfoSource) + .doOnNext(new Consumer() { + @Override + public void accept(RMIServiceInfo svcInfo) throws Exception { + Log.trace("received info : {}", svcInfo); + } + }) + .onErrorReturn(new Function() { + @Override + public RMIServiceInfo apply(Throwable throwable) throws Exception { + return RMIServiceInfo.builder().build(); + } + }) + .filter(new Predicate() { + @Override + public boolean test(RMIServiceInfo discovered) throws Exception { + if(!once) { + return true; + } + return discoveryCache.add(discovered); + } }) - .doOnError(throwable -> { - if(throwable instanceof TimeoutException) { - Log.debug("Discovery Timeout"); - } else { - Log.warn("{}", throwable); + .filter(new Predicate() { + @Override + public boolean test(RMIServiceInfo rmiServiceInfo) throws Exception { + return info.equals(rmiServiceInfo); } - close(); - disposableMap.remove(service); - listener.onDiscoveryFinished(); }) - .doOnComplete(() -> { - close(); - disposableMap.remove(service); - listener.onDiscoveryFinished(); + .doOnNext(new Consumer() { + @Override + public void accept(RMIServiceInfo discovered) throws Exception { + Log.debug("Discovered New Service : {} @ {}", discovered.getName(), discovered.getProxyFactoryHint()); + info.copyFrom(discovered); + } }) - .onErrorReturn(throwable -> RMIServiceProxy.NULL_PROXY) - .filter(proxy -> !RMIServiceProxy.NULL_PROXY.equals(proxy)) - .subscribe(listener::onDiscovered)); + .timeout(timeout, unit) + .doOnDispose(new Action() { + @Override + public void run() throws Exception { + onStopDiscovery(); + disposableMap.remove(service); + listener.onDiscoveryFinished(); + } + }) + .subscribeOn(Schedulers.io()) + .subscribe(listener::onDiscovered, new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + if(throwable instanceof TimeoutException) { + Log.debug("Discovery Timeout"); + } else { + Log.warn("{}", throwable); + } + onStopDiscovery(); + disposableMap.remove(service); + listener.onDiscoveryFinished(); + } + }, new Action() { + @Override + public void run() throws Exception { + onStopDiscovery(); + disposableMap.remove(service); + listener.onDiscoveryFinished(); + } + })); + } + @Override + public void startDiscovery(@NonNull Class service, boolean once, @NonNull ServiceDiscoveryListener listener) throws InstantiationException, IllegalAccessException, UnknownHostException { + startDiscovery(service, once, TIMEOUT_IN_SEC, TimeUnit.SECONDS, listener); } + @Override + public void startDiscovery(@NonNull final Class service, final boolean once, long timeout, @NonNull TimeUnit unit, @NonNull final ServiceDiscoveryListener listener) throws IllegalAccessException, InstantiationException, UnknownHostException { + InetAddress localHost = InetAddress.getLocalHost(); + startDiscovery(service, once, timeout, unit, localHost, listener); + } + + + @Override public void cancelDiscovery(Class service) { Disposable disposable = disposableMap.get(service); @@ -111,7 +184,6 @@ public void cancelDiscovery(Class service) { disposable.dispose(); } - - protected abstract RMIServiceInfo receiveServiceInfo(Converter converter) throws IOException; - protected abstract void close(); + protected abstract void onStartDiscovery(DiscoveryEventListener listener, InetAddress network); + protected abstract void onStopDiscovery(); } diff --git a/src/main/java/com/doodream/rmovjs/sdp/ServiceAdvertiser.java b/src/main/java/com/doodream/rmovjs/sdp/ServiceAdvertiser.java index 0ddca8e..1742c39 100644 --- a/src/main/java/com/doodream/rmovjs/sdp/ServiceAdvertiser.java +++ b/src/main/java/com/doodream/rmovjs/sdp/ServiceAdvertiser.java @@ -4,16 +4,27 @@ import com.doodream.rmovjs.serde.Converter; import java.io.IOException; +import java.net.InetAddress; public interface ServiceAdvertiser { + + /** + * start to send multicast packet to advertise service through all possible network interfaces + * @param info service information to be advertised + * @param block block if true, otherwise return immediately + * @throws IOException + */ + void startAdvertiser(RMIServiceInfo info, boolean block) throws IOException; + + /** - * start service advertising - * should not block - * @param info - * @param block + * start to send multicast packet to advertise service through given interface + * @param info service information to be advertised + * @param block block if true, otherwise return immediately + * @param inf network interface * @throws IOException */ - void startAdvertiser(RMIServiceInfo info, Converter converter, boolean block) throws IOException; + void startAdvertiser(RMIServiceInfo info, boolean block, InetAddress inf) throws IOException; /** * stop advertising diff --git a/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscovery.java b/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscovery.java index 1362802..202031c 100644 --- a/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscovery.java +++ b/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscovery.java @@ -1,10 +1,13 @@ package com.doodream.rmovjs.sdp; import java.io.IOException; +import java.net.InetAddress; import java.util.concurrent.TimeUnit; public interface ServiceDiscovery { - void startDiscovery(Class service , ServiceDiscoveryListener listener, long timeout, TimeUnit unit) throws IOException, IllegalAccessException, InstantiationException; - void startDiscovery(Class service, ServiceDiscoveryListener listener) throws IOException, InstantiationException, IllegalAccessException; + void startDiscovery(Class service , boolean once, long timeout, TimeUnit unit, InetAddress network, ServiceDiscoveryListener listener) throws IOException, IllegalAccessException, InstantiationException; + void startDiscovery(Class service, boolean once, InetAddress network, ServiceDiscoveryListener listener) throws IOException, InstantiationException, IllegalAccessException; + void startDiscovery(Class service, boolean once, long timeout, TimeUnit unit, ServiceDiscoveryListener listener) throws IOException, IllegalAccessException, InstantiationException; + void startDiscovery(Class service, boolean once, ServiceDiscoveryListener listener) throws IOException, IllegalAccessException, InstantiationException; void cancelDiscovery(Class service) throws IOException; } diff --git a/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscoveryListener.java b/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscoveryListener.java index e085fb9..97dbc8e 100644 --- a/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscoveryListener.java +++ b/src/main/java/com/doodream/rmovjs/sdp/ServiceDiscoveryListener.java @@ -1,9 +1,9 @@ package com.doodream.rmovjs.sdp; -import com.doodream.rmovjs.net.RMIServiceProxy; +import com.doodream.rmovjs.model.RMIServiceInfo; public interface ServiceDiscoveryListener { - void onDiscovered(RMIServiceProxy proxy); + void onDiscovered(RMIServiceInfo info); void onDiscoveryStarted(); void onDiscoveryFinished() throws IllegalAccessException; } diff --git a/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceAdvertiser.java b/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceAdvertiser.java index d297d6b..124183c 100644 --- a/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceAdvertiser.java +++ b/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceAdvertiser.java @@ -2,19 +2,26 @@ import com.doodream.rmovjs.model.RMIServiceInfo; import com.doodream.rmovjs.serde.Converter; +import com.doodream.rmovjs.serde.json.JsonConverter; import io.reactivex.Observable; import io.reactivex.disposables.CompositeDisposable; +import io.reactivex.functions.*; import io.reactivex.schedulers.Schedulers; +import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.UnsupportedEncodingException; import java.net.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Enumeration; +import java.util.List; import java.util.concurrent.TimeUnit; /** - * this class provices simple service advertising capability whose intented use is testing though, + * this class provides simple service advertising capability whose intended use is testing though, * can be used a simple service discovery scenario. * * advertiser start to advertise its RMIServiceInfo with broadcasting datagram socket @@ -26,43 +33,74 @@ public class SimpleServiceAdvertiser implements ServiceAdvertiser { public static final String MULTICAST_GROUP_IP = "224.0.2.118"; // AD-HOC block 1 private CompositeDisposable compositeDisposable = new CompositeDisposable(); + + public static InetAddress getGroupAddress() throws UnknownHostException { + return InetAddress.getByName(MULTICAST_GROUP_IP); + } + @Override - public synchronized void startAdvertiser(RMIServiceInfo info, Converter converter, boolean block) throws IOException { + public synchronized void startAdvertiser(final RMIServiceInfo info, boolean block) throws IOException { - Observable tickObservable = Observable.interval(0L, 3L, TimeUnit.SECONDS); + InetAddress localhost = InetAddress.getLocalHost(); + startAdvertiser(info, block, localhost); - compositeDisposable.add(tickObservable - .map(aLong -> info) - .map(i -> buildLocalPacket(i, converter)) - .doOnNext(this::broadcast) - .doOnError(Throwable::printStackTrace) - .subscribeOn(Schedulers.io()) - .subscribe()); + } + + @Override + public void startAdvertiser(RMIServiceInfo info, boolean block, InetAddress inf) throws IOException { + final Observable tickObservable = Observable.interval(0L, 3L, TimeUnit.SECONDS); + final JsonConverter converter = new JsonConverter(); + final MulticastSocket socket = new MulticastSocket(BROADCAST_PORT); + socket.setInterface(inf); + socket.setTimeToLive(2); - Observable packetObservable = tickObservable - .map(aLong -> info) - .map(i -> buildMulticastPacket(i, converter)) - .doOnNext(this::broadcast) - .doOnError(Throwable::printStackTrace); - Log.info("advertising service : {} {} @ {}", info.getName(), info.getVersion(), MULTICAST_GROUP_IP); + Observable packetObservable = tickObservable + .map(new Function() { + @Override + public RMIServiceInfo apply(Long aLong) throws Exception { + return info; + } + }) + .map(new Function() { + @Override + public DatagramPacket apply(RMIServiceInfo info) throws Exception { + return buildMulticastPacket(info, converter); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(DatagramPacket datagramPacket) throws Exception { + Log.debug("send Service Info @ {}", socket.getInterface()); + socket.send(datagramPacket); + } + }) + .doOnDispose(new Action() { + @Override + public void run() throws Exception { + if(!socket.isClosed()) { + socket.close(); + } + } + }) + .doOnError(new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + onError(throwable); + } + }) + .subscribeOn(Schedulers.io()); if(!block) { - compositeDisposable.add(packetObservable.subscribeOn(Schedulers.io()).subscribe()); - return; + compositeDisposable.add(packetObservable.subscribe()); + } else { + packetObservable.blockingSubscribe(); } - packetObservable.blockingSubscribe(); } - private DatagramPacket buildLocalPacket(RMIServiceInfo i, Converter converter) throws UnsupportedEncodingException, UnknownHostException { - byte[] infoByteString = converter.convert(i); - return new DatagramPacket(infoByteString, infoByteString.length, new InetSocketAddress(BROADCAST_PORT)); + private void onError(Throwable throwable) { + Log.error(throwable.getLocalizedMessage()); } - private void broadcast(DatagramPacket datagramPacket) throws IOException { - DatagramSocket socket = new DatagramSocket(); - socket.send(datagramPacket); - socket.close(); - } private DatagramPacket buildMulticastPacket(RMIServiceInfo info, Converter converter) throws UnsupportedEncodingException, UnknownHostException { byte[] infoByteString = converter.convert(info); diff --git a/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceDiscovery.java b/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceDiscovery.java index ab4da12..c95586a 100644 --- a/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceDiscovery.java +++ b/src/main/java/com/doodream/rmovjs/sdp/SimpleServiceDiscovery.java @@ -1,13 +1,21 @@ package com.doodream.rmovjs.sdp; import com.doodream.rmovjs.model.RMIServiceInfo; -import com.doodream.rmovjs.serde.Converter; +import com.doodream.rmovjs.serde.json.JsonConverter; +import io.reactivex.Observable; +import io.reactivex.disposables.CompositeDisposable; +import io.reactivex.functions.Action; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.schedulers.Schedulers; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; -import java.net.DatagramPacket; -import java.net.InetAddress; -import java.net.MulticastSocket; -import java.util.Arrays; +import java.net.*; +import java.nio.channels.DatagramChannel; +import java.nio.channels.MulticastChannel; +import java.util.*; import java.util.concurrent.TimeUnit; /** @@ -18,31 +26,88 @@ */ public class SimpleServiceDiscovery extends BaseServiceDiscovery { - private MulticastSocket serviceBroadcastSocket; - public SimpleServiceDiscovery() throws IOException { - super(100L, TimeUnit.MILLISECONDS); - serviceBroadcastSocket = new MulticastSocket(SimpleServiceAdvertiser.BROADCAST_PORT); - serviceBroadcastSocket.joinGroup(InetAddress.getByName(SimpleServiceAdvertiser.MULTICAST_GROUP_IP)); + private static final Logger Log = LoggerFactory.getLogger(SimpleServiceDiscovery.class); + private final CompositeDisposable disposable; + private final JsonConverter converter; + + public SimpleServiceDiscovery() { + super(); + disposable = new CompositeDisposable(); + converter = new JsonConverter(); + } @Override - protected RMIServiceInfo receiveServiceInfo(Converter converter) throws IOException { - byte[] buffer = new byte[64 * 1024]; - Arrays.fill(buffer, (byte) 0); - DatagramPacket packet = new DatagramPacket(buffer, buffer.length); - serviceBroadcastSocket.receive(packet); - return converter.invert(packet.getData(), RMIServiceInfo.class); + protected void onStartDiscovery(DiscoveryEventListener listener, InetAddress network) { + try { + Log.debug("subscribe SDP on {}", network); + final MulticastSocket socket = new MulticastSocket(SimpleServiceAdvertiser.BROADCAST_PORT); + socket.setInterface(network); + socket.joinGroup(SimpleServiceAdvertiser.getGroupAddress()); + disposable.add(listenMulticast(socket, 1000L, TimeUnit.MILLISECONDS) + .map(new Function() { + @Override + public RMIServiceInfo apply(byte[] bytes) throws Exception { + return converter.invert(bytes, RMIServiceInfo.class); + } + }) + .doOnDispose(new Action() { + @Override + public void run() throws Exception { + if (!socket.isClosed()) { + socket.close(); + } + } + }) + .subscribeOn(Schedulers.io()) + .subscribe(new Consumer() { + @Override + public void accept(RMIServiceInfo info) throws Exception { + Log.debug("service discovered ({}) on {}", info.getName(), socket.getInterface()); + listener.onDiscovered(info); + } + }, new Consumer() { + @Override + public void accept(Throwable throwable) throws Exception { + listener.onError(throwable); + } + }, new Action() { + @Override + public void run() throws Exception { + listener.onStop(); + } + })); + } catch (IOException e) { + Log.error("fail to start discovery", e); + } + } + + private Observable listenMulticast(MulticastSocket socket, long interval, TimeUnit timeUnit) throws IOException { + return Observable.interval(interval, timeUnit) + .map(new Function() { + @Override + public DatagramPacket apply(Long aLong) throws Exception { + byte[] buffer = new byte[64 * 1024]; + Arrays.fill(buffer, (byte) 0); + return new DatagramPacket(buffer, buffer.length); + } + }) + .map(new Function() { + @Override + public byte[] apply(DatagramPacket datagramPacket) throws Exception { + socket.receive(datagramPacket); + Log.debug("service info from {}", datagramPacket.getAddress()); + return datagramPacket.getData(); + } + }); } @Override - protected void close() { - if(serviceBroadcastSocket == null) { - return; - } - if(serviceBroadcastSocket.isClosed()) { - return; + protected void onStopDiscovery() { + if(!disposable.isDisposed()) { + disposable.dispose(); } - serviceBroadcastSocket.close(); + disposable.clear(); } } diff --git a/src/main/java/com/doodream/rmovjs/serde/Converter.java b/src/main/java/com/doodream/rmovjs/serde/Converter.java index 7dc473c..4087e3e 100644 --- a/src/main/java/com/doodream/rmovjs/serde/Converter.java +++ b/src/main/java/com/doodream/rmovjs/serde/Converter.java @@ -2,7 +2,6 @@ import java.io.InputStream; import java.io.OutputStream; -import java.io.UnsupportedEncodingException; import java.lang.reflect.Type; /** @@ -17,11 +16,9 @@ public interface Converter { Writer writer(OutputStream outputStream); - byte[] convert(Object src) throws UnsupportedEncodingException; + byte[] convert(Object src); - T invert(byte[] b, Class cls) throws UnsupportedEncodingException; + T invert(byte[] b, Class cls); - T invert(byte[] b, Class rawClass, Class ...parameters); - - T resolve(Object unresolved, Type type); + Object resolve(Object unresolved, Type type) throws ClassNotFoundException, IllegalAccessException, InstantiationException; } diff --git a/src/main/java/com/doodream/rmovjs/serde/Reader.java b/src/main/java/com/doodream/rmovjs/serde/Reader.java index 40f5bae..d49afd1 100644 --- a/src/main/java/com/doodream/rmovjs/serde/Reader.java +++ b/src/main/java/com/doodream/rmovjs/serde/Reader.java @@ -8,5 +8,4 @@ */ public interface Reader { T read(Class cls) throws IOException; - T read(Class rawClass, Class parameter) throws IOException; } diff --git a/src/main/java/com/doodream/rmovjs/serde/bson/BsonConverter.java b/src/main/java/com/doodream/rmovjs/serde/bson/BsonConverter.java new file mode 100644 index 0000000..2d86541 --- /dev/null +++ b/src/main/java/com/doodream/rmovjs/serde/bson/BsonConverter.java @@ -0,0 +1,193 @@ +package com.doodream.rmovjs.serde.bson; + +import com.doodream.rmovjs.serde.Converter; +import com.doodream.rmovjs.serde.Reader; +import com.doodream.rmovjs.serde.Writer; +import com.doodream.rmovjs.util.Types; +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.PropertyAccessor; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.MapperFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.exc.MismatchedInputException; +import de.undercouch.bson4jackson.BsonFactory; +import de.undercouch.bson4jackson.BsonGenerator; +import de.undercouch.bson4jackson.BsonParser; +import io.reactivex.Observable; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.lang.reflect.*; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public class BsonConverter implements Converter { + private static final Logger Log = LoggerFactory.getLogger(BsonConverter.class); + + private ObjectMapper objectMapper; + private BsonFactory bsonFactory; + public BsonConverter() { + + bsonFactory = new BsonFactory(); + bsonFactory.enable(BsonGenerator.Feature.ENABLE_STREAMING); + + objectMapper = new ObjectMapper(bsonFactory + .disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET) + .enable(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT)) + .setSerializationInclusion(JsonInclude.Include.NON_NULL) + .disable(MapperFeature.AUTO_DETECT_IS_GETTERS, MapperFeature.AUTO_DETECT_GETTERS) + .setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); + } + @Override + public Reader reader(final InputStream inputStream) { + try { + return new Reader() { + + private BsonParser parser = bsonFactory.createParser(new BufferedInputStream(inputStream)); + + @Override + public synchronized T read(Class cls) throws IOException { + return parser.readValueAs(cls); + } + }; + } catch (IOException e) { + Log.error(e.getLocalizedMessage()); + return null; + } + } + + @Override + public Writer writer(final OutputStream outputStream) { + try { + return new Writer() { + private BsonGenerator bsonGenerator = bsonFactory.createGenerator(outputStream); + + @Override + public synchronized void write(Object src) throws IOException { + bsonGenerator.writeObject(src); + } + }; + } catch (IOException e) { + Log.error(e.getLocalizedMessage()); + return null; + } + } + + @Override + public byte[] convert(Object src) { + try { + return objectMapper.writeValueAsBytes(src); + } catch (JsonProcessingException e) { + e.printStackTrace(); + } + return new byte[0]; + } + + @Override + public T invert(byte[] b, Class cls) { + try { + return objectMapper.readValue(b, cls); + } catch (IOException e) { + e.printStackTrace(); + } + return null; + } + + + @Override + public Object resolve(final Object unresolved, Type type) throws InstantiationException, IllegalAccessException { + if(unresolved == null) { + return null; + } + Class clsz; + try { + if (type instanceof ParameterizedType) { + clsz = Class.forName(((ParameterizedType) type).getRawType().getTypeName()); + } else { + clsz = Class.forName(type.getTypeName()); + } + } catch (ClassNotFoundException e) { + return unresolved; + } + final Class cls = clsz; + final Class unresolvedCls = unresolved.getClass(); + Log.debug("resolve {} -> {}", unresolvedCls, cls); + + if(unresolvedCls.equals(LinkedHashMap.class)) { + return resolveKvMap((Map) unresolved, cls); + } + + if(unresolvedCls.equals(ArrayList.class)) { + Type[] typeArguments = ((ParameterizedType) type).getActualTypeArguments(); + if(typeArguments == null || (typeArguments.length == 0)) { + return unresolved; + } + ArrayList unresolvedList = (ArrayList) unresolved; + return Observable.fromIterable(unresolvedList).map(new Function() { + @Override + public Object apply(Object o) throws Exception { + return resolve(o, typeArguments[0]); + }}).toList().blockingGet(); + } + + if(cls.equals(unresolvedCls) || + Types.isCastable(unresolved, type) || + Types.isCastable(unresolved, cls)) { + return cls.cast(unresolved); + } + + + + if(unresolvedCls.getSuperclass().equals(Number.class)) { + + } + + try { + Constructor constructor = cls.getConstructor(unresolvedCls); + return constructor.newInstance(unresolved); + } catch (NoSuchMethodException | InvocationTargetException ignored) { + + } + try { + Method valueOf = cls.getMethod("valueOf", String.class); + return valueOf.invoke(null, String.valueOf(unresolved)); + } catch (NoSuchMethodException | InvocationTargetException ignored) { + + } + return unresolved; + + } + + private Object resolveKvMap(final Map map, Class cls) throws IllegalAccessException, InstantiationException { + final Object resolved = cls.newInstance(); + Observable.fromArray(cls.getDeclaredFields()) + .filter(new Predicate() { + @Override + public boolean test(Field field) throws Exception { + return map.containsKey(field.getName()); + } + }) + .doOnNext(new Consumer() { + @Override + public void accept(Field field) throws Exception { + field.setAccessible(true); + field.set(resolved, map.get(field.getName())); + } + }) + .blockingSubscribe(); + + return resolved; + } + +} diff --git a/src/main/java/com/doodream/rmovjs/serde/json/JsonConverter.java b/src/main/java/com/doodream/rmovjs/serde/json/JsonConverter.java index 1e59ffc..a011974 100644 --- a/src/main/java/com/doodream/rmovjs/serde/json/JsonConverter.java +++ b/src/main/java/com/doodream/rmovjs/serde/json/JsonConverter.java @@ -6,30 +6,23 @@ import com.doodream.rmovjs.serde.Reader; import com.doodream.rmovjs.serde.Writer; import com.google.gson.*; -import com.google.gson.reflect.TypeToken; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.lang.reflect.ParameterizedType; + +import javax.xml.bind.DatatypeConverter; +import java.io.*; import java.lang.reflect.Type; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; -import java.util.Base64; public class JsonConverter implements Converter { - private static final Logger Log = LoggerFactory.getLogger(JsonConverter.class); private static class ByteArrayToBase64TypeAdapter implements JsonSerializer, JsonDeserializer { public byte[] deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException { - byte[] data = Base64.getDecoder().decode(json.getAsString()); + byte[] data = DatatypeConverter.parseBase64Binary(json.getAsString()); return data; } public JsonElement serialize(byte[] src, Type typeOfSrc, JsonSerializationContext context) { - return new JsonPrimitive(Base64.getEncoder().withoutPadding().encodeToString(src)); + return new JsonPrimitive(DatatypeConverter.printBase64Binary(src)); } } @@ -53,26 +46,7 @@ public Class read(com.google.gson.stream.JsonReader jsonReader) throws IOExcepti } return null; } - }).registerTypeAdapter(Class.class, new TypeAdapter>() { - - @Override - public void write(com.google.gson.stream.JsonWriter jsonWriter, Class objectClass) throws IOException { - jsonWriter.value(objectClass.getName()); - - } - - @Override - public Class read(com.google.gson.stream.JsonReader jsonReader) throws IOException { - try { - return (Class) Class.forName(jsonReader.nextString()); - } catch (ClassNotFoundException e) { - e.printStackTrace(); - } - return null; - } - - }) - .registerTypeAdapter(SessionControlMessage.class, new TypeAdapter>() { + }).registerTypeAdapter(SessionControlMessage.class, new TypeAdapter>() { @Override public void write(com.google.gson.stream.JsonWriter jsonWriter, SessionControlMessage sessionControlMessage) throws IOException { if(sessionControlMessage != null) { @@ -116,34 +90,32 @@ public SessionControlMessage read(com.google.gson.stream.JsonReader jsonReade }) .create(); - private static Type getType(Class rawClass, Class ...parameter) { - return new ParameterizedType() { - @Override - public Type[] getActualTypeArguments() { - return parameter; - } - @Override - public Type getRawType() { - return rawClass; - } + @Override + public Reader reader(final InputStream inputStream) { + return new Reader() { + private BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)); @Override - public Type getOwnerType() { - return null; + public T read(Class cls) throws IOException { + String line = reader.readLine(); + if(line == null) { + return null; + } + line = line.trim(); + return invert(StandardCharsets.UTF_8.encode(line).array(), cls); } }; } - - @Override - public Reader reader(InputStream inputStream) { - return new JsonReader(this, inputStream); - } - @Override - public Writer writer(OutputStream outputStream) { - return new JsonWriter(this, outputStream); + public Writer writer(final OutputStream outputStream) { + return new Writer() { + @Override + public void write(Object src) throws IOException { + outputStream.write(convert(src)); + } + }; } @Override @@ -158,13 +130,10 @@ public T invert(byte[] b, Class cls) { } @Override - public T invert(byte[] b, Class rawClass, Class ...parameter) { - ByteBuffer buffer = ByteBuffer.wrap(b); - return GSON.fromJson(StandardCharsets.UTF_8.decode(buffer).toString(), getType(rawClass, parameter)); - } - - @Override - public T resolve(Object unresolved, Type type) { + public Object resolve(Object unresolved, Type type) { + if(unresolved == null) { + return null; + } return GSON.fromJson(GSON.toJson(unresolved), type); } diff --git a/src/main/java/com/doodream/rmovjs/serde/json/JsonReader.java b/src/main/java/com/doodream/rmovjs/serde/json/JsonReader.java deleted file mode 100644 index 1e54756..0000000 --- a/src/main/java/com/doodream/rmovjs/serde/json/JsonReader.java +++ /dev/null @@ -1,41 +0,0 @@ -package com.doodream.rmovjs.serde.json; - -import com.doodream.rmovjs.serde.Converter; -import com.doodream.rmovjs.serde.Reader; - -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStream; -import java.io.InputStreamReader; -import java.nio.charset.StandardCharsets; - -public class JsonReader implements Reader { - private BufferedReader mBufferedReader; - private Converter mConverter; - - JsonReader(Converter converter, InputStream is) { - mConverter = converter; - mBufferedReader = new BufferedReader(new InputStreamReader(is)); - } - - - @Override - public synchronized T read(Class cls) throws IOException { - String line = mBufferedReader.readLine(); - if(line == null) { - return null; - } - line = line.trim(); - return mConverter.invert(StandardCharsets.UTF_8.encode(line).array(), cls); - } - - @Override - public synchronized T read(Class rawClass, Class parameter) throws IOException { - String line = mBufferedReader.readLine(); - if(line == null) { - return null; - } - line = line.trim(); - return mConverter.invert(StandardCharsets.UTF_8.encode(line).array(), rawClass, parameter); - } -} diff --git a/src/main/java/com/doodream/rmovjs/serde/json/JsonWriter.java b/src/main/java/com/doodream/rmovjs/serde/json/JsonWriter.java deleted file mode 100644 index 9506c9c..0000000 --- a/src/main/java/com/doodream/rmovjs/serde/json/JsonWriter.java +++ /dev/null @@ -1,27 +0,0 @@ -package com.doodream.rmovjs.serde.json; - -import com.doodream.rmovjs.serde.Converter; -import com.doodream.rmovjs.serde.Writer; - -import java.io.IOException; -import java.io.OutputStream; -import java.nio.ByteBuffer; -import java.nio.channels.Channels; -import java.nio.channels.WritableByteChannel; - -public class JsonWriter implements Writer { - - private Converter mConverter; - private WritableByteChannel mChannelOut; - - JsonWriter(Converter converter, OutputStream os) { - mConverter = converter; - mChannelOut = Channels.newChannel(os); - } - - @Override - public synchronized void write(Object src) throws IOException { - ByteBuffer json = ByteBuffer.wrap(mConverter.convert(src)); - mChannelOut.write(json); - } -} diff --git a/src/main/java/com/doodream/rmovjs/server/RMIController.java b/src/main/java/com/doodream/rmovjs/server/RMIController.java index 3ca4f48..5ce7a39 100644 --- a/src/main/java/com/doodream/rmovjs/server/RMIController.java +++ b/src/main/java/com/doodream/rmovjs/server/RMIController.java @@ -10,8 +10,13 @@ import com.doodream.rmovjs.net.session.BlobSession; import com.doodream.rmovjs.parameter.Param; import com.doodream.rmovjs.serde.Converter; +import com.google.common.base.Preconditions; import io.reactivex.Observable; import io.reactivex.Single; +import io.reactivex.functions.BiConsumer; +import io.reactivex.functions.BiFunction; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -21,11 +26,9 @@ import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.lang.reflect.Type; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; @AllArgsConstructor @Builder @@ -36,8 +39,8 @@ public class RMIController { private static final Logger Log = LoggerFactory.getLogger(RMIController.class); private Controller controller; private Map endpointMap; + private Class stub; private Object impl; - private Class itfcCls; /** * create {@link RMIController} by analyzing fields of {@link RMIService} @@ -47,28 +50,84 @@ public class RMIController { * @throws InstantiationException fail to resolve implementation class of controller */ static public RMIController create(Field field) throws IllegalAccessException, InstantiationException { - Controller controller = field.getAnnotation(Controller.class); - Class cls = field.getType(); - assert controller != null; + return create(field, new Object[0]); + } + + /** + * + * @param field + * @param controllerImpls + * @return + * @throws IllegalAccessException + * @throws InstantiationException + */ + public static RMIController create(final Field field, Object[] controllerImpls) throws IllegalAccessException, InstantiationException { + + final Controller controller = field.getAnnotation(Controller.class); + + Preconditions.checkNotNull(controller, "controller should be annotated with @Controller"); + final Class cls = field.getType(); Class module = controller.module(); - Object impl = module.newInstance(); - Observable endpointsObservable = Observable.fromArray(cls.getMethods()) - .filter(RMIMethod::isValidMethod) - .map(method -> Endpoint.create(controller, method)); - Single> endpointLookupSingle = endpointsObservable - .collectInto(new HashMap<>(), RMIController::collectMethod); + final Object impl = Observable.fromArray(controllerImpls) + .filter(new Predicate() { + @Override + public boolean test(Object o) throws Exception { + return isImplementOf(o, field.getGenericType()); + } + }) + .defaultIfEmpty(module.newInstance()) + .blockingFirst(); + + Preconditions.checkNotNull(impl, "implementation should not null"); + + Observable endpointObservable = Observable.fromArray(cls.getDeclaredMethods()) + .filter(new Predicate() { + @Override + public boolean test(Method method) throws Exception { + return RMIMethod.isValidMethod(method); + } + }) + .map(new Function() { + @Override + public Endpoint apply(Method method) throws Exception { + return Endpoint.create(controller, method); + } + }); + + Single> endpointLookupSingle = endpointObservable + .collectInto(new HashMap(), new BiConsumer, Endpoint>() { + @Override + public void accept(HashMap stringEndpointHashMap, Endpoint endpoint) throws Exception { + RMIController.collectMethod(stringEndpointHashMap, endpoint); + } + }); return Observable.just(RMIController.builder()) - .map(controllerBuilder -> controllerBuilder.impl(impl)) - .map(controllerBuilder -> controllerBuilder.controller(controller)) - .map(controllerBuilder -> controllerBuilder.itfcCls(cls)) - .zipWith(endpointLookupSingle.toObservable(), RMIControllerBuilder::endpointMap) - .map(RMIControllerBuilder::build) + .map(new Function() { + @Override + public RMIControllerBuilder apply(RMIControllerBuilder builder) throws Exception { + return builder.impl(impl).controller(controller).stub(cls); + } + }) + .zipWith(endpointLookupSingle.toObservable(), new BiFunction, RMIControllerBuilder>() { + @Override + public RMIControllerBuilder apply(RMIControllerBuilder builder, HashMap stringEndpointHashMap) throws Exception { + return builder.endpointMap(stringEndpointHashMap); + } + }) + .map(new Function() { + @Override + public RMIController apply(RMIControllerBuilder builder) throws Exception { + return builder.build(); + } + }) .blockingFirst(); + } + /** * collect methods({@link Endpoint}) into map for lookup * @param map map to collect endpoint into @@ -84,7 +143,13 @@ private static void collectMethod(HashMap map, Endpoint endpoi * @return return true if the controller valid, otherwise return false */ public static boolean isValidController(Field field) { - return field.getAnnotation(Controller.class) != null; + Controller controller = field.getAnnotation(Controller.class); + return controller != null; + } + + + private static boolean isImplementOf(Object o, Type itfcType) { + return Arrays.asList(o.getClass().getGenericInterfaces()).contains(itfcType); } /** @@ -102,7 +167,7 @@ List getEndpoints() { * @throws InvocationTargetException exception occurred within the method call * @throws IllegalAccessException */ - Response handleRequest(Request request, Converter converter) throws InvocationTargetException, IllegalAccessException { + Response handleRequest(final Request request, final Converter converter) throws InvocationTargetException, IllegalAccessException { Endpoint endpoint = endpointMap.get(request.getEndpoint()); @@ -113,17 +178,29 @@ Response handleRequest(Request request, Converter converter) throws InvocationTa Observable typeObservable = Observable.fromArray(endpoint.getJMethod().getGenericParameterTypes()); List params = Observable.fromIterable(request.getParams()) - .sorted(Param::sort) - .zipWith(typeObservable, (param, type) -> param.resolve(converter, type)) - .map(o -> { - if(o instanceof BlobSession) { - return request.getSession(); + .sorted(new Comparator() { + @Override + public int compare(Param o1, Param o2) { + return Param.sort(o1, o2); + } + }) + .zipWith(typeObservable, new BiFunction() { + @Override + public Object apply(Param param, Type type) throws Exception { + return param.resolve(converter,type); + } + }) + .map(new Function() { + @Override + public Object apply(Object o) throws Exception { + if(o instanceof BlobSession) { + return request.getSession(); + } + return o; } - return o; }) .toList().blockingGet(); - return (Response) endpoint.getJMethod().invoke(getImpl(), params.toArray()); } } diff --git a/src/main/java/com/doodream/rmovjs/server/RMIService.java b/src/main/java/com/doodream/rmovjs/server/RMIService.java index f279e55..10ca0c6 100644 --- a/src/main/java/com/doodream/rmovjs/server/RMIService.java +++ b/src/main/java/com/doodream/rmovjs/server/RMIService.java @@ -10,6 +10,10 @@ import com.doodream.rmovjs.serde.Converter; import com.google.common.base.Preconditions; import io.reactivex.Observable; +import io.reactivex.functions.BiConsumer; +import io.reactivex.functions.Consumer; +import io.reactivex.functions.Function; +import io.reactivex.functions.Predicate; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -19,9 +23,14 @@ import java.io.IOException; import java.lang.reflect.Constructor; +import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; +import java.net.InetAddress; +import java.net.NetworkInterface; import java.util.Arrays; import java.util.HashMap; +import java.util.List; +import java.util.Locale; /** * Created by innocentevil on 18. 5. 4. @@ -48,6 +57,117 @@ public class RMIService { protected static final String TAG = RMIService.class.getCanonicalName(); + /** + * + * @param cls + * @param advertiser + * @param controllerImpls + * @param + * @return + */ + public static RMIService create(Class cls, ServiceAdvertiser advertiser, final Object ...controllerImpls) throws IllegalAccessException, InvocationTargetException, InstantiationException { + + Service service = cls.getAnnotation(Service.class); + final String[] params = service.params(); + + Constructor constructor = Observable.fromArray(service.adapter().getConstructors()) + .filter(new Predicate>() { + @Override + public boolean test(Constructor constructor) throws Exception { + return constructor.getParameterCount() == params.length; + } + }) + .blockingFirst(); + + Preconditions.checkNotNull(constructor); + + ServiceAdapter adapter = (ServiceAdapter) constructor.newInstance(params); + + final RMIServiceInfo serviceInfo = RMIServiceInfo.builder() + .name(service.name()) + .adapter(service.adapter()) + .negotiator(service.negotiator()) + .converter(service.converter()) + .params(Arrays.asList(params)) + .version(Properties.getVersionString()) + .build(); + + final Converter converter = service.converter().newInstance(); + Preconditions.checkNotNull(converter, "converter is not declared"); + final RMIServiceBuilder builder = RMIService.builder(); + + Observable basicControllerObservable = Observable.fromArray(BasicService.class.getDeclaredFields()) + .filter(new Predicate() { + @Override + public boolean test(Field field) throws Exception { + return RMIController.isValidController(field); + } + }) + .map(new Function() { + @Override + public RMIController apply(Field field) throws Exception { + return RMIController.create(field); + } + }) + .cache(); + + Observable controllerObservable = Observable.fromArray(cls.getDeclaredFields()) + .filter(new Predicate() { + @Override + public boolean test(Field field) throws Exception { + return RMIController.isValidController(field); + } + }) + .map(new Function() { + @Override + public RMIController apply(Field field) throws Exception { + return RMIController.create(field, controllerImpls); + } + }) + .cache(); + + controllerObservable + .map(new Function() { + @Override + public ControllerInfo apply(RMIController rmiController) throws Exception { + return ControllerInfo.build(rmiController); + } + }) + .toList() + .doOnSuccess(new Consumer>() { + @Override + public void accept(List controllerInfos) throws Exception { + serviceInfo.setControllerInfos(controllerInfos); + } + }) + .subscribe(); + + controllerObservable = controllerObservable.mergeWith(basicControllerObservable); + + + controllerObservable.collectInto(new HashMap(), new BiConsumer, RMIController>() { + @Override + public void accept(HashMap stringRMIControllerHashMap, RMIController rmiController) throws Exception { + RMIService.buildControllerMap(stringRMIControllerHashMap, rmiController); + } + }).doOnSuccess(new Consumer>() { + @Override + public void accept(HashMap controllerMap) throws Exception { + builder.controllerMap(controllerMap); + } + }).subscribe(); + + + return builder + .adapter(adapter) + .service(service) + .advertiser(advertiser) + .converter(converter) + .serviceInfo(serviceInfo) + .build(); + + } + /** * * @param cls service definition class @@ -59,15 +179,20 @@ public class RMIService { */ public static RMIService create(Class cls, ServiceAdvertiser advertiser) throws IllegalAccessException, InstantiationException, InvocationTargetException { Service service = cls.getAnnotation(Service.class); - String[] params = service.params(); + final String[] params = service.params(); Constructor constructor = Observable.fromArray(service.adapter().getConstructors()) - .filter(ctor -> ctor.getParameterCount() == params.length) + .filter(new Predicate>() { + @Override + public boolean test(Constructor constructor) throws Exception { + return constructor.getParameterCount() == params.length; + } + }) .blockingFirst(); ServiceAdapter adapter = (ServiceAdapter) constructor.newInstance(((Object[]) params)); - RMIServiceInfo serviceInfo = RMIServiceInfo.builder() + final RMIServiceInfo serviceInfo = RMIServiceInfo.builder() .name(service.name()) .adapter(service.adapter()) .negotiator(service.negotiator()) @@ -76,33 +201,71 @@ public static RMIService create(Class cls, ServiceAdvertiser advertiser) .version(Properties.getVersionString()) .build(); - final Converter converter = (Converter) serviceInfo.getConverter().newInstance(); + final Converter converter = service.converter().newInstance(); Preconditions.checkNotNull(converter, "converter is not declared"); - RMIServiceBuilder builder = RMIService.builder(); + final RMIServiceBuilder builder = RMIService.builder(); // register controller for BasicService Observable basicControllerObservable = Observable.fromArray(BasicService.class.getDeclaredFields()) - .filter(RMIController::isValidController) - .map(RMIController::create) + .filter(new Predicate() { + @Override + public boolean test(Field field) throws Exception { + return RMIController.isValidController(field); + } + }) + .map(new Function() { + @Override + public RMIController apply(Field field) throws Exception { + return RMIController.create(field); + } + }) .cache(); Observable controllerObservable = Observable.fromArray(cls.getDeclaredFields()) - .filter(RMIController::isValidController) - .map(RMIController::create) + .filter(new Predicate() { + @Override + public boolean test(Field field) throws Exception { + return RMIController.isValidController(field); + } + }) + .map(new Function() { + @Override + public RMIController apply(Field field) throws Exception { + return RMIController.create(field); + } + }) .cache(); controllerObservable - .map(ControllerInfo::build) + .map(new Function() { + @Override + public ControllerInfo apply(RMIController rmiController) throws Exception { + return ControllerInfo.build(rmiController); + } + }) .toList() - .doOnSuccess(serviceInfo::setControllerInfos) + .doOnSuccess(new Consumer>() { + @Override + public void accept(List controllerInfos) throws Exception { + serviceInfo.setControllerInfos(controllerInfos); + } + }) .subscribe(); controllerObservable = controllerObservable.mergeWith(basicControllerObservable); - controllerObservable.collectInto(new HashMap<>(), RMIService::buildControllerMap) - .doOnSuccess(builder::controllerMap) - .subscribe(); + controllerObservable.collectInto(new HashMap(), new BiConsumer, RMIController>() { + @Override + public void accept(HashMap stringRMIControllerHashMap, RMIController rmiController) throws Exception { + RMIService.buildControllerMap(stringRMIControllerHashMap, rmiController); + } + }).doOnSuccess(new Consumer>() { + @Override + public void accept(HashMap controllerMap) throws Exception { + builder.controllerMap(controllerMap); + } + }).subscribe(); return builder @@ -119,22 +282,56 @@ public static RMIService create(Class cls, ServiceAdvertiser advertiser) * @param map map used to collect controller * @param controller controller to be collected */ - private static void buildControllerMap(HashMap map, RMIController controller) { + private static void buildControllerMap(final HashMap map, final RMIController controller) { Observable.fromIterable(controller.getEndpoints()) - .doOnNext(s -> map.put(s, controller)) + .doOnNext(new Consumer() { + @Override + public void accept(String s) throws Exception { + map.put(s, controller); + } + }) .subscribe(); } /** - * start to listen for client connection while advertising its service + * start listening for client connection over default network interface, while advertising service * @param block if true, this call will block indefinite time, otherwise return immediately * @throws IOException server 측 네트워크 endpoint 생성의 실패 혹은 I/O 오류 * @throws IllegalAccessError the error thrown when {@link ServiceAdapter} fails to resolve dependency object (e.g. negotiator, * @throws InstantiationException if dependent class represents an abstract class,an interface, an array class, a primitive type, or void;or if the class has no nullary constructor; */ public void listen(boolean block) throws IOException, IllegalAccessException, InstantiationException { - serviceInfo.setProxyFactoryHint(adapter.listen(serviceInfo, converter, this::routeRequest)); - advertiser.startAdvertiser(serviceInfo, converter, block); + // TODO: 18. 11. 19 start multiple service adapter + // TODO: 18. 11. 19 pass network parameter + InetAddress localhost = InetAddress.getLocalHost(); + listen(block, localhost); + } + + /** + * start listening for client connection over given network interface, while advertising service + * @param block if true, this call will block indefinite time, otherwise return immediately + * @param network address of network interface + * @throws IllegalAccessException server 측 네트워크 endpoint 생성의 실패 혹은 I/O 오류 + * @throws IOException the error thrown when {@link ServiceAdapter} fails to resolve dependency object (e.g. negotiator, + * @throws InstantiationException if dependent class represents an abstract class,an interface, an array class, a primitive type, or void;or if the class has no nullary constructor; + */ + public void listen(boolean block, InetAddress network) throws IllegalAccessException, IOException, InstantiationException { + + NetworkInterface networkInterface = NetworkInterface.getByInetAddress(network); + if(!networkInterface.isUp()) { + throw new IOException(String.format(Locale.ENGLISH, "network (%s) is not up", networkInterface.getDisplayName())); + } + if(!networkInterface.supportsMulticast()) { + throw new IOException(String.format(Locale.ENGLISH, "given network (%s) doesn\'t support multicast", networkInterface.getDisplayName())); + } + + serviceInfo.setProxyFactoryHint(adapter.listen(serviceInfo, converter, network, new Function() { + @Override + public Response apply(Request request) throws Exception { + return routeRequest(request); + } + })); + advertiser.startAdvertiser(serviceInfo, block, network); } /** @@ -162,6 +359,7 @@ private Response routeRequest(Request request) throws IllegalAccessException, In } return end(response, request); } catch (InvocationTargetException e) { + Log.error("InvocationError : {}", e); return end(Response.from(RMIError.INTERNAL_SERVER_ERROR), request); } } diff --git a/src/main/java/com/doodream/rmovjs/server/svc/HealthCheckController.java b/src/main/java/com/doodream/rmovjs/server/svc/HealthCheckController.java index 2838820..0b236d0 100644 --- a/src/main/java/com/doodream/rmovjs/server/svc/HealthCheckController.java +++ b/src/main/java/com/doodream/rmovjs/server/svc/HealthCheckController.java @@ -1,6 +1,5 @@ package com.doodream.rmovjs.server.svc; -import com.doodream.rmovjs.Properties; import com.doodream.rmovjs.annotation.method.Get; import com.doodream.rmovjs.model.Response; diff --git a/src/main/java/com/doodream/rmovjs/util/LruCache.java b/src/main/java/com/doodream/rmovjs/util/LruCache.java index 3981e4b..e9a03e6 100644 --- a/src/main/java/com/doodream/rmovjs/util/LruCache.java +++ b/src/main/java/com/doodream/rmovjs/util/LruCache.java @@ -8,7 +8,7 @@ public class LruCache implements Map { private LinkedHashMap internalMap; - public LruCache(int maxObjCount) { + public LruCache(final int maxObjCount) { internalMap = new LinkedHashMap (maxObjCount + 1, .75F, true) { @Override protected boolean removeEldestEntry(Map.Entry eldest) { diff --git a/src/main/java/com/doodream/rmovjs/util/Types.java b/src/main/java/com/doodream/rmovjs/util/Types.java index 217bd7c..6d9d254 100644 --- a/src/main/java/com/doodream/rmovjs/util/Types.java +++ b/src/main/java/com/doodream/rmovjs/util/Types.java @@ -11,8 +11,8 @@ import java.util.regex.Pattern; public class Types { - private static Pattern INTERNAL_TYPE_SELECTOR_PATTERN = Pattern.compile("([^\\<\\>]+)\\<([\\s\\S]+)\\>"); private static final Logger Log = LoggerFactory.getLogger(Types.class); + private static Pattern INTERNAL_TYPE_SELECTOR_PATTERN = Pattern.compile("([^\\<\\>]+)\\<([\\s\\S]+)\\>"); public static Type[] unwrapType(String typeName) throws ClassNotFoundException, IllegalArgumentException { Matcher matcher = INTERNAL_TYPE_SELECTOR_PATTERN.matcher(typeName); @@ -23,10 +23,10 @@ public static Type[] unwrapType(String typeName) throws ClassNotFoundException, List types = new ArrayList<>(); for (String s : split) { try { - Type[] parameters = unwrapType(s); + final Type[] parameters = unwrapType(s); matcher = INTERNAL_TYPE_SELECTOR_PATTERN.matcher(s); if(matcher.matches()) { - Type rawClass = Class.forName(matcher.group(1)); + final Type rawClass = Class.forName(matcher.group(1)); types.add(new ParameterizedType() { @Override public Type[] getActualTypeArguments() { @@ -97,4 +97,42 @@ private static int findNextTypeParameterIndex(String original) { } return 0; } + + public static boolean isCastable(T body, Class rawCls) { + try { + rawCls.cast(body); + return true; + } catch (ClassCastException ignored) { + } + return false; + } + + public static boolean isCastable(T body, Type type) { + try { + ((Class) type).cast(body); + return true; + } catch (ClassCastException ignored) { + } + return false; + + } + + public static Type getType(Class rawCls, Type ...typeParameter) { + return new ParameterizedType() { + @Override + public Type[] getActualTypeArguments() { + return typeParameter; + } + + @Override + public Type getRawType() { + return rawCls; + } + + @Override + public Type getOwnerType() { + return null; + } + }; + } } diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml deleted file mode 100644 index d250d63..0000000 --- a/src/main/resources/log4j2.xml +++ /dev/null @@ -1,16 +0,0 @@ - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml index 4b12a6d..8790d23 100644 --- a/src/main/resources/logback.xml +++ b/src/main/resources/logback.xml @@ -1,7 +1,7 @@ - %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + %d{HH:mm:ss.SSS} [YARMI] [%thread] %-5level %logger{36} - %msg%n diff --git a/src/test/java/com/doodream/rmovjs/test/ConverterTest.java b/src/test/java/com/doodream/rmovjs/test/ConverterTest.java new file mode 100644 index 0000000..21139c7 --- /dev/null +++ b/src/test/java/com/doodream/rmovjs/test/ConverterTest.java @@ -0,0 +1,119 @@ +package com.doodream.rmovjs.test; + + +import com.doodream.rmovjs.model.Response; +import com.doodream.rmovjs.serde.Converter; +import com.doodream.rmovjs.serde.Reader; +import com.doodream.rmovjs.serde.Writer; +import com.doodream.rmovjs.serde.bson.BsonConverter; +import com.doodream.rmovjs.serde.json.JsonConverter; +import com.doodream.rmovjs.test.service.User; +import com.doodream.rmovjs.util.Types; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.lang.reflect.Type; +import java.net.InterfaceAddress; +import java.net.NetworkInterface; +import java.net.SocketException; +import java.util.*; + +public class ConverterTest { + + + private List converters; + + @Before + public void setup() { + converters = Arrays.asList( + new JsonConverter(), + new BsonConverter() + ); + } + + @Test + public void testNetworkInterface() throws SocketException { + List interfaces = Collections.list(NetworkInterface.getNetworkInterfaces()); + Assert.assertNotNull(interfaces); + for (NetworkInterface ifc : interfaces) { + List addresses = ifc.getInterfaceAddresses(); + for (InterfaceAddress address : addresses) { + + System.out.printf("%s : %s(%d)\n", ifc.getDisplayName(), address.getAddress(), address.getNetworkPrefixLength()); + System.out.printf("Broadcast : %s\n", address.getBroadcast()); + } + } + } + + @Test + public void converterSerDeserTest() throws ClassNotFoundException, IOException, InstantiationException, IllegalAccessException { + for (Converter converter : converters) { + Assert.assertTrue(testPrimitiveType(converter, 1.3, double.class)); + Assert.assertTrue(testPrimitiveType(converter, 1, int.class)); + Assert.assertTrue(testNumericObject(converter, 1.3f)); + Assert.assertTrue(testNumericObject(converter, 100L)); + Assert.assertTrue(testNumericObject(converter, 100)); + Assert.assertTrue(testNumericObject(converter, 1.3)); + Assert.assertTrue(testSimpleObject(converter)); + Assert.assertTrue(testGenericObject(converter)); + Assert.assertTrue(testComplexGeneric(converter)); + } + } + + private boolean testPrimitiveType(Converter converter, T v, Class cls) throws ClassNotFoundException, InstantiationException, IllegalAccessException, IOException { + Object resolved = converter.resolve(v, cls); + return resolved.equals(v); + } + + private boolean testNumericObject(Converter converter, T v) throws ClassNotFoundException, InstantiationException, IllegalAccessException, IOException { + T result = testObjectTransfer(converter, v, v.getClass()); + return v.equals(result); + } + + private boolean testSimpleObject(Converter converter) throws IOException, InstantiationException, IllegalAccessException, ClassNotFoundException { + User user = new User(); + user.setAge(30); + User userBody = testObjectTransfer(converter, user, User.class); + return userBody.equals(user); + } + + public boolean testGenericObject(Converter converter) throws ClassNotFoundException, InstantiationException, IllegalAccessException, IOException { + List users = new LinkedList<>(); + User testUser = new User(); + testUser.setAge(30); + testUser.setName("James"); + users.add(testUser); + List usersResult = testObjectTransfer(converter, users, Types.getType(List.class, User.class)); + return users.equals(usersResult); + } + + private boolean testComplexGeneric(Converter converter) throws ClassNotFoundException, InstantiationException, IllegalAccessException, IOException { + List users = new ArrayList<>(); + User user = new User(); + user.setName("david"); + user.setAge(30); + List> userLists = new ArrayList<>(); + userLists.add(users); + + List> userListResult = testObjectTransfer(converter, userLists, Types.getType(List.class, Types.getType(List.class, User.class))); + return userListResult.equals(userLists); + } + + private T testObjectTransfer(Converter converter, T src, Type type) throws IOException, InstantiationException, IllegalAccessException, ClassNotFoundException { + final Response response = Response.success(src); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + Writer writer = converter.writer(baos); + writer.write(response); + ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray()); + Reader reader = converter.reader(bais); + Response parsedResponse = reader.read(Response.class); + if(Types.isCastable(parsedResponse.getBody(), type)) { + return (T) parsedResponse.getBody(); + } + return (T) converter.resolve(parsedResponse.getBody(), type); + } +} diff --git a/src/test/java/com/doodream/rmovjs/test/RMIBasicTest.java b/src/test/java/com/doodream/rmovjs/test/RMIBasicTest.java deleted file mode 100644 index 6938274..0000000 --- a/src/test/java/com/doodream/rmovjs/test/RMIBasicTest.java +++ /dev/null @@ -1,50 +0,0 @@ -package com.doodream.rmovjs.test; - -import com.doodream.rmovjs.net.RMIServiceProxy; -import com.doodream.rmovjs.sdp.ServiceDiscoveryListener; -import com.doodream.rmovjs.sdp.SimpleServiceAdvertiser; -import com.doodream.rmovjs.sdp.SimpleServiceDiscovery; -import com.doodream.rmovjs.server.RMIService; -import com.doodream.rmovjs.test.service.TestService; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; - -import java.io.IOException; - -public class RMIBasicTest { - - private RMIService service; - - @Before - public void setup() throws Exception { - SimpleServiceAdvertiser advertiser = new SimpleServiceAdvertiser(); - service = RMIService.create(TestService.class, advertiser); - service.listen(false); - } - - @After - public void exit() throws Exception { - service.stop(); - } - - @Test - public void createTestClient() throws IOException, IllegalAccessException, InstantiationException { - new SimpleServiceDiscovery().startDiscovery(TestService.class, new ServiceDiscoveryListener() { - @Override - public void onDiscovered(RMIServiceProxy proxy) { - - } - - @Override - public void onDiscoveryStarted() { - - } - - @Override - public void onDiscoveryFinished() { - - } - }); - } -} diff --git a/src/test/java/com/doodream/rmovjs/test/service/User.java b/src/test/java/com/doodream/rmovjs/test/service/User.java index eaf0814..efa43eb 100644 --- a/src/test/java/com/doodream/rmovjs/test/service/User.java +++ b/src/test/java/com/doodream/rmovjs/test/service/User.java @@ -1,5 +1,8 @@ package com.doodream.rmovjs.test.service; +import lombok.Data; + +@Data public class User { String name; int age; diff --git a/src/test/resources/logback.xml b/src/test/resources/logback.xml new file mode 100644 index 0000000..8790d23 --- /dev/null +++ b/src/test/resources/logback.xml @@ -0,0 +1,11 @@ + + + + %d{HH:mm:ss.SSS} [YARMI] [%thread] %-5level %logger{36} - %msg%n + + + + + + + \ No newline at end of file