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

Widening.

Browse Source 
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4524 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2007-02-02 23:40:42 +00:00
parent cadde41ccb
commit 3b2c3c921b
1 changed files with 48 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/peer/server/PeerManager.java
+48 -63
View File
@@ -51,17 +51,16 @@ import com.threerings.presents.peer.server.persist.NodeRepository;
import static com.threerings.presents.Log.log; import static com.threerings.presents.Log.log;
/** /**
* Manages connections to the other nodes in a Presents server cluster. Each * Manages connections to the other nodes in a Presents server cluster. Each server maintains a
* server maintains a client connection to the other servers and subscribes to * client connection to the other servers and subscribes to the {@link NodeObject} of all peer
* the {@link NodeObject} of all peer servers and uses those objects to * servers and uses those objects to communicate cross-node information.
* communicate cross-node information.
*/ */
public class PeerManager public class PeerManager
implements ClientManager.ClientObserver implements ClientManager.ClientObserver
{ {
/** /**
* Creates a peer manager which will create a {@link NodeRepository} which * Creates a peer manager which will create a {@link NodeRepository} which will be used to
* will be used to publish our existence and discover the other nodes. * publish our existence and discover the other nodes.
*/ */
public PeerManager (ConnectionProvider conprov, Invoker invoker) public PeerManager (ConnectionProvider conprov, Invoker invoker)
throws PersistenceException throws PersistenceException
@@ -79,24 +78,21 @@ public class PeerManager
} }
/** /**
* Initializes this peer manager and initiates the process of connecting to * Initializes this peer manager and initiates the process of connecting to its peer
* its peer nodes. This will also reconfigure the ConnectionManager and * nodes. This will also reconfigure the ConnectionManager and ClientManager with peer related
* ClientManager with peer related bits, so this should not be called until * bits, so this should not be called until <em>after</em> the main server has set up its
* <em>after</em> the main server has set up its client factory and * client factory and authenticator.
* authenticator.
* *
* @param nodeName this node's unique name. * @param nodeName this node's unique name.
* @param sharedSecret a shared secret used to allow the peers to * @param sharedSecret a shared secret used to allow the peers to authenticate with one
* authenticate with one another. * another.
* @param hostName the DNS name of the server running this node. * @param hostName the DNS name of the server running this node.
* @param publicHostName if non-null, a separate public DNS hostname by * @param publicHostName if non-null, a separate public DNS hostname by which the node is to be
* which the node is to be known to normal clients (we may want inter-peer * known to normal clients (we may want inter-peer communication to take place over a different
* communication to take place over a different network than the * network than the communication between real clients and the various peer servers).
* communication between real clients and the various peer servers).
* @param port the port on which other nodes should connect to us. * @param port the port on which other nodes should connect to us.
* @param conprov used to obtain our JDBC connections. * @param conprov used to obtain our JDBC connections.
* @param invoker we will perform all database operations on the supplied * @param invoker we will perform all database operations on the supplied invoker thread.
* invoker thread.
*/ */
public void init (String nodeName, String sharedSecret, String hostName, public void init (String nodeName, String sharedSecret, String hostName,
String publicHostName, int port) String publicHostName, int port)
@@ -109,11 +105,9 @@ public class PeerManager
// wire ourselves into the server // wire ourselves into the server
PresentsServer.conmgr.setAuthenticator( PresentsServer.conmgr.setAuthenticator(
new PeerAuthenticator( new PeerAuthenticator(this, PresentsServer.conmgr.getAuthenticator()));
this, PresentsServer.conmgr.getAuthenticator()));
PresentsServer.clmgr.setClientFactory( PresentsServer.clmgr.setClientFactory(
new PeerClientFactory( new PeerClientFactory(this, PresentsServer.clmgr.getClientFactory()));
this, PresentsServer.clmgr.getClientFactory()));
// create our node object // create our node object
_nodeobj = PresentsServer.omgr.registerObject(createNodeObject()); _nodeobj = PresentsServer.omgr.registerObject(createNodeObject());
@@ -121,13 +115,12 @@ public class PeerManager
// register ourselves with the node table // register ourselves with the node table
_invoker.postUnit(new Invoker.Unit() { _invoker.postUnit(new Invoker.Unit() {
public boolean invoke () { public boolean invoke () {
NodeRecord record = new NodeRecord( NodeRecord record = new NodeRecord(_nodeName, _hostName, _publicHostName, _port);
_nodeName, _hostName, _publicHostName, _port);
try { try {
_noderepo.updateNode(record); _noderepo.updateNode(record);
} catch (PersistenceException pe) { } catch (PersistenceException pe) {
log.warning("Failed to register node record " + log.warning("Failed to register node record [rec=" + record +
"[rec=" + record + ", error=" + pe + "]."); ", error=" + pe + "].");
} }
return false; return false;
} }
@@ -136,8 +129,8 @@ public class PeerManager
// register ourselves as a client observer // register ourselves as a client observer
PresentsServer.clmgr.addClientObserver(this); PresentsServer.clmgr.addClientObserver(this);
// and start our peer refresh interval (this need not use a runqueue as // and start our peer refresh interval (this need not use a runqueue as all it will do is
// all it will do is post an invoker unit) // post an invoker unit)
new Interval() { new Interval() {
public void expired () { public void expired () {
refreshPeers(); refreshPeers();
@@ -149,9 +142,8 @@ public class PeerManager
} }
/** /**
* Call this when the server is shutting down to give this node a chance to * Call this when the server is shutting down to give this node a chance to cleanly logoff from
* cleanly logoff from its peers and remove its record from the nodes * its peers and remove its record from the nodes table.
* table.
*/ */
public void shutdown () public void shutdown ()
{ {
@@ -169,8 +161,8 @@ public class PeerManager
*/ */
public boolean isAuthenticPeer (PeerCreds creds) public boolean isAuthenticPeer (PeerCreds creds)
{ {
return PeerCreds.createPassword( return PeerCreds.createPassword(creds.getNodeName(), _sharedSecret).equals(
creds.getNodeName(), _sharedSecret).equals(creds.getPassword()); creds.getPassword());
} }
// documentation inherited from interface ClientManager.ClientObserver // documentation inherited from interface ClientManager.ClientObserver
@@ -187,10 +179,8 @@ public class PeerManager
// sanity check // sanity check
if (_nodeobj.clients.contains(clinfo)) { if (_nodeobj.clients.contains(clinfo)) {
log.warning("Received clientSessionDidStart() for already " + log.warning("Received clientSessionDidStart() for already registered client!? " +
"registered client!? " + "[old=" + _nodeobj.clients.get(clinfo.getKey()) + ", new=" + clinfo + "].");
"[old=" + _nodeobj.clients.get(clinfo.getKey()) +
", new=" + clinfo + "].");
// go ahead and update the record // go ahead and update the record
_nodeobj.updateClients(clinfo); _nodeobj.updateClients(clinfo);
} else { } else {
@@ -206,9 +196,9 @@ public class PeerManager
return; return;
} }
// we scan through the list instead of relying on ClientInfo.getKey() // we scan through the list instead of relying on ClientInfo.getKey() because we want
// because we want derived classes to be able to override that for // derived classes to be able to override that for lookups that happen way more frequently
// lookups that happen way more frequently than logging off // than logging off
Name username = client.getCredentials().getUsername(); Name username = client.getCredentials().getUsername();
for (ClientInfo clinfo : _nodeobj.clients) { for (ClientInfo clinfo : _nodeobj.clients) {
if (clinfo.username.equals(username)) { if (clinfo.username.equals(username)) {
@@ -216,8 +206,7 @@ public class PeerManager
return; return;
} }
} }
log.warning("Session ended for unregistered client " + log.warning("Session ended for unregistered client [who=" + username + "].");
"[who=" + username + "].");
} }
/** /**
@@ -228,8 +217,8 @@ public class PeerManager
} }
/** /**
* Reloads the list of peer nodes from our table and refreshes each with a * Reloads the list of peer nodes from our table and refreshes each with a call to {@link
* call to {@link #refreshPeer}. * #refreshPeer}.
*/ */
protected void refreshPeers () protected void refreshPeers ()
{ {
@@ -253,8 +242,7 @@ public class PeerManager
try { try {
refreshPeer(record); refreshPeer(record);
} catch (Exception e) { } catch (Exception e) {
log.log(Level.WARNING, "Failure refreshing peer " + log.log(Level.WARNING, "Failure refreshing peer " + record + ".", e);
record + ".", e);
} }
} }
} }
@@ -263,8 +251,8 @@ public class PeerManager
} }
/** /**
* Ensures that we have a connection to the specified node if it has * Ensures that we have a connection to the specified node if it has checked in since we last
* checked in since we last failed to connect. * failed to connect.
*/ */
protected void refreshPeer (NodeRecord record) protected void refreshPeer (NodeRecord record)
{ {
@@ -276,8 +264,8 @@ public class PeerManager
} }
/** /**
* Creates the appropriate derived class of {@link NodeObject} which will * Creates the appropriate derived class of {@link NodeObject} which will be registered with
* be registered with the distributed object system. * the distributed object system.
*/ */
protected NodeObject createNodeObject () protected NodeObject createNodeObject ()
{ {
@@ -285,8 +273,8 @@ public class PeerManager
} }
/** /**
* Creates a {@link ClientInfo} record which will subsequently be * Creates a {@link ClientInfo} record which will subsequently be initialized by a call to
* initialized by a call to {@link #initClientInfo}. * {@link #initClientInfo}.
*/ */
protected ClientInfo createClientInfo () protected ClientInfo createClientInfo ()
{ {
@@ -302,8 +290,7 @@ public class PeerManager
} }
/** /**
* Creates a {@link PeerNode} to manage our connection to the specified * Creates a {@link PeerNode} to manage our connection to the specified peer.
* peer.
*/ */
protected PeerNode createPeerNode (NodeRecord record) protected PeerNode createPeerNode (NodeRecord record)
{ {
@@ -333,8 +320,8 @@ public class PeerManager
public void refresh (NodeRecord record) public void refresh (NodeRecord record)
{ {
// if the hostname of this node changed, kill our existing client // if the hostname of this node changed, kill our existing client connection and
// connection and connect anew // connect anew
if (!record.hostName.equals(_record.hostName) && if (!record.hostName.equals(_record.hostName) &&
_client.isActive()) { _client.isActive()) {
_client.logoff(false); _client.logoff(false);
@@ -345,8 +332,8 @@ public class PeerManager
return; return;
} }
// if our client hasn't updated its record since we last tried to // if our client hasn't updated its record since we last tried to logon, then just
// logon, then just chill // chill
if (_lastConnectStamp > record.lastUpdated.getTime()) { if (_lastConnectStamp > record.lastUpdated.getTime()) {
log.fine("Not reconnecting to stale client [record=" + _record + log.fine("Not reconnecting to stale client [record=" + _record +
", lastTry=" + new Date(_lastConnectStamp) + "]."); ", lastTry=" + new Date(_lastConnectStamp) + "].");
@@ -354,8 +341,7 @@ public class PeerManager
} }
// otherwise configure our client with the right bits and logon // otherwise configure our client with the right bits and logon
_client.setCredentials( _client.setCredentials(new PeerCreds(_record.nodeName, _sharedSecret));
new PeerCreds(_record.nodeName, _sharedSecret));
_client.setServer(record.hostName, new int[] { _record.port }); _client.setServer(record.hostName, new int[] { _record.port });
_client.logon(); _client.logon();
_lastConnectStamp = System.currentTimeMillis(); _lastConnectStamp = System.currentTimeMillis();
@@ -388,8 +374,7 @@ public class PeerManager
log.info("Connected to peer " + _record + "."); log.info("Connected to peer " + _record + ".");
// subscribe to this peer's node object // subscribe to this peer's node object
PeerBootstrapData pdata = (PeerBootstrapData) PeerBootstrapData pdata = (PeerBootstrapData)client.getBootstrapData();
client.getBootstrapData();
client.getDObjectManager().subscribeToObject(pdata.nodeOid, this); client.getDObjectManager().subscribeToObject(pdata.nodeOid, this);
} }