apache/cloudstack
1
0
Fork 0
mirror of https://github.com/apache/cloudstack.git synced 2025-10-26 08:42:29 +01:00
Code Issues Packages Projects Releases Wiki Activity

Modify RPC callback style to be in consist with the latest AsyncMethod callback style

Browse Source
This commit is contained in:
Kelven Yang 2012-12-18 17:09:56 -08:00
parent e6ac3fd199
commit 644b783110
6 changed files with 59 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
framework/ipc/src/org/apache/cloudstack/framework/async/AsyncCallbackHandler.java
View File

@ -1,30 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cloudstack.framework.async;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface AsyncCallbackHandler {
String operationName();
}

82
framework/ipc/src/org/apache/cloudstack/framework/rpc/RpcCallbackDispatcher.java
View File

@ -20,24 +20,47 @@ package org.apache.cloudstack.framework.rpc;
import java.lang.reflect.InvocationTargetException; import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method; import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import net.sf.cglib.proxy.Enhancer;
import net.sf.cglib.proxy.MethodInterceptor;
import net.sf.cglib.proxy.MethodProxy;
public class RpcCallbackDispatcher { public class RpcCallbackDispatcher<T> {
private Method _callbackMethod;
private static Map<Class<?>, Map<String, Method>> s_handlerCache = new HashMap<Class<?>, Map<String, Method>>(); private T _targetObject;
public static boolean dispatch(Object target, RpcClientCall clientCall) { private RpcCallbackDispatcher(T target) {
_targetObject = target;
}
@SuppressWarnings("unchecked")
public T getTarget() {
return (T)Enhancer.create(_targetObject.getClass(), new MethodInterceptor() {
@Override
public Object intercept(Object arg0, Method arg1, Object[] arg2,
MethodProxy arg3) throws Throwable {
_callbackMethod = arg1;
return null;
}
});
}
public RpcCallbackDispatcher<T> setCallback(Object useless) {
return this;
}
public static <P> RpcCallbackDispatcher<P> create(P target) {
return new RpcCallbackDispatcher<P>(target);
}
public boolean dispatch(RpcClientCall clientCall) {
assert(clientCall != null); assert(clientCall != null);
assert(target != null);
if(_callbackMethod == null)
Method handler = resolveHandler(target.getClass(), clientCall.getCommand());
if(handler == null)
return false; return false;
try { try {
handler.invoke(target, clientCall); _callbackMethod.invoke(_targetObject, clientCall, clientCall.getContext());
} catch (IllegalArgumentException e) { } catch (IllegalArgumentException e) {
throw new RpcException("IllegalArgumentException when invoking RPC callback for command: " + clientCall.getCommand()); throw new RpcException("IllegalArgumentException when invoking RPC callback for command: " + clientCall.getCommand());
} catch (IllegalAccessException e) { } catch (IllegalAccessException e) {
@ -48,41 +71,4 @@ public class RpcCallbackDispatcher {
return true; return true;
} }
public static Method resolveHandler(Class<?> handlerClz, String command) {
synchronized(s_handlerCache) {
Map<String, Method> handlerMap = getAndSetHandlerMap(handlerClz);
Method handler = handlerMap.get(command);
if(handler != null)
return handler;
for(Method method : handlerClz.getDeclaredMethods()) {
RpcCallbackHandler annotation = method.getAnnotation(RpcCallbackHandler.class);
if(annotation != null) {
if(annotation.command().equals(command)) {
method.setAccessible(true);
handlerMap.put(command, method);
return method;
}
}
}
}
return null;
}
private static Map<String, Method> getAndSetHandlerMap(Class<?> handlerClz) {
Map<String, Method> handlerMap;
synchronized(s_handlerCache) {
handlerMap = s_handlerCache.get(handlerClz);
if(handlerMap == null) {
handlerMap = new HashMap<String, Method>();
s_handlerCache.put(handlerClz, handlerMap);
}
}
return handlerMap;
}
} }

30
framework/ipc/src/org/apache/cloudstack/framework/rpc/RpcCallbackHandler.java
View File

@ -1,30 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cloudstack.framework.rpc;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RpcCallbackHandler {
String command();
}

6
framework/ipc/src/org/apache/cloudstack/framework/rpc/RpcClientCall.java
View File

@ -29,11 +29,11 @@ public interface RpcClientCall {
RpcClientCall setCommandArg(Object arg); RpcClientCall setCommandArg(Object arg);
Object getCommandArg(); Object getCommandArg();
RpcClientCall setContextParam(String key, Object param); RpcClientCall setContext(Object param);
<T> T getContextParam(String key); <T> T getContext();
<T> RpcClientCall addCallbackListener(RpcCallbackListener<T> listener); <T> RpcClientCall addCallbackListener(RpcCallbackListener<T> listener);
RpcClientCall setCallbackDispatcherTarget(Object target); RpcClientCall setCallbackDispatcher(RpcCallbackDispatcher dispatcher);
RpcClientCall setOneway(); RpcClientCall setOneway();

29
framework/ipc/src/org/apache/cloudstack/framework/rpc/RpcClientCallImpl.java
View File

@ -23,19 +23,20 @@ import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
public class RpcClientCallImpl implements RpcClientCall { public class RpcClientCallImpl implements RpcClientCall {
private String _command; private String _command;
private Object _commandArg; private Object _commandArg;
private int _timeoutMilliseconds = DEFAULT_RPC_TIMEOUT; private int _timeoutMilliseconds = DEFAULT_RPC_TIMEOUT;
private Map<String, Object> _contextParams = new HashMap<String, Object>(); private Object _contextObject;
private boolean _oneway = false; private boolean _oneway = false;
@SuppressWarnings("rawtypes") @SuppressWarnings("rawtypes")
private List<RpcCallbackListener> _callbackListeners = new ArrayList<RpcCallbackListener>(); private List<RpcCallbackListener> _callbackListeners = new ArrayList<RpcCallbackListener>();
private Object _callbackDispatcherTarget;
@SuppressWarnings("rawtypes")
private RpcCallbackDispatcher _callbackDispatcher;
private RpcProvider _rpcProvider; private RpcProvider _rpcProvider;
private long _startTickInMs; private long _startTickInMs;
@ -81,16 +82,15 @@ public class RpcClientCallImpl implements RpcClientCall {
} }
@Override @Override
public RpcClientCall setContextParam(String key, Object param) { public RpcClientCall setContext(Object param) {
assert(key != null); _contextObject = param;
_contextParams.put(key, param);
return this; return this;
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Override @Override
public <T> T getContextParam(String key) { public <T> T getContext() {
return (T)_contextParams.get(key); return (T)_contextObject;
} }
@Override @Override
@ -101,11 +101,10 @@ public class RpcClientCallImpl implements RpcClientCall {
} }
@Override @Override
public RpcClientCall setCallbackDispatcherTarget(Object target) { public RpcClientCall setCallbackDispatcher(RpcCallbackDispatcher dispatcher) {
_callbackDispatcherTarget = target; _callbackDispatcher = dispatcher;
return this; return this;
} }
@Override @Override
public RpcClientCall setOneway() { public RpcClientCall setOneway() {
@ -210,8 +209,8 @@ public class RpcClientCallImpl implements RpcClientCall {
for(@SuppressWarnings("rawtypes") RpcCallbackListener listener: _callbackListeners) for(@SuppressWarnings("rawtypes") RpcCallbackListener listener: _callbackListeners)
listener.onSuccess(resultObject); listener.onSuccess(resultObject);
} else { } else {
if(_callbackDispatcherTarget != null) if(_callbackDispatcher != null)
RpcCallbackDispatcher.dispatch(_callbackDispatcherTarget, this); _callbackDispatcher.dispatch(this);
} }
} }
@ -228,8 +227,8 @@ public class RpcClientCallImpl implements RpcClientCall {
for(@SuppressWarnings("rawtypes") RpcCallbackListener listener: _callbackListeners) for(@SuppressWarnings("rawtypes") RpcCallbackListener listener: _callbackListeners)
listener.onFailure(e); listener.onFailure(e);
} else { } else {
if(_callbackDispatcherTarget != null) if(_callbackDispatcher != null)
RpcCallbackDispatcher.dispatch(_callbackDispatcherTarget, this); _callbackDispatcher.dispatch(this);
} }
} }
} }

