RpcTreeCache


1   package org.jboss.cache.rpc;
2   
3   import org.jboss.cache.CacheImpl;
4   import org.jboss.cache.config.Configuration;
5   import org.jboss.cache.marshall.MethodCall;
6   import org.jboss.cache.marshall.MethodCallFactory;
7   import org.jgroups.JChannel;
8   
9   import java.lang.reflect.Method  ;
10  import java.util.ArrayList  ;
11  import java.util.HashMap  ;
12  import java.util.List  ;
13  import java.util.Map  ;
14  import java.util.Vector  ;
15  
16  /**
17   * {@link CacheImpl} extension that adds a general purpose RPC functionality
18   * to allow clients to make/receive RPC calls over the same JGroups Channel
19   * used by the cache.
20   * <p/>
21   * Services wishing to receive remote calls should register a unique service
22   * name and an object on which the remote calls for that service can be invoked.
23   * </p>
24   * <p/>
25   * Clients wishing to make RPC calls need to know the unique service name, which
26   * they can pass to one of the flavors of <code>callRemoteMethods</code>.
27   * </p>
28   * <p/>
29   * <strong>NOTE: </strong> The purpose of this class is to allow services that
30   * want to use a CacheImpl to avoid also having to use a HAPartition (and thus
31   * potentially requiring a duplicate JGroups Channel).
32   *
33   * @author <a HREF="mailto://brian.stansberry@jboss.com">Brian Stansberry</a>
34   * @version $Revision$
35   * @deprecated This class will be removed when JGroups adds a multiplexing
36   *             capability.
37   */
38  public class RpcTreeCache extends CacheImpl//implements RpcTreeCacheMBean // TODO; Sort out proper JMX interfaces
39  {
40  
41     /**
42      * The {@link #_dispatchRpcCall(String,MethodCall)} method
43      */
44     public static final Method   dispatchRpcCallMethod;
45  
46     static
47     {
48        try
49        {
50           dispatchRpcCallMethod = RpcTreeCache.class.getDeclaredMethod("_dispatchRpcCall", String  .class, MethodCall.class);
51        }
52        catch (NoSuchMethodException   ex)
53        {
54           ex.printStackTrace();
55           throw new ExceptionInInitializerError  (ex.toString());
56        }
57     }
58  
59     /**
60      * <code>Map</code> of registered RPC handlers
61      */
62     protected Map   rpcHandlers = new HashMap  ();
63  
64     /**
65      * Creates a channel with the given properties. Connects to the channel, then creates a PullPushAdapter
66      * and starts it
67      */
68     public RpcTreeCache(String   cluster_name, String   props, long state_fetch_timeout) throws Exception  
69     {
70        //super(cluster_name, props, state_fetch_timeout);
71        // TODO; Use factory to create cache
72     }
73  
74     /**
75      * Default constructor.
76      *
77      * @throws Exception
78      */
79     public RpcTreeCache() throws Exception  
80     {
81        super();
82     }
83  
84     /**
85      * Expects an already connected channel. Creates a PullPushAdapter and starts it
86      */
87     public RpcTreeCache(JChannel channel) throws Exception  
88     {
89        super(channel);
90     }
91  
92     /**
93      * Calls a remote method on nodes in the cluster, targeted at
94      * objects registered under a given <code>serviceName</code>.
95      *
96      * @param serviceName  name of a callback handler that will have been
97      *                     registered on the remote end via
98      *                     {@link #registerRPCHandler(String,Object)}.
99      * @param members      Vector, each of whose members is the Address of one
100     *                     the nodes in the cache's
101     *                     {@link CacheImpl#getMembers() member list}.
102     *                     If <code>null</code>, the method will be invoked on
103     *                     all members.
104     * @param method       method to execute
105     * @param args         method arguments
106     * @param synchronous  <code>true</code> if the call should block until
107     *                     all members respond (or timeout); <code>false</code>
108     *                     if the call should return immediately without
109     *                     waiting for responses
110     * @param exclude_self should the call be invoked on the callee?
111     * @param timeout      how long to wait for synchronous responses
112     * @return List containing the responses that were received, or
113     *         <code>null</code> if the call is asynchronous.
114     *         Elements of the list will either be a returned value
115     *         or an exception if one was returned.  Any
116     *         NoHandlerForRPCException returned will be removed.
117     * @throws Exception
118     */
119    public List   callRemoteMethods(String   serviceName, Vector   members, Method   method, Object  [] args, boolean synchronous, boolean exclude_self, long timeout) throws Exception  
120    {
121 
122       return callRemoteMethods(serviceName, members, MethodCallFactory.create(method, args), synchronous, exclude_self, timeout);
123    }
124 
125    /**
126     * Calls a remote method on nodes in the cluster, targeted at
127     * objects registered under a given <code>serviceName</code>.
128     * <p/>
129     * If the cache's <code>cache mode</code> is <code>CacheImpl.LOCAL</code>
130     * and parameter <code>exclude_self</code> is <code>false</code>
131     * this request will be made directly to
132     * </p>
133     *
134     * @param serviceName  name of a callback handler that will have been
135     *                     registered on the remote end via
136     *                     {@link #registerRPCHandler(String,Object)}.
137     * @param method_call  method call to execute
138     * @param synchronous  <code>true</code> if the call should block until
139     *                     all members respond (or timeout); <code>false</code>
140     *                     if the call should return immediately without
141     *                     waiting for responses
142     * @param exclude_self should the call be invoked on the callee?
143     * @param timeout      how long to wait for synchronous responses
144     * @return List containing the responses that were received, or
145     *         <code>null</code> if the call is asynchronous.
146     *         Elements of the list will either be a returned value
147     *         or an exception if one was returned.  Any
148     *         NoHandlerForRPCException returned will be removed.
149     * @throws Exception
150     */
151    public List   callRemoteMethods(String   serviceName, Vector   mbrs,
152                                  MethodCall method_call, boolean synchronous,
153                                  boolean exclude_self, long timeout)
154            throws Exception  
155    {
156       List   responses = null;
157 
158       if (getConfiguration().getCacheMode() == Configuration.CacheMode.LOCAL)
159       {
160          // If cache mode is local, we have no channel
161          // so we have to make local calls to our registered handler
162 
163          if (synchronous)
164          {
165             // For synchronous calls we have to return something
166             responses = new ArrayList  ();
167 
168             if (!exclude_self)
169             {
170                // Make the call locally and add a valid response to result list
171                Object   resp = _dispatchRpcCall(serviceName, method_call);
172                if ((resp instanceof NoHandlerForRPCException) == false)
173                {
174                   responses.add(_dispatchRpcCall(serviceName, method_call));
175                }
176             }
177             // else just return an empty list
178 
179          }
180          else if (!exclude_self)
181          {
182             // Asynchronous, so we don't return anything,
183             // but we still want to make the local call
184             _dispatchRpcCall(serviceName, method_call);
185          }
186       }
187       else
188       {
189          // Cache mode is not LOCAL
190          // Need to make a call on the cluster
191 
192          // Wrap the ultimate target in a MethodCall pointing at 
193          // the _dispatchRpcCall method
194          MethodCall wrapper = MethodCallFactory.create(dispatchRpcCallMethod,
195                  serviceName, method_call);
196 
197          responses = callRemoteMethods(mbrs, wrapper, synchronous,
198                  exclude_self, timeout);
199 
200          // Remove any NoHandlerForRPCException
201          // Its inefficient doing this here, but if we add it to
202          // CacheImpl.callRemoteMethods we slow down normal cache ops
203          if (responses != null)
204          {
205             for (int i = 0; i < responses.size(); i++)
206             {
207                Object   obj = responses.get(i);
208                if (obj instanceof NoHandlerForRPCException)
209                {
210                   responses.remove(i);
211                   i--;
212                }
213             }
214          }
215       }
216 
217       return responses;
218    }
219 
220    /**
221     * Calls a remote method on nodes in the cluster, targeted at
222     * objects registered under a given <code>serviceName</code>.
223     *
224     * @param serviceName  name of a callback handler that will have been
225     *                     registered on the remote end via
226     *                     {@link #registerRPCHandler(String,Object)}.
227     * @param members      Vector, each of whose members is the Address of one
228     *                     the nodes in the cache's
229     *                     {@link CacheImpl#getMembers() member list}.
230     *                     If <code>null</code>, the method will be invoked on
231     *                     all members.
232     * @param method_name  name of the method to execute
233     * @param args         method arguments
234     * @param synchronous  <code>true</code> if the call should block until
235     *                     all members respond (or timeout); <code>false</code>
236     *                     if the call should return immediately without
237     *                     waiting for responses
238     * @param exclude_self should the call be invoked on the callee?
239     * @param timeout      how long to wait for synchronous responses
240     * @return List containing the responses that were received, or
241     *         <code>null</code> if the call is asynchronous.
242     *         Elements of the list will either be a returned value
243     *         or an exception if one was returned.  Any
244     *         NoHandlerForRPCException returned will be removed.
245     * @throws NoHandlerForRPCException if no handler is registered on this node
246     *                                  under <code>serviceName</code>.
247     * @throws Exception
248     */
249    public List   callRemoteMethods(String   serviceName, Vector   members,
250                                  String   method_name, Class  [] types, Object  [] args, boolean synchronous,
251                                  boolean exclude_self, long timeout) throws Exception  
252    {
253       Object   handler = rpcHandlers.get(serviceName);
254       if (handler == null)
255       {
256          String   msg = "No rpc handler registered under: " + serviceName;
257 
258          log.trace(msg);
259 
260          throw new NoHandlerForRPCException(msg);
261       }
262 
263       Method   method = handler.getClass().getDeclaredMethod(method_name, types);
264       return callRemoteMethods(serviceName, members, method, args,
265               synchronous, exclude_self, timeout);
266    }
267 
268 
269    /**
270     * Registers the given object as the on which any MethodCall associated with
271     * the given service should be invoked.
272     *
273     * @param serviceName name of a service that will be receiving RPC calls
274     * @param handler     object on which RPC calls for
275     *                    <code>serviceName</code> can be invoked.
276     * @see #_dispatchRpcCall
277     */
278    public void registerRPCHandler(String   serviceName, Object   handler)
279    {
280       rpcHandlers.put(serviceName, handler);
281    }
282 
283    /**
284     * Removes the given object as a handler for RPC calls for the given
285     * service.
286     *
287     * @param serviceName name of a service that will be receiving RPC calls
288     * @param handler     object that was previously passed to
289     *                    {@link #registerRPCHandler(String,Object)} for
290     *                    <code>serviceName</code>.
291     * @param handler
292     */
293    public void unregisterRPCHandler(String   serviceName, Object   handler)
294    {
295       Object   registered = rpcHandlers.remove(serviceName);
296       if (handler != registered)
297       {
298          // Put it back
299          rpcHandlers.put(serviceName, handler);
300       }
301    }
302 
303    /**
304     * Looks up the RPC handler for <code>serviceName</code> and invokes
305     * the method call on it.
306     *
307     * @param serviceName the service
308     * @param call        the call to invoke
309     * @return the result of the call, or <code>NoHandlerForRPCException</code>
310     *         if no handler was registered for
311     */
312    public Object   _dispatchRpcCall(String   serviceName, MethodCall call)
313    {
314       Object   retval = null;
315       Object   handler = rpcHandlers.get(serviceName);
316       if (handler == null)
317       {
318          String   msg = "No rpc handler registered under: " + serviceName;
319 
320          log.trace(msg);
321 
322          return new NoHandlerForRPCException(msg, getLocalAddress());
323       }
324 
325       try
326       {
327          retval = call.invoke(handler);
328       }
329       catch (Throwable   t)
330       {
331          log.trace("rpc call threw exception", t);
332          retval = t;
333       }
334 
335       return retval;
336    }
337 
338 
339 }
340
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags