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

Added support for "streamable closures". The idea with these are that the

Browse Source 
streaming mechanism will quietly gloss over the funny little outer class
reference stuffed into your closure as long as you promise not to use it
(because it's going to be null on the other side of the wire).

This will allow us to write:

_peermgr.invokeNodeAction(new NodeAction() {
    ...
});

which we presently have a big fat warning against, since it didn't work until
about 30 seconds from now.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6550 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2011-03-29 23:28:26 +00:00
parent 7c07bfcedc
commit 064fbe425a
3 changed files with 159 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/threerings/io/Streamable.java
+12 -6
View File
@@ -25,20 +25,26 @@ package com.threerings.io;
* Marks an object as streamable, meaning that it can be written to {@link ObjectOutputStream}
* instances and read from {@link ObjectInputStream} instances.
*
* <p> All non-<code>transient</code> fields will be automatically written and restored for a
* {@link Streamable} instance. Classes that wish to stream transient fields or customize the
* streaming process should implement methods with the following signatures:
* <p> All non-{@code transient} fields will be automatically written and restored for a {@link
* Streamable} instance. Classes that wish to stream transient fields or customize the streaming
* process should implement methods with the following signatures: </p>
*
* <p><code>
* public void writeObject ({@link ObjectOutputStream} out);
* public void readObject ({@link ObjectInputStream} in);
* </code>
* </code></p>
*
* <p> They can then handle the entirety of the streaming process, or call {@link
* ObjectOutputStream#defaultWriteObject} and {@link ObjectInputStream#defaultReadObject} from
* within their <code>writeObject</code> and <code>readObject</code> methods to perform the
* standard streaming in addition to their customized behavior.
* within their {@code writeObject} and {@code readObject} methods to perform the standard
* streaming in addition to their customized behavior.</p>
*/
public interface Streamable
{
/**
* A marker interface for streamable classes that expect to be extended anonymously, but for
* which the implicit outer class reference can (and should) be ignored. This allows one to
* package up units of code and ship them between peers, or even between client and server.
*/
public interface Closure extends Streamable {}
}
src/main/java/com/threerings/io/Streamer.java
+60 -13
View File
@@ -22,6 +22,7 @@
package com.threerings.io;
import java.lang.reflect.Array;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
@@ -251,7 +252,6 @@ public abstract class Streamer
}
if (Modifier.isFinal(componentType.getModifiers())) {
return new FinalArrayStreamer(componentType, delegate);
} else {
return new ArrayStreamer(componentType, delegate);
}
@@ -309,7 +309,6 @@ public abstract class Streamer
// if there is no reader and no writer, we can do a simpler thing
if ((reader == null) && (writer == null)) {
return new ClassStreamer(target);
} else {
return new CustomClassStreamer(target, reader, writer);
}
@@ -365,7 +364,11 @@ public abstract class Streamer
if (ObjectInputStream.STREAM_DEBUG) {
log.info(in.hashCode() + ": Creating object '" + _target.getName() + "'.");
}
return _target.newInstance();
return _ctor.newInstance(_ctorArgs);
} catch (InvocationTargetException ite) {
String errmsg = "Error instantiating object [type=" + _target.getName() + "]";
throw (IOException) new IOException(errmsg).initCause(ite.getCause());
} catch (InstantiationException ie) {
String errmsg = "Error instantiating object [type=" + _target.getName() + "]";
@@ -415,8 +418,9 @@ public abstract class Streamer
*/
protected void initMarshallers ()
{
// reflect on all the object's fields and remove all marked with NotStreamable
// reflect on all the object's fields
List<Field> fields = Lists.newArrayList();
// this will read all non-static, non-transient fields into our fields list
ClassUtil.getFields(_target, fields);
// Checks whether or not we should stream the fields in alphabetical order.
@@ -426,8 +430,13 @@ public abstract class Streamer
QuickSort.sort(fields, FIELD_NAME_ORDER);
}
_fields = Iterables.toArray(
Iterables.filter(fields, _isStreamableFieldPred), Field.class);
// note whether this class is a streamable closure
final boolean isClosure = Streamable.Closure.class.isAssignableFrom(_target);
// remove all marked with NotStreamable, and if we're a streamable closure, remove any
// anonymous inner class reference
Predicate<Field> filter = isClosure ? IS_STREAMCLOSURE : IS_STREAMABLE;
_fields = Iterables.toArray(Iterables.filter(fields, filter), Field.class);
int fcount = _fields.length;
// obtain field marshallers for all of our fields
@@ -445,6 +454,25 @@ public abstract class Streamer
_fields[ii].getName() + ".");
}
}
// obtain the constructor we'll use to create instances
_ctorArgs = isClosure ? SINGLE_NULL_ARG : NO_ARGS;
for (Constructor<?> ctor : _target.getDeclaredConstructors()) {
if (ctor.getParameterTypes().length == _ctorArgs.length) {
if (_ctor != null) {
throw new RuntimeException(
"Streamable has multiple applicable ctors [class=" + _target.getName() +
", argCount=" + _ctorArgs.length + "]");
}
_ctor = ctor;
// keep going, to be sure we catch conflicting ctors
}
}
// if this is a streamable closure, we need to make the constructor accessible
if (isClosure) {
_ctor.setAccessible(true);
}
}
@Override
@@ -458,6 +486,12 @@ public abstract class Streamer
/** The class for which this streamer instance is configured. */
protected Class<?> _target;
/** The constructor we use to create instances. */
protected Constructor<?> _ctor;
/** The arguments we pass to said constructor (empty or a single null). */
protected Object[] _ctorArgs;
/** The non-transient, non-static public fields that we will stream when requested. */
protected Field[] _fields;
@@ -953,13 +987,6 @@ public abstract class Streamer
}
}
/** A simple predicate to filter "NotStreamable" members from a Streamable object's fields. */
protected static final Predicate<Field> _isStreamableFieldPred = new Predicate<Field>() {
public boolean apply (Field obj) {
return (obj.getAnnotation(NotStreamable.class) == null);
}
};
/** The name of the custom reader method. */
protected static final String READER_METHOD_NAME = "readObject";
@@ -971,4 +998,24 @@ public abstract class Streamer
/** The argument list for the custom writer method. */
protected static final Class<?>[] WRITER_ARGS = { ObjectOutputStream.class };
/** Filters "NotStreamable" members from a field list. */
protected static final Predicate<Field> IS_STREAMABLE = new Predicate<Field>() {
public boolean apply (Field obj) {
return (obj.getAnnotation(NotStreamable.class) == null);
}
};
/** Filters "NotStreamable" members and enclosing class refs from a field list. */
protected static final Predicate<Field> IS_STREAMCLOSURE = new Predicate<Field>() {
public boolean apply (Field obj) {
return IS_STREAMABLE.apply(obj) && !obj.isSynthetic();
}
};
/** Used by {@link ClassStreamer} to create instances. */
protected static final Object[] NO_ARGS = new Object[] {};
/** Used by {@link ClassStreamer} to create instances. */
protected static final Object[] SINGLE_NULL_ARG = new Object[] { null };
}
src/test/java/com/threerings/io/StreamableTest.java
+87 -16
View File
@@ -325,12 +325,7 @@ public class StreamableTest
throws IOException, ClassNotFoundException
{
Widget w = new Widget();
ByteArrayOutputStream bout = new ByteArrayOutputStream();
ObjectOutputStream oout = new ObjectOutputStream(bout);
oout.writeObject(w);
ObjectInputStream oin = new ObjectInputStream(
new ByteArrayInputStream(bout.toByteArray()));
assertEquals(w, oin.readObject());
assertEquals(w, unflatten(flatten(w)));
}
@Test
@@ -340,10 +335,7 @@ public class StreamableTest
Widget w = new Widget();
// make sure that we serialize to the expected stream of bytes
ByteArrayOutputStream bout = new ByteArrayOutputStream();
ObjectOutputStream oout = new ObjectOutputStream(bout);
oout.writeObject(w);
byte[] data = bout.toByteArray();
byte[] data = flatten(w);
// uncomment this and rerun the tests to generate an updated WIRE_DATA blob
// printWireData(w);
@@ -353,17 +345,13 @@ public class StreamableTest
assertEquals(StringUtil.hexlate(data), StringUtil.hexlate(WIRE_DATA));
// make sure that we unserialize a known stream of bytes to the expected object
ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(WIRE_DATA));
assertEquals(w, oin.readObject());
assertEquals(w, unflatten(WIRE_DATA));
}
protected void printWireData (Object o)
throws IOException
{
ByteArrayOutputStream bout = new ByteArrayOutputStream();
ObjectOutputStream oout = new ObjectOutputStream(bout);
oout.writeObject(o);
String dstr = StringUtil.wordWrap(StringUtil.hexlate(bout.toByteArray()), 80);
String dstr = StringUtil.wordWrap(StringUtil.hexlate(flatten(o)), 80);
dstr = StringUtil.join(dstr.split("\n"), "\" +\n \"");
System.out.println(" protected static final byte[] WIRE_DATA = "
+ "StringUtil.unhexlate(\n \"" + dstr + "\");");
@@ -405,6 +393,89 @@ public class StreamableTest
assertEquals(tup, otup);
}
@Test(expected=RuntimeException.class)
public void testUnlabledClosureFail ()
throws IOException, ClassNotFoundException
{
abstract class Action implements Streamable {
public int count;
public String arg;
public abstract String act ();
}
Action act = new Action() {
public String act () {
return count + ":" + arg;
}
};
act.count = 3;
act.arg = "hello";
Action react = (Action)unflatten(flatten(act));
assertEquals(act.act(), react.act());
}
@Test
public void testClosure ()
throws IOException, ClassNotFoundException
{
abstract class Action implements Streamable.Closure {
public int count;
public String arg;
public abstract String act ();
}
Action act = new Action() {
public String act () {
return count + ":" + arg;
}
};
act.count = 3;
act.arg = "hello";
Action react = (Action)unflatten(flatten(act));
assertEquals(act.act(), react.act());
}
// unfortunately we can't warn you if you do something naughty in your closure, but since we
// flatten and unflatten closures even when running on the local peer, the programmer should
// find about about funny business early enough
@Test(expected=NullPointerException.class)
public void testNaughtyClosureFail ()
throws IOException, ClassNotFoundException
{
abstract class Action implements Streamable.Closure {
public int count;
public String arg;
public abstract String act ();
}
Action act = new Action() {
public String act () {
return count + ":" + arg + ":" + naughtyOuterCall();
}
};
act.count = 3;
act.arg = "hello";
Action react = (Action)unflatten(flatten(act));
assertEquals(act.act(), react.act());
}
protected int naughtyOuterCall ()
{
return 42;
}
protected static byte[] flatten (Object object)
throws IOException
{
ByteArrayOutputStream bout = new ByteArrayOutputStream();
ObjectOutputStream oout = new ObjectOutputStream(bout);
oout.writeObject(object);
return bout.toByteArray();
}
protected static Object unflatten (byte[] data)
throws IOException, ClassNotFoundException
{
return new ObjectInputStream(new ByteArrayInputStream(data)).readObject();
}
protected static final byte[] WIRE_DATA = StringUtil.unhexlate(
"ffff0027636f6d2e746872656572696e67732e696f2e53747265616d61626c655465737424576964" +
"676574017f00617fff7fffffff7fffffffffffffff7f7fffff7fefffffffffffff0101017f010061" +