[prev in list] [next in list] [prev in thread] [next in thread]
List: avro-commits
Subject: svn commit: r944045 - in /avro/trunk: ./ doc/src/content/xdocs/ lang/java/src/java/org/apache/avro/
From: cutting () apache ! org
Date: 2010-05-13 23:01:20
Message-ID: 20100513230121.3A06C23889DA () eris ! apache ! org
[Download RAW message or body]
Author: cutting
Date: Thu May 13 23:01:20 2010
New Revision: 944045
URL: http://svn.apache.org/viewvc?rev=944045&view=rev
Log:
AVRO-285. Specify one-way messages and implement in Java.
Modified:
avro/trunk/CHANGES.txt
avro/trunk/doc/src/content/xdocs/spec.xml
avro/trunk/lang/java/src/java/org/apache/avro/Protocol.java
avro/trunk/lang/java/src/java/org/apache/avro/ipc/HttpTransceiver.java
avro/trunk/lang/java/src/java/org/apache/avro/ipc/Requestor.java
avro/trunk/lang/java/src/java/org/apache/avro/ipc/Responder.java
avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketServer.java
avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketTransceiver.java
avro/trunk/lang/java/src/java/org/apache/avro/ipc/Transceiver.java
avro/trunk/lang/java/src/java/org/apache/avro/specific/SpecificCompiler.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestBulkData.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceReflect.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceSpecific.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolDatagram.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGeneric.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGenericMeta.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolHttp.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolParsing.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflect.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflectMeta.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecific.java
avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecificMeta.java
avro/trunk/share/test/schemas/simple.avpr
Modified: avro/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/avro/trunk/CHANGES.txt?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/CHANGES.txt (original)
+++ avro/trunk/CHANGES.txt Thu May 13 23:01:20 2010
@@ -8,6 +8,8 @@ Avro 1.4.0 (unreleased)
AVRO-493. Add support for Hadoop Mapreduce with Avro data files. (cutting)
+ AVRO-285: Specify one-way messages and implement in Java. (cutting)
+
IMPROVEMENTS
AVRO-486. DataFile.open for the ruby side (jmhodges)
Modified: avro/trunk/doc/src/content/xdocs/spec.xml
URL: http://svn.apache.org/viewvc/avro/trunk/doc/src/content/xdocs/spec.xml?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/doc/src/content/xdocs/spec.xml (original)
+++ avro/trunk/doc/src/content/xdocs/spec.xml Thu May 13 23:01:20 2010
@@ -699,14 +699,17 @@
<li>a <em>request</em>, a list of named,
typed <em>parameter</em> schemas (this has the same form
as the fields of a record declaration);</li>
- <li>a <em>response</em> schema; and</li>
- <li>an optional union of <em>error</em> schemas.</li>
+ <li>a <em>response</em> schema; </li>
+ <li>an optional union of <em>error</em> schemas; and</li>
+ <li>an optional <em>one-way</em> boolean parameter.</li>
</ul>
<p>A request parameter list is processed equivalently to an
anonymous record. Since record field lists may vary between
reader and writer, request parameters may also differ
between the caller and responder, and such differences are
resolved in the same manner as record field differences.</p>
+ <p>The one-way parameter may only be true when the response type
+ is <code>"null"</code> and no errors are listed.</p>
</section>
<section>
<title>Sample Protocol</title>
@@ -752,16 +755,23 @@
<li><strong>transmission of request messages</strong>
</li>
<li><strong>receipt of corresponding response messages</strong>
- <p>Servers will send a response message back to the client
- corresponding to each request message. The mechanism of
- that correspondance is transport-specific. For example,
- in HTTP it might be implicit, since HTTP directly supports
- requests and responses. But a transport that multiplexes
- many client threads over a single socket would need to tag
+ <p>Servers may send a response message back to the client
+ corresponding to a request message. The mechanism of
+ correspondance is transport-specific. For example, in
+ HTTP it is implicit, since HTTP directly supports requests
+ and responses. But a transport that multiplexes many
+ client threads over a single socket would need to tag
messages with unique identifiers.</p>
</li>
</ul>
+ <p>Transports may be either <em>stateless</em>
+ or <em>stateful</em>. In a stateless transport, messaging
+ assumes no established connection state, while stateful
+ transports establish connections that may be used for multiple
+ messages. This distinction is discussed further in
+ the <a href="#handshake">handshake</a> section below.</p>
+
<section>
<title>HTTP as Transport</title>
<p>When
@@ -777,6 +787,7 @@
response. The HTTP Content-Type of requests and responses
should be specified as "avro/binary". Requests should be
made using the POST method.</p>
+ <p>HTTP is used by Avro as a stateless transport.</p>
</section>
</section>
@@ -820,19 +831,27 @@
directly, without having to copy it.</p>
</section>
- <section>
+ <section id="handshake">
<title>Handshake</title>
- <p>RPC requests and responses are prefixed by handshakes. The
- purpose of the handshake is to ensure that the client and the
- server have each other's protocol definition, so that the
- client can correctly deserialize responses, and the server can
- correctly deserialize requests. Both clients and servers
+ <p>The purpose of the handshake is to ensure that the client
+ and the server have each other's protocol definition, so that
+ the client can correctly deserialize responses, and the server
+ can correctly deserialize requests. Both clients and servers
should maintain a cache of recently seen protocols, so that,
in most cases, a handshake will be completed without extra
round-trip network exchanges or the transmission of full
protocol text.</p>
+ <p>RPC requests and responses may not be processed until a
+ handshake has been completed. With a stateless transport, all
+ requests and responses are prefixed by handshakes. With a
+ stateful transport, handshakes are only attached to requests
+ and responses until a successful handshake response has been
+ returned over a connection. After this, request and response
+ payloads are sent without handshakes for the lifetime of that
+ connection.</p>
+
<p>The handshake process uses the following record schemas:</p>
<source>
@@ -935,7 +954,11 @@
declaration.</li>
</ul>
- <p>The format of a call response is:</p>
+
+ <p>When a message is declared one-way and a stateful
+ connection has been established by a successful handshake
+ response, no response data is sent. Otherwise the format of
+ the call response is:</p>
<ul>
<li><em>response metadata</em>, a map with values of
type <code>bytes</code></li>
Modified: avro/trunk/lang/java/src/java/org/apache/avro/Protocol.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/Protocol.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/Protocol.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/Protocol.java Thu May 13 23:01:20 \
2010 @@ -44,8 +44,13 @@ import org.codehaus.jackson.JsonGenerato
* <li>a list of named <i>messages</i>, each of which specifies,
* <ul>
* <li><i>request</i>, the parameter schemas;
- * <li><i>response</i>, the response schema;
- * <li><i>errors</i>, a list of potential error schema names.
+ * <li>one of either;
+ * <ul><li>one-way</li></ul>
+ * or
+ * <ul>
+ * <li><i>response</i>, the response schema;
+ * <li><i>errors</i>, an optional list of potential error schema names.
+ * </ul>
* </ul>
* </ul>
*/
@@ -58,17 +63,12 @@ public class Protocol {
private String name;
private String doc;
private Schema request;
- private Schema response;
- private Schema errors;
/** Construct a message. */
- private Message(String name, String doc, Schema request,
- Schema response, Schema errors) {
+ private Message(String name, String doc, Schema request) {
this.name = name;
this.doc = doc;
this.request = request;
- this.response = response;
- this.errors = errors;
}
/** The name of this message. */
@@ -76,10 +76,15 @@ public class Protocol {
/** The parameters of this message. */
public Schema getRequest() { return request; }
/** The returned data. */
- public Schema getResponse() { return response; }
+ public Schema getResponse() { return Schema.create(Schema.Type.NULL); }
/** Errors that might be thrown. */
- public Schema getErrors() { return errors; }
+ public Schema getErrors() {
+ return Schema.createUnion(new ArrayList<Schema>());
+ }
+ /** Returns true if this is a one-way message, with no response or errors.*/
+ public boolean isOneWay() { return true; }
+
public String toString() {
try {
StringWriter writer = new StringWriter();
@@ -97,36 +102,69 @@ public class Protocol {
gen.writeFieldName("request");
request.fieldsToJson(types, gen);
- gen.writeFieldName("response");
- response.toJson(types, gen);
-
- List<Schema> errTypes = errors.getTypes(); // elide system error
- if (errTypes.size() > 1) {
- Schema errs = Schema.createUnion(errTypes.subList(1, errTypes.size()));
- gen.writeFieldName("errors");
- errs.toJson(types, gen);
- }
-
+ toJson1(gen);
gen.writeEndObject();
}
+ void toJson1(JsonGenerator gen) throws IOException {
+ gen.writeStringField("response", "null");
+ gen.writeBooleanField("one-way", true);
+ }
+
public boolean equals(Object o) {
if (o == this) return true;
if (!(o instanceof Message)) return false;
Message that = (Message)o;
return this.name.equals(that.name)
- && this.request.equals(that.request)
- && this.response.equals(that.response)
- && this.errors.equals(that.errors);
+ && this.request.equals(that.request);
}
public int hashCode() {
- return name.hashCode()
- + request.hashCode() + response.hashCode() + errors.hashCode();
+ return name.hashCode() + request.hashCode();
}
- public String getDoc() {
- return doc;
+ public String getDoc() { return doc; }
+
+ }
+
+ private class TwoWayMessage extends Message {
+ private Schema response;
+ private Schema errors;
+
+ /** Construct a message. */
+ private TwoWayMessage(String name, String doc, Schema request,
+ Schema response, Schema errors) {
+ super(name, doc, request);
+ this.response = response;
+ this.errors = errors;
+ }
+
+ @Override public Schema getResponse() { return response; }
+ @Override public Schema getErrors() { return errors; }
+ @Override public boolean isOneWay() { return false; }
+
+ @Override public boolean equals(Object o) {
+ if (!super.equals(o)) return false;
+ if (!(o instanceof TwoWayMessage)) return false;
+ TwoWayMessage that = (TwoWayMessage)o;
+ return this.response.equals(that.response)
+ && this.errors.equals(that.errors);
+ }
+
+ @Override public int hashCode() {
+ return super.hashCode() + response.hashCode() + errors.hashCode();
+ }
+
+ @Override void toJson1(JsonGenerator gen) throws IOException {
+ gen.writeFieldName("response");
+ response.toJson(types, gen);
+
+ List<Schema> errs = errors.getTypes(); // elide system error
+ if (errs.size() > 1) {
+ Schema union = Schema.createUnion(errs.subList(1, errs.size()));
+ gen.writeFieldName("errors");
+ union.toJson(types, gen);
+ }
}
}
@@ -182,12 +220,17 @@ public class Protocol {
/** The messages of this protocol. */
public Map<String,Message> getMessages() { return messages; }
+ /** Create a one-way message. */
+ public Message createMessage(String name, String doc, Schema request) {
+ return new Message(name, doc, request);
+ }
+
+ /** Create a two-way message. */
public Message createMessage(String name, String doc, Schema request,
Schema response, Schema errors) {
- return new Message(name, doc, request, response, errors);
+ return new TwoWayMessage(name, doc, request, response, errors);
}
-
public boolean equals(Object o) {
if (o == this) return true;
if (!(o instanceof Protocol)) return false;
@@ -354,14 +397,33 @@ public class Protocol {
}
Schema request = Schema.createRecord(fields);
+ boolean oneWay = false;
+ JsonNode oneWayNode = json.get("one-way");
+ if (oneWayNode != null) {
+ if (!oneWayNode.isBoolean())
+ throw new SchemaParseException("one-way must be boolean: "+json);
+ oneWay = oneWayNode.getBooleanValue();
+ }
+
JsonNode responseNode = json.get("response");
- if (responseNode == null)
+ if (!oneWay && responseNode == null)
throw new SchemaParseException("No response specified: "+json);
+
+ JsonNode decls = json.get("errors");
+
+ if (oneWay) {
+ if (decls != null)
+ throw new SchemaParseException("one-way can't have errors: "+json);
+ if (responseNode != null
+ && Schema.parse(responseNode, types).getType() != Schema.Type.NULL)
+ throw new SchemaParseException("One way response must be null: "+json);
+ return new Message(messageName, doc, request);
+ }
+
Schema response = Schema.parse(responseNode, types);
List<Schema> errs = new ArrayList<Schema>();
errs.add(SYSTEM_ERROR); // every method can throw
- JsonNode decls = json.get("errors");
if (decls != null) {
if (!decls.isArray())
throw new SchemaParseException("Errors not an array: "+json);
@@ -376,8 +438,8 @@ public class Protocol {
}
}
String doc = parseDocNode(json);
- return new Message(messageName, doc, request, response,
- Schema.createUnion(errs));
+ return new TwoWayMessage(messageName, doc, request, response,
+ Schema.createUnion(errs));
}
public static void main(String[] args) throws Exception {
Modified: avro/trunk/lang/java/src/java/org/apache/avro/ipc/HttpTransceiver.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/ipc/HttpTransceiver.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/ipc/HttpTransceiver.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/ipc/HttpTransceiver.java Thu May 13 \
23:01:20 2010 @@ -38,26 +38,19 @@ public class HttpTransceiver extends Tra
public HttpTransceiver(URL url) { this.url = url; }
public String getRemoteName() { return this.url.toString(); }
-
- @Override
- public synchronized List<ByteBuffer> transceive(List<ByteBuffer> request)
- throws IOException {
- this.connection = (HttpURLConnection)url.openConnection();
- connection.setRequestMethod("POST");
- connection.setRequestProperty("Content-Type", CONTENT_TYPE);
- connection.setRequestProperty("Content-Length",
- Integer.toString(getLength(request)));
- connection.setDoOutput(true);
- //LOG.info("Connecting to: "+url);
- return super.transceive(request);
- }
-
+
public synchronized List<ByteBuffer> readBuffers() throws IOException {
return readBuffers(connection.getInputStream());
}
public synchronized void writeBuffers(List<ByteBuffer> buffers)
throws IOException {
+ connection = (HttpURLConnection)url.openConnection();
+ connection.setRequestMethod("POST");
+ connection.setRequestProperty("Content-Type", CONTENT_TYPE);
+ connection.setRequestProperty("Content-Length",
+ Integer.toString(getLength(buffers)));
+ connection.setDoOutput(true);
writeBuffers(buffers, connection.getOutputStream());
}
Modified: avro/trunk/lang/java/src/java/org/apache/avro/ipc/Requestor.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/ipc/Requestor.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/ipc/Requestor.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/ipc/Requestor.java Thu May 13 \
23:01:20 2010 @@ -83,8 +83,9 @@ public abstract class Requestor {
}
/** Writes a request message and reads a response or error message. */
- public Object request(String messageName, Object request)
+ public synchronized Object request(String messageName, Object request)
throws Exception {
+ Transceiver t = getTransceiver();
BinaryDecoder in = null;
Message m;
RPCContext context = new RPCContext();
@@ -92,7 +93,7 @@ public abstract class Requestor {
ByteBufferOutputStream bbo = new ByteBufferOutputStream();
Encoder out = new BinaryEncoder(bbo);
- writeHandshake(out); // prepend handshake
+ writeHandshake(out); // prepend handshake if needed
// use local protocol to write request
m = getLocal().getMessages().get(messageName);
@@ -105,24 +106,32 @@ public abstract class Requestor {
}
META_WRITER.write(context.requestCallMeta(), out);
- out.writeString(m.getName()); // write message name
+ out.writeString(m.getName()); // write message name
writeRequest(m.getRequest(), request, out); // write request payload
- List<ByteBuffer> response = // transceive
- getTransceiver().transceive(bbo.getBufferList());
-
- ByteBufferInputStream bbi = new ByteBufferInputStream(response);
- in = DecoderFactory.defaultFactory().createBinaryDecoder(bbi, in);
+ if (m.isOneWay() && t.isConnected()) { // send one-way message
+ t.writeBuffers(bbo.getBufferList());
+ return null;
+ } else { // two-way message
+ List<ByteBuffer> response = t.transceive(bbo.getBufferList());
+ ByteBufferInputStream bbi = new ByteBufferInputStream(response);
+ in = DecoderFactory.defaultFactory().createBinaryDecoder(bbi, in);
+ }
} while (!readHandshake(in));
// use remote protocol to read response
- m = getRemote().getMessages().get(messageName);
- if (m == null)
+ Message rm = getRemote().getMessages().get(messageName);
+ if (rm == null)
throw new AvroRuntimeException("Not a remote message: "+messageName);
+ if (m.isOneWay() != rm.isOneWay())
+ throw new AvroRuntimeException("Not both one-way messages: "+messageName);
+
+ if (m.isOneWay() && t.isConnected()) return null; // one-way w/ handshake
+
context.setRequestCallMeta(META_READER.read(null, in));
if (!in.readBoolean()) { // no error
- Object response = readResponse(m.getResponse(), in);
+ Object response = readResponse(rm.getResponse(), in);
context.setResponse(response);
for (RPCPlugin plugin : rpcMetaPlugins) {
plugin.clientReceiveResponse(context);
@@ -130,7 +139,7 @@ public abstract class Requestor {
return response;
} else {
- Exception error = readError(m.getErrors(), in);
+ Exception error = readError(rm.getErrors(), in);
context.setError(error);
for (RPCPlugin plugin : rpcMetaPlugins) {
plugin.clientReceiveResponse(context);
@@ -152,6 +161,7 @@ public abstract class Requestor {
new SpecificDatumReader<HandshakeResponse>(HandshakeResponse.class);
private void writeHandshake(Encoder out) throws IOException {
+ if (getTransceiver().isConnected()) return;
MD5 localHash = new MD5();
localHash.bytes(local.getMD5());
String remoteName = transceiver.getRemoteName();
@@ -177,6 +187,7 @@ public abstract class Requestor {
}
private boolean readHandshake(Decoder in) throws IOException {
+ if (getTransceiver().isConnected()) return true;
boolean established = false;
HandshakeResponse handshake = HANDSHAKE_READER.read(null, in);
switch (handshake.match) {
@@ -205,6 +216,8 @@ public abstract class Requestor {
for (RPCPlugin plugin : rpcMetaPlugins) {
plugin.clientFinishConnect(context);
}
+ if (established)
+ getTransceiver().setRemote(remote);
return established;
}
Modified: avro/trunk/lang/java/src/java/org/apache/avro/ipc/Responder.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/ipc/Responder.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/ipc/Responder.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/ipc/Responder.java Thu May 13 \
23:01:20 2010 @@ -85,37 +85,49 @@ public abstract class Responder {
/** Called by a server to deserialize a request, compute and serialize
* a response or error. */
public List<ByteBuffer> respond(List<ByteBuffer> buffers) throws IOException {
+ return respond(buffers, null);
+ }
+
+ /** Called by a server to deserialize a request, compute and serialize a
+ * response or error. Transciever is used by connection-based servers to
+ * track handshake status of connection. */
+ public List<ByteBuffer> respond(List<ByteBuffer> buffers,
+ Transceiver connection) throws IOException {
Decoder in = DecoderFactory.defaultFactory().createBinaryDecoder(
new ByteBufferInputStream(buffers), null);
ByteBufferOutputStream bbo = new ByteBufferOutputStream();
Encoder out = new BinaryEncoder(bbo);
Exception error = null;
RPCContext context = new RPCContext();
+ boolean wasConnected = connection != null && connection.isConnected();
try {
- Protocol remote = handshake(in, out);
+ Protocol remote = handshake(in, out, connection);
if (remote == null) // handshake failed
return bbo.getBufferList();
// read request using remote protocol specification
context.setRequestCallMeta(META_READER.read(null, in));
String messageName = in.readString(null).toString();
- Message m = remote.getMessages().get(messageName);
- if (m == null)
+ Message rm = remote.getMessages().get(messageName);
+ if (rm == null)
throw new AvroRuntimeException("No such remote message: "+messageName);
- context.setMessage(m);
+ context.setMessage(rm);
- Object request = readRequest(m.getRequest(), in);
+ Object request = readRequest(rm.getRequest(), in);
for (RPCPlugin plugin : rpcMetaPlugins) {
plugin.serverReceiveRequest(context);
}
// create response using local protocol specification
- m = getLocal().getMessages().get(messageName);
+ Message m = getLocal().getMessages().get(messageName);
if (m == null)
throw new AvroRuntimeException("No message named "+messageName
+" in "+getLocal());
+ if (m.isOneWay() != rm.isOneWay())
+ throw new AvroRuntimeException("Not both one-way: "+messageName);
+
Object response = null;
try {
response = respond(m, request);
@@ -125,6 +137,9 @@ public abstract class Responder {
context.setError(error);
}
+ if (m.isOneWay() && wasConnected) // no response data
+ return null;
+
for (RPCPlugin plugin : rpcMetaPlugins) {
plugin.serverSendResponse(context);
}
@@ -154,8 +169,10 @@ public abstract class Responder {
private SpecificDatumReader<HandshakeRequest> handshakeReader =
new SpecificDatumReader<HandshakeRequest>(HandshakeRequest.class);
- private Protocol handshake(Decoder in, Encoder out)
+ private Protocol handshake(Decoder in, Encoder out, Transceiver connection)
throws IOException {
+ if (connection != null && connection.isConnected())
+ return connection.getRemote();
HandshakeRequest request = (HandshakeRequest)handshakeReader.read(null, in);
Protocol remote = protocols.get(request.clientHash);
if (remote == null && request.clientProtocol != null) {
@@ -184,6 +201,10 @@ public abstract class Responder {
response.meta = context.responseHandshakeMeta();
handshakeWriter.write(response, out);
+
+ if (connection != null && response.match != HandshakeMatch.NONE)
+ connection.setRemote(remote);
+
return remote;
}
Modified: avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketServer.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketServer.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketServer.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketServer.java Thu May 13 \
23:01:20 2010 @@ -89,7 +89,7 @@ public class SocketServer extends Thread
try {
try {
while (true) {
- writeBuffers(responder.respond(readBuffers()));
+ writeBuffers(responder.respond(readBuffers(), this));
}
} catch (ClosedChannelException e) {
return;
Modified: avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketTransceiver.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketTransceiver.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketTransceiver.java \
(original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/ipc/SocketTransceiver.java Thu May \
13 23:01:20 2010 @@ -28,6 +28,8 @@ import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.avro.Protocol;
+
/** A socket-based {@link Transceiver} implementation. This uses a simple,
* non-standard wire protocol and is not intended for production services. */
public class SocketTransceiver extends Transceiver {
@@ -36,6 +38,8 @@ public class SocketTransceiver extends T
private SocketChannel channel;
private ByteBuffer header = ByteBuffer.allocate(4);
+
+ private Protocol remote;
public SocketTransceiver(SocketAddress address) throws IOException {
this(SocketChannel.open(address));
@@ -74,6 +78,7 @@ public class SocketTransceiver extends T
public synchronized void writeBuffers(List<ByteBuffer> buffers)
throws IOException {
+ if (buffers == null) return; // no data to write
for (ByteBuffer buffer : buffers) {
writeLength(buffer.limit()); // length-prefix
channel.write(buffer);
@@ -88,7 +93,17 @@ public class SocketTransceiver extends T
channel.write(header);
}
- public void close() throws IOException {
+ @Override public boolean isConnected() { return remote != null; }
+
+ @Override public void setRemote(Protocol remote) {
+ this.remote = remote;
+ }
+
+ @Override public Protocol getRemote() {
+ return remote;
+ }
+
+ @Override public void close() throws IOException {
if (channel.isOpen()) {
LOG.info("closing to "+getRemoteName());
channel.close();
Modified: avro/trunk/lang/java/src/java/org/apache/avro/ipc/Transceiver.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/ipc/Transceiver.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/ipc/Transceiver.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/ipc/Transceiver.java Thu May 13 \
23:01:20 2010 @@ -23,22 +23,50 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
-/** Base class for transmitters and recievers of raw binary messages. */
+import org.apache.avro.Protocol;
+
+/** Base transport class used by {@link Requestor}. */
public abstract class Transceiver implements Closeable {
public abstract String getRemoteName();
+ /** Called by {@link Requestor#request(String,Object)} for two-way messages.
+ * By default calls {@link #writeBuffers(List)} followed by
+ * {@link #readBuffers()}. */
public synchronized List<ByteBuffer> transceive(List<ByteBuffer> request)
throws IOException {
writeBuffers(request);
return readBuffers();
}
+ /** Called by the default definition of {@link #transceive(List)}.*/
public abstract List<ByteBuffer> readBuffers() throws IOException;
+ /** Called by {@link Requestor#request(String,Object)} for one-way messages.*/
public abstract void writeBuffers(List<ByteBuffer> buffers)
throws IOException;
+ /** True if a handshake has been completed for this connection. Used to
+ * determine whether a handshake need be completed prior to a one-way
+ * message. Requests and responses are always prefixed by handshakes, but
+ * one-way messages. If the first request sent over a connection is one-way,
+ * then a handshake-only response is returned. Subsequent one-way messages
+ * over the connection will have no response data sent. Returns false by
+ * default. */
+ public boolean isConnected() { return false; }
+
+ /** Called with the remote protocol when a handshake has been completed.
+ * After this has been called and while a connection is maintained, {@link
+ * #isConnected()} should return true and #getRemote() should return this
+ * protocol. Does nothing by default. */
+ public void setRemote(Protocol protocol) {}
+
+ /** Returns the protocol passed to {@link #setRemote(Protocol)}. Throws
+ * IllegalStateException by default. */
+ public Protocol getRemote() {
+ throw new IllegalStateException("Not connected.");
+ }
+
public void close() throws IOException {}
}
Modified: avro/trunk/lang/java/src/java/org/apache/avro/specific/SpecificCompiler.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/specific/SpecificCompiler.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/specific/SpecificCompiler.java \
(original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/specific/SpecificCompiler.java Thu \
May 13 23:01:20 2010 @@ -188,10 +188,14 @@ public class SpecificCompiler {
String name = e.getKey();
Message message = e.getValue();
Schema request = message.getRequest();
- Schema response = message.getResponse();
+ String response = message.isOneWay() ? "void"
+ : unbox(message.getResponse());
doc(out, 1, e.getValue().getDoc());
- line(out, 1, unbox(response)+" "+ mangle(name)+"("+params(request)+")");
- line(out, 2,"throws \
org.apache.avro.ipc.AvroRemoteException"+errors(message.getErrors())+";"); + \
line(out, 1, response+" "+ mangle(name)+"("+params(request)+")" + + \
(message.isOneWay() ? "" + : (" throws \
org.apache.avro.ipc.AvroRemoteException" + \
+errors(message.getErrors()))) + +";");
}
line(out, 0, "}");
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestBulkData.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestBulkData.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestBulkData.java (original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestBulkData.java Thu May 13 \
23:01:20 2010 @@ -24,7 +24,7 @@ import org.apache.avro.ipc.HttpTransceiv
import org.apache.avro.ipc.Server;
import org.apache.avro.ipc.Transceiver;
-import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -64,6 +64,7 @@ public class TestBulkData {
@Before
public void startServer() throws Exception {
+ if (server != null) return;
server =
new HttpServer(new SpecificResponder(BulkData.class, new BulkDataImpl()),
0);
@@ -84,8 +85,8 @@ public class TestBulkData {
proxy.write(DATA.duplicate());
}
- @After
- public void stopServer() throws Exception {
+ @AfterClass
+ public static void stopServer() throws Exception {
server.close();
}
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceReflect.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceReflect.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceReflect.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceReflect.java Thu \
May 13 23:01:20 2010 @@ -28,8 +28,9 @@ import java.net.InetSocketAddress;
public class TestNamespaceReflect extends TestNamespaceSpecific {
- @Before
+ @Before @Override
public void testStartServer() throws Exception {
+ if (server != null) return;
server = new SocketServer(new ReflectResponder(TestNamespace.class, new \
TestImpl()), new InetSocketAddress(0));
client = new SocketTransceiver(new InetSocketAddress(server.getPort()));
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceSpecific.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceSpecific.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceSpecific.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestNamespaceSpecific.java Thu \
May 13 23:01:20 2010 @@ -28,7 +28,7 @@ import org.apache.avro.test.util.MD5;
import org.apache.avro.test.errors.TestError;
import org.apache.avro.test.namespace.TestRecord;
import org.apache.avro.util.Utf8;
-import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -54,6 +54,7 @@ public class TestNamespaceSpecific {
@Before
public void testStartServer() throws Exception {
+ if (server != null) return;
server = new SocketServer(new SpecificResponder(TestNamespace.class, new \
TestImpl()), new InetSocketAddress(0));
client = new SocketTransceiver(new InetSocketAddress(server.getPort()));
@@ -83,8 +84,8 @@ public class TestNamespaceSpecific {
assertEquals("an error", error.message.toString());
}
- @After
- public void testStopServer() throws IOException {
+ @AfterClass
+ public static void testStopServer() throws IOException {
client.close();
server.close();
}
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolDatagram.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolDatagram.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolDatagram.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolDatagram.java Thu \
May 13 23:01:20 2010 @@ -29,8 +29,9 @@ import java.net.InetSocketAddress;
public class TestProtocolDatagram extends TestProtocolSpecific {
- @Before
+ @Before @Override
public void testStartServer() throws Exception {
+ if (server != null) return;
server =
new DatagramServer(new SpecificResponder(Simple.class, new TestImpl()),
new InetSocketAddress("localhost",
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGeneric.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGeneric.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGeneric.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGeneric.java Thu \
May 13 23:01:20 2010 @@ -28,7 +28,7 @@ import org.apache.avro.ipc.SocketServer;
import org.apache.avro.ipc.SocketTransceiver;
import org.apache.avro.ipc.Transceiver;
import org.apache.avro.util.Utf8;
-import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -99,6 +99,7 @@ public class TestProtocolGeneric {
@Before
public void testStartServer() throws Exception {
+ if (server != null) return;
server = new SocketServer(new TestResponder(), new InetSocketAddress(0));
client = new SocketTransceiver(new InetSocketAddress(server.getPort()));
requestor = new GenericRequestor(PROTOCOL, client);
@@ -187,8 +188,8 @@ public class TestProtocolGeneric {
}
}
- @After
- public void testStopServer() throws IOException {
+ @AfterClass
+ public static void testStopServer() throws IOException {
client.close();
server.close();
}
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGenericMeta.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGenericMeta.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGenericMeta.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolGenericMeta.java \
Thu May 13 23:01:20 2010 @@ -27,8 +27,9 @@ import org.junit.Before;
public class TestProtocolGenericMeta extends TestProtocolGeneric {
- @Before
+ @Before @Override
public void testStartServer() throws Exception {
+ if (server != null) return;
Responder responder = new TestResponder();
responder.addRPCPlugin(new RPCMetaTestPlugin("key1"));
responder.addRPCPlugin(new RPCMetaTestPlugin("key2"));
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolHttp.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolHttp.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolHttp.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolHttp.java Thu May \
13 23:01:20 2010 @@ -28,8 +28,9 @@ import java.net.URL;
public class TestProtocolHttp extends TestProtocolSpecific {
- @Before
+ @Before @Override
public void testStartServer() throws Exception {
+ if (server != null) return;
server =
new HttpServer(new SpecificResponder(Simple.class, new TestImpl()), 0);
client =
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolParsing.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolParsing.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolParsing.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolParsing.java Thu \
May 13 23:01:20 2010 @@ -18,12 +18,15 @@
package org.apache.avro;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.IOException;
import org.junit.Test;
+import org.apache.avro.Protocol.Message;
+
public class TestProtocolParsing {
public static Protocol getSimpleProtocol() throws IOException {
File file = new File("../../share/test/schemas/simple.avpr");
@@ -36,7 +39,49 @@ public class TestProtocolParsing {
Protocol protocol = getSimpleProtocol();
assertEquals(protocol.getDoc(), "Protocol used for testing.");
- assertEquals(5, protocol.getMessages().size());
+ assertEquals(6, protocol.getMessages().size());
assertEquals("Pretend you're in a cave!", \
protocol.getMessages().get("echo").getDoc()); }
+
+ private static Message parseMessage(String message) throws Exception {
+ return Protocol.parse("{\"protocol\": \"org.foo.Bar\","
+ +"\"types\": [],"
+ +"\"messages\": {"
+ + message
+ + "}}").getMessages().values().iterator().next();
+ }
+
+ @Test public void oneWay() throws Exception {
+ Message m;
+ // permit one-way messages w/ null resposne
+ m = parseMessage("\"ack\": {"
+ +"\"request\": [],"
+ +"\"response\": \"null\","
+ +"\"one-way\": true}");
+ assertTrue(m.isOneWay());
+ // permit one-way messages w/o response
+ m = parseMessage("\"ack\": {"
+ +"\"request\": [],"
+ +"\"one-way\": true}");
+ assertTrue(m.isOneWay());
+ }
+
+ @Test(expected=SchemaParseException.class)
+ public void oneWayResponse() throws Exception {
+ // prohibit one-way messages with a non-null response type
+ parseMessage("\"ack\": {"
+ +"\"request\": [\"string\"],"
+ +"\"response\": \"string\","
+ +"\"one-way\": true}");
+ }
+
+ @Test(expected=SchemaParseException.class)
+ public void oneWayError() throws Exception {
+ // prohibit one-way messages with errors
+ parseMessage("\"ack\": {"
+ +"\"request\": [\"string\"],"
+ +"\"errors\": [],"
+ +"\"one-way\": true}");
+ }
+
}
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflect.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflect.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflect.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflect.java Thu \
May 13 23:01:20 2010 @@ -24,7 +24,7 @@ import org.apache.avro.ipc.SocketTransce
import org.apache.avro.reflect.ReflectRequestor;
import org.apache.avro.reflect.ReflectResponder;
-import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -69,6 +69,7 @@ public class TestProtocolReflect {
@Before
public void testStartServer() throws Exception {
+ if (server != null) return;
server = new SocketServer(new ReflectResponder(Simple.class, new TestImpl()),
new InetSocketAddress(0));
client = new SocketTransceiver(new InetSocketAddress(server.getPort()));
@@ -117,8 +118,8 @@ public class TestProtocolReflect {
assertEquals("foo", error.getMessage());
}
- @After
- public void testStopServer() throws IOException {
+ @AfterClass
+ public static void testStopServer() throws IOException {
client.close();
server.close();
}
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflectMeta.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflectMeta.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflectMeta.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolReflectMeta.java \
Thu May 13 23:01:20 2010 @@ -28,8 +28,9 @@ import java.net.InetSocketAddress;
public class TestProtocolReflectMeta extends TestProtocolReflect {
- @Before
+ @Before @Override
public void testStartServer() throws Exception {
+ if (server != null) return;
ReflectResponder rresp = new ReflectResponder(Simple.class, new TestImpl());
rresp.addRPCPlugin(new RPCMetaTestPlugin("key1"));
rresp.addRPCPlugin(new RPCMetaTestPlugin("key2"));
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecific.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecific.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecific.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecific.java Thu \
May 13 23:01:20 2010 @@ -30,9 +30,9 @@ import org.apache.avro.test.MD5;
import org.apache.avro.test.TestError;
import org.apache.avro.test.TestRecord;
import org.apache.avro.util.Utf8;
-import org.junit.After;
-import org.junit.Before;
import org.junit.Test;
+import org.junit.Before;
+import org.junit.AfterClass;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -51,6 +51,8 @@ public class TestProtocolSpecific {
protected static final File SERVER_PORTS_DIR
= new File(System.getProperty("test.dir", "/tmp")+"/server-ports/");
+ public static int ackCount;
+
public static class TestImpl implements Simple {
public Utf8 hello(Utf8 greeting) { return new Utf8("goodbye"); }
public int add(int arg1, int arg2) { return arg1 + arg2; }
@@ -61,6 +63,7 @@ public class TestProtocolSpecific {
error.message = new Utf8("an error");
throw error;
}
+ public void ack() { ackCount++; }
}
protected static Server server;
@@ -69,6 +72,7 @@ public class TestProtocolSpecific {
@Before
public void testStartServer() throws Exception {
+ if (server != null) return;
server = new SocketServer(new SpecificResponder(Simple.class, new TestImpl()),
new InetSocketAddress(0));
client = new SocketTransceiver(new InetSocketAddress(server.getPort()));
@@ -137,8 +141,18 @@ public class TestProtocolSpecific {
assertEquals("an error", error.message.toString());
}
- @After
- public void testStopServer() throws IOException {
+ @Test
+ public void testOneWay() throws IOException {
+ ackCount = 0;
+ proxy.ack();
+ proxy.hello(new Utf8("foo")); // intermix normal req
+ proxy.ack();
+ try { Thread.sleep(100); } catch (InterruptedException e) {}
+ assertEquals(2, ackCount);
+ }
+
+ @AfterClass
+ public static void testStopServer() throws IOException {
client.close();
server.close();
}
Modified: avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecificMeta.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecificMeta.java?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecificMeta.java \
(original)
+++ avro/trunk/lang/java/src/test/java/org/apache/avro/TestProtocolSpecificMeta.java \
Thu May 13 23:01:20 2010 @@ -27,11 +27,11 @@ import org.apache.avro.specific.Specific
import org.apache.avro.test.Simple;
import org.junit.Before;
-
public class TestProtocolSpecificMeta extends TestProtocolSpecific {
- @Before
+ @Before @Override
public void testStartServer() throws Exception {
+ if (server != null) return;
Responder responder = new SpecificResponder(Simple.class, new TestImpl());
responder.addRPCPlugin(new RPCMetaTestPlugin("key1"));
responder.addRPCPlugin(new RPCMetaTestPlugin("key2"));
Modified: avro/trunk/share/test/schemas/simple.avpr
URL: http://svn.apache.org/viewvc/avro/trunk/share/test/schemas/simple.avpr?rev=944045&r1=944044&r2=944045&view=diff
==============================================================================
--- avro/trunk/share/test/schemas/simple.avpr (original)
+++ avro/trunk/share/test/schemas/simple.avpr Thu May 13 23:01:20 2010
@@ -51,6 +51,13 @@
"request": [],
"response": "null",
"errors": ["TestError"]
+ },
+
+ "ack": {
+ "doc": "Send a one way message",
+ "request": [],
+ "response": "null",
+ "one-way": true
}
}
[prev in list] [next in list] [prev in thread] [next in thread]
Configure |
About |
News |
Add a list |
Sponsored by KoreLogic