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

Created a peer-aware DatabaseConfigRegistry that broadcasts changed runtime

Browse Source 
configuration information to other peers so that they all remain in sync.

Also nixed constructor arguments to the peer manager so that it can be created
at server construct time like all other managers and be available for the
twisty maze of inter-registration that takes place during the manager init
process.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4814 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2007-08-16 20:53:17 +00:00
parent b18ebdbc53
commit f90a39207f
5 changed files with 159 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/admin/Log.java
+8 -6
View File
@@ -21,19 +21,21 @@
package com.threerings.admin;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
* A placeholder class that contains a reference to the log object used by
* the admin services.
* Contains a reference to the log object used by the Admin services.
*/
public class Log
{
public static com.samskivert.util.Log log =
new com.samskivert.util.Log("admin");
/** We dispatch our log messages through this logger. */
public static Logger log = Logger.getLogger("com.threerings.narya.admin");
/** Convenience function. */
public static void debug (String message)
{
log.debug(message);
log.fine(message);
}
/** Convenience function. */
@@ -51,6 +53,6 @@ public class Log
/** Convenience function. */
public static void logStackTrace (Throwable t)
{
log.logStackTrace(com.samskivert.util.Log.WARNING, t);
log.log(Level.WARNING, t.getMessage(), t);
}
}
src/java/com/threerings/admin/server/ConfigRegistry.java
+20 -13
View File
@@ -147,21 +147,25 @@ public abstract class ConfigRegistry
}
}
// from SetListener
public void entryAdded (EntryAddedEvent event)
{
serializeAttribute(event.getName());
}
// from SetListener
public void entryUpdated (EntryUpdatedEvent event)
{
serializeAttribute(event.getName());
}
// from SetListener
public void entryRemoved (EntryRemovedEvent event)
{
serializeAttribute(event.getName());
}
// from ElementUpdateListener
public void elementUpdated (ElementUpdatedEvent event)
{
Object value;
@@ -175,6 +179,7 @@ public abstract class ConfigRegistry
updateValue(event.getName(), value);
}
// from AttributeChangeListener
public void attributeChanged (AttributeChangedEvent event)
{
// mirror this configuration update to the persistent config
@@ -216,8 +221,10 @@ public abstract class ConfigRegistry
}
}
/** Initializes a single field of a config distributed object from its corresponding value
* in the associated config repository. */
/**
* Initializes a single field of a config distributed object from its corresponding value
* in the associated config repository.
*/
protected void initField (Field field)
{
String key = nameToKey(field.getName());
@@ -293,15 +300,6 @@ public abstract class ConfigRegistry
}
}
/**
* Converts a config object field name (someConfigMember) to a configuration key
* (some_config_member).
*/
protected String nameToKey (String attributeName)
{
return StringUtil.unStudlyName(attributeName).toLowerCase();
}
/**
* Get the specified attribute from the configuration object, and serialize it.
*/
@@ -318,7 +316,7 @@ public abstract class ConfigRegistry
}
if (value instanceof Streamable) {
serialize(key, value);
serialize(attributeName, key, value);
} else {
Log.info("Unable to flush config obj change [cobj=" + object.getClass().getName() +
", key=" + key + ", type=" + value.getClass().getName() +
@@ -329,7 +327,7 @@ public abstract class ConfigRegistry
/**
* Save the specified object as serialized data associated with the specified key.
*/
protected void serialize (String key, Object value)
protected void serialize (String name, String key, Object value)
{
ByteArrayOutInputStream out = new ByteArrayOutInputStream();
ObjectOutputStream oout = new ObjectOutputStream(out);
@@ -342,6 +340,15 @@ public abstract class ConfigRegistry
}
}
/**
* Converts a config object field name (someConfigMember) to a configuration key
* (some_config_member).
*/
protected String nameToKey (String attributeName)
{
return StringUtil.unStudlyName(attributeName).toLowerCase();
}
protected abstract boolean getValue (String field, boolean defval);
protected abstract short getValue (String field, short defval);
protected abstract int getValue (String field, int defval);
src/java/com/threerings/admin/server/PeeredDatabaseConfigRegistry.java
+124
View File
@@ -0,0 +1,124 @@
//
// $Id$
//
// Narya library - tools for developing networked games
// Copyright (C) 2002-2007 Three Rings Design, Inc., All Rights Reserved
// http://www.threerings.net/code/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.admin.server;
import java.util.ArrayList;
import java.util.logging.Level;
import com.samskivert.io.PersistenceException;
import com.samskivert.jdbc.depot.PersistenceContext;
import com.samskivert.util.Invoker;
import com.threerings.io.SimpleStreamableObject;
import com.threerings.io.Streamable;
import com.threerings.util.StreamableTuple;
import com.threerings.presents.dobj.AttributeChangedEvent;
import com.threerings.presents.dobj.DObject;
import com.threerings.presents.peer.server.PeerManager;
import static com.threerings.admin.Log.log;
/**
* A database backed config registry that registers with the peer system and synchronizes with its
* peers when configuration fields are changed.
*/
public class PeeredDatabaseConfigRegistry extends DatabaseConfigRegistry
{
public PeeredDatabaseConfigRegistry (PersistenceContext ctx, Invoker invoker,
PeerManager peermgr)
throws PersistenceException
{
super(ctx, invoker, "");
_peermgr = peermgr;
}
@Override // from ConfigRegistry
protected ObjectRecord createObjectRecord (String path, DObject object)
{
return new PeerDatabaseObjectRecord(path, object);
}
protected class PeerDatabaseObjectRecord extends DatabaseObjectRecord
implements PeerManager.StaleCacheObserver
{
public PeerDatabaseObjectRecord (String path, DObject object)
{
super(path, object);
_peermgr.addStaleCacheObserver(PEER_CACHE_PREFIX + _path, this);
}
// from interface PeerManager.StaleCacheObserver
public void changedCacheData (Streamable data)
{
@SuppressWarnings("unchecked") StreamableTuple<String,Object> change =
(StreamableTuple<String,Object>)data;
// note that we should ignore the attribute change event we're about to generate
// because it is not a real configuration change but rather a sync
try {
object.changeAttribute(change.left, change.right);
_pendingSyncs.add(change.left);
} catch (Exception e) {
log.log(Level.WARNING, "Config attribute sync failed " + change + ".", e);
}
}
@Override // from ObjectRecord
public void attributeChanged (AttributeChangedEvent event)
{
// if this was a pending sync event, don't pass it to our parent as it is not a real
// configuration change event
if (!_pendingSyncs.remove(event.getName())) {
super.attributeChanged(event);
}
}
@Override // from ObjectRecord
protected void updateValue (String name, Object value)
{
super.updateValue(name, value);
fieldUpdated(name, value);
}
@Override // from ObjectRecord
protected void serialize (String name, String key, Object value)
{
super.serialize(name, key, value);
fieldUpdated(name, value);
}
protected void fieldUpdated (String field, Object value)
{
// broadcast to the other nodes that this value has changed
_peermgr.broadcastStaleCacheData(
PEER_CACHE_PREFIX + _path, new StreamableTuple<String,Object>(field, value));
}
protected ArrayList<String> _pendingSyncs = new ArrayList<String>();
}
protected PeerManager _peermgr;
/** Prefixed to our cache invalidation notifications. */
protected static final String PEER_CACHE_PREFIX = "PeerConfigCache:";
}
src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java
-12
View File
@@ -21,9 +21,6 @@
package com.threerings.crowd.peer.server;
import com.samskivert.jdbc.depot.PersistenceContext;
import com.samskivert.util.Invoker;
import com.threerings.util.Name;
import com.threerings.presents.data.ClientObject;
@@ -52,15 +49,6 @@ import com.threerings.crowd.peer.data.CrowdPeerMarshaller;
public class CrowdPeerManager extends PeerManager
implements CrowdPeerProvider, ChatProvider.TellForwarder
{
/**
* Creates a peer manager that integrates Crowd services across a cluster
* of servers.
*/
public CrowdPeerManager (PersistenceContext ctx, Invoker invoker)
{
super(ctx, invoker);
}
// documentation inherited from interface CrowdPeerProvider
public void deliverTell (ClientObject caller, UserMessage message,
Name target, ChatService.TellListener listener)
src/java/com/threerings/presents/peer/server/PeerManager.java
+7 -15
View File
@@ -128,17 +128,6 @@ public class PeerManager
public void fail (String peerName);
}
/**
* Creates a peer manager which will create a {@link NodeRepository} which will be used to
* publish our existence and discover the other nodes.
*/
public PeerManager (PersistenceContext ctx, Invoker invoker)
{
_invoker = invoker;
_noderepo = new NodeRepository(ctx);
PresentsServer.registerShutdowner(this);
}
/**
* Returns the distributed object that represents this node to its peers.
*/
@@ -153,6 +142,8 @@ public class PeerManager
* bits, so this should not be called until <em>after</em> the main server has set up its
* client factory and authenticator.
*
* @param ctx used to communicate with the database.
* @param invoker we will perform all database operations on the supplied invoker thread.
* @param nodeName this node's unique name.
* @param sharedSecret a shared secret used to allow the peers to authenticate with one
* another.
@@ -161,16 +152,17 @@ public class PeerManager
* known to normal clients (we may want inter-peer communication to take place over a different
* network than the communication between real clients and the various peer servers).
* @param port the port on which other nodes should connect to us.
* @param conprov used to obtain our JDBC connections.
* @param invoker we will perform all database operations on the supplied invoker thread.
*/
public void init (String nodeName, String sharedSecret, String hostName,
String publicHostName, int port)
public void init (PersistenceContext ctx, Invoker invoker, String nodeName,
String sharedSecret, String hostName, String publicHostName, int port)
{
_invoker = invoker;
_noderepo = new NodeRepository(ctx);
_nodeName = nodeName;
_sharedSecret = sharedSecret;
// wire ourselves into the server
PresentsServer.registerShutdowner(this);
PresentsServer.conmgr.setAuthenticator(
new PeerAuthenticator(this, PresentsServer.conmgr.getAuthenticator()));
PresentsServer.clmgr.setClientFactory(