AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Break out a method to find all applicable nodes for a NodeRequest, and add a class to hold a node name and an oid to fully address a DObject in a peered environment.

Browse Source 
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6476 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Charlie Groves
2011-02-04 04:46:17 +00:00
parent 5b6c9a71e5
commit 540f84194d
2 changed files with 108 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/threerings/presents/peer/data/DObjectAddress.java
+61
View File
@@ -0,0 +1,61 @@
//
// $Id$
//
// Narya library - tools for developing networked games
// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved
// http://code.google.com/p/narya/
//
// This library is free software; you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published
// by the Free Software Foundation; either version 2.1 of the License, or
// (at your option) any later version.
//
// This library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
// Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
package com.threerings.presents.peer.data;
import com.google.common.base.Objects;
import com.threerings.io.SimpleStreamableObject;
/**
* Identifies a DObject on a peer.
*/
public class DObjectAddress extends SimpleStreamableObject
{
public DObjectAddress () {} // Deserialization
public DObjectAddress (String nodeName, int oid)
{
this.nodeName = nodeName;
this.oid = oid;
}
public String nodeName;
public int oid;
@Override
public int hashCode ()
{
return 31 * Objects.hashCode(nodeName) + oid;
}
@Override
public boolean equals (Object obj)
{
if (!(obj instanceof DObjectAddress)) {
return false;
}
DObjectAddress o = (DObjectAddress)obj;
return oid == o.oid
&& (nodeName == o.nodeName || (nodeName != null && nodeName.equals(o.nodeName)));
}
}
src/main/java/com/threerings/presents/peer/server/PeerManager.java
+47 -21
View File
@@ -55,6 +55,8 @@ import com.threerings.io.ObjectInputStream;
import com.threerings.io.ObjectOutputStream;
import com.threerings.io.Streamable;
import com.threerings.presents.peer.data.DObjectAddress;
import com.threerings.util.Name;
import com.threerings.presents.annotation.MainInvoker;
@@ -488,15 +490,7 @@ public abstract class PeerManager
byte[] requestBytes = flattenRequest(request);
// build a set of node names (including the local node) to which to send the request
final Set<String> nodes = Sets.newHashSet();
if (request.isApplicable(_nodeobj)) {
nodes.add(_nodeobj.nodeName);
}
for (PeerNode peer : _peers.values()) {
if (request.isApplicable(peer.nodeobj)) {
nodes.add(peer.getNodeName());
}
}
final Set<String> nodes = findApplicableNodes(request);
if (nodes.isEmpty()) {
listener.requestsProcessed(new NodeRequestsResultImpl<T>());
return;
@@ -512,13 +506,13 @@ public abstract class PeerManager
@SuppressWarnings("unchecked")
T castResult = (T) result;
results.put(node, castResult);
nodeDone(node);
nodeDone();
}
public void requestFailed (String cause) {
failures.put(node, cause);
nodeDone(node);
nodeDone();
}
protected void nodeDone (String node) {
protected void nodeDone () {
if (completedNodes.incrementAndGet() == nodes.size()) {
// if all nodes have responded, let caller know
listener.requestsProcessed(new NodeRequestsResultImpl<T>(results, failures));
@@ -528,6 +522,23 @@ public abstract class PeerManager
}
}
/**
* Returns all nodes for which <code>request.isApplicable</code> returns true.
*/
public Set<String> findApplicableNodes (NodeRequest request)
{
Set<String> nodes = Sets.newHashSet();
if (request.isApplicable(_nodeobj)) {
nodes.add(_nodeobj.nodeName);
}
for (PeerNode peer : _peers.values()) {
if (request.isApplicable(peer.nodeobj)) {
nodes.add(peer.getNodeName());
}
}
return nodes;
}
/**
* Invokes a node request on a specific node and returns the result through the listener.
*/
@@ -537,6 +548,7 @@ public abstract class PeerManager
invokeNodeRequest(nodeName, flattenRequest(request), listener);
}
/**
* Initiates a proxy on an object that is managed by the specified peer. The object will be
* proxied into this server's distributed object space and its local oid reported back to the
@@ -548,27 +560,41 @@ public abstract class PeerManager
* object.
*/
public <T extends DObject> void proxyRemoteObject (
String nodeName, int remoteOid, final ResultListener<Integer> listener)
String nodeName, final int remoteOid, final ResultListener<Integer> listener)
{
final Client peer = getPeerClient(nodeName);
proxyRemoteObject(new DObjectAddress(nodeName, remoteOid), listener);
}
public <T extends DObject> void proxyRemoteObject (
final DObjectAddress remote, final ResultListener<Integer> listener)
{
if (remote.nodeName.equals(_nodeName)) {
System.out.println("Found locally!");
_omgr.postRunnable(new Runnable() {
public void run () {
listener.requestCompleted(remote.oid);
}
});
return;
}
final Client peer = getPeerClient(remote.nodeName);
if (peer == null) {
String errmsg = "Have no connection to peer [node=" + nodeName + "].";
String errmsg = "Have no connection to peer [node=" + remote.nodeName + "].";
listener.requestFailed(new ObjectAccessException(errmsg));
return;
}
final Tuple<String, Integer> key = Tuple.newTuple(nodeName, remoteOid);
if (_proxies.containsKey(key)) {
String errmsg = "Cannot proxy already proxied object [key=" + key + "].";
if (_proxies.containsKey(remote)) {
String errmsg = "Cannot proxy already proxied object [key=" + remote + "].";
listener.requestFailed(new ObjectAccessException(errmsg));
return;
}
// issue a request to subscribe to the remote object
peer.getDObjectManager().subscribeToObject(remoteOid, new Subscriber<T>() {
peer.getDObjectManager().subscribeToObject(remote.oid, new Subscriber<T>() {
public void objectAvailable (T object) {
// make a note of this proxy mapping
_proxies.put(key, new Tuple<Subscriber<?>, DObject>(this, object));
_proxies.put(remote, new Tuple<Subscriber<?>, DObject>(this, object));
// map the object into our local oid space
_omgr.registerProxyObject(object, peer.getDObjectManager());
// then tell the caller about the (now remapped) oid
@@ -1581,7 +1607,7 @@ public abstract class PeerManager
protected ArrayIntSet _suboids = new ArrayIntSet();
/** Contains a mapping of proxied objects to subscriber instances. */
protected Map<Tuple<String,Integer>,Tuple<Subscriber<?>,DObject>> _proxies = Maps.newHashMap();
protected Map<DObjectAddress, Tuple<Subscriber<?>, DObject>> _proxies = Maps.newHashMap();
/** Our stale cache observers. */
protected Map<String, ObserverList<StaleCacheObserver>> _cacheobs = Maps.newHashMap();