15
framework/ipc/test/org/apache/cloudstack/framework/codestyle/ClientOnlyEventDrivenStyle.java
View File

@ -18,7 +18,7 @@
*/ */
package org.apache.cloudstack.framework.codestyle; package org.apache.cloudstack.framework.codestyle;
import org.apache.cloudstack.framework.rpc.RpcCallbackHandler; import org.apache.cloudstack.framework.rpc.RpcCallbackDispatcher;
import org.apache.cloudstack.framework.rpc.RpcClientCall; import org.apache.cloudstack.framework.rpc.RpcClientCall;
import org.apache.cloudstack.framework.rpc.RpcException; import org.apache.cloudstack.framework.rpc.RpcException;
import org.apache.cloudstack.framework.rpc.RpcIOException; import org.apache.cloudstack.framework.rpc.RpcIOException;
@ -30,17 +30,16 @@ public class ClientOnlyEventDrivenStyle {
public void AsyncCallRpcService() { public void AsyncCallRpcService() {
String cmd = new String(); String cmd = new String();
RpcCallbackDispatcher<ClientOnlyEventDrivenStyle> callbackDispatcher = RpcCallbackDispatcher.create(this);
callbackDispatcher.setCallback(callbackDispatcher.getTarget().OnAsyncCallRpcServiceCallback(null, null));
_rpcProvider.newCall("host-2").setCommand("TestCommand").setCommandArg(cmd).setTimeout(10000) _rpcProvider.newCall("host-2").setCommand("TestCommand").setCommandArg(cmd).setTimeout(10000)
.setCallbackDispatcherTarget(this) .setCallbackDispatcher(callbackDispatcher)
.setContextParam("origCmd", cmd) // save context object for callback handler .setContext("Context Object") // save context object for callback handler
.apply(); .apply();
} }
@RpcCallbackHandler(command="TestCommand") public Void OnAsyncCallRpcServiceCallback(RpcClientCall call, String context) {
public void OnAsyncCallRpcServiceCallback(RpcClientCall call) {
try { try {
String origCmd = call.getContextParam("origCmd"); // restore calling context at callback handler
String answer = call.get(); String answer = call.get();
} catch(RpcTimeoutException e) { } catch(RpcTimeoutException e) {
@ -49,5 +48,7 @@ public class ClientOnlyEventDrivenStyle {
} catch(RpcException e) { } catch(RpcException e) {
} }
return null;
} }
} }