You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sz...@apache.org on 2012/01/12 03:54:02 UTC
svn commit: r1230378 [2/2] - in
/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common:
./ src/main/java/ src/main/java/org/apache/hadoop/ipc/
src/main/java/org/apache/hadoop/ipc/protobuf/ src/proto/
src/test/java/org/apache/hadoop/...
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java Thu Jan 12 02:54:01 2012
@@ -1,3 +1,20 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
// Generated by the protocol buffer compiler. DO NOT EDIT!
// source: hadoop_rpc.proto
@@ -18,6 +35,14 @@ public final class HadoopRpcProtos {
// optional bytes request = 2;
boolean hasRequest();
com.google.protobuf.ByteString getRequest();
+
+ // required string declaringClassProtocolName = 3;
+ boolean hasDeclaringClassProtocolName();
+ String getDeclaringClassProtocolName();
+
+ // required uint64 clientProtocolVersion = 4;
+ boolean hasClientProtocolVersion();
+ long getClientProtocolVersion();
}
public static final class HadoopRpcRequestProto extends
com.google.protobuf.GeneratedMessage
@@ -90,9 +115,53 @@ public final class HadoopRpcProtos {
return request_;
}
+ // required string declaringClassProtocolName = 3;
+ public static final int DECLARINGCLASSPROTOCOLNAME_FIELD_NUMBER = 3;
+ private java.lang.Object declaringClassProtocolName_;
+ public boolean hasDeclaringClassProtocolName() {
+ return ((bitField0_ & 0x00000004) == 0x00000004);
+ }
+ public String getDeclaringClassProtocolName() {
+ java.lang.Object ref = declaringClassProtocolName_;
+ if (ref instanceof String) {
+ return (String) ref;
+ } else {
+ com.google.protobuf.ByteString bs =
+ (com.google.protobuf.ByteString) ref;
+ String s = bs.toStringUtf8();
+ if (com.google.protobuf.Internal.isValidUtf8(bs)) {
+ declaringClassProtocolName_ = s;
+ }
+ return s;
+ }
+ }
+ private com.google.protobuf.ByteString getDeclaringClassProtocolNameBytes() {
+ java.lang.Object ref = declaringClassProtocolName_;
+ if (ref instanceof String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+ declaringClassProtocolName_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+
+ // required uint64 clientProtocolVersion = 4;
+ public static final int CLIENTPROTOCOLVERSION_FIELD_NUMBER = 4;
+ private long clientProtocolVersion_;
+ public boolean hasClientProtocolVersion() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public long getClientProtocolVersion() {
+ return clientProtocolVersion_;
+ }
+
private void initFields() {
methodName_ = "";
request_ = com.google.protobuf.ByteString.EMPTY;
+ declaringClassProtocolName_ = "";
+ clientProtocolVersion_ = 0L;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -103,6 +172,14 @@ public final class HadoopRpcProtos {
memoizedIsInitialized = 0;
return false;
}
+ if (!hasDeclaringClassProtocolName()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasClientProtocolVersion()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
memoizedIsInitialized = 1;
return true;
}
@@ -116,6 +193,12 @@ public final class HadoopRpcProtos {
if (((bitField0_ & 0x00000002) == 0x00000002)) {
output.writeBytes(2, request_);
}
+ if (((bitField0_ & 0x00000004) == 0x00000004)) {
+ output.writeBytes(3, getDeclaringClassProtocolNameBytes());
+ }
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ output.writeUInt64(4, clientProtocolVersion_);
+ }
getUnknownFields().writeTo(output);
}
@@ -133,6 +216,14 @@ public final class HadoopRpcProtos {
size += com.google.protobuf.CodedOutputStream
.computeBytesSize(2, request_);
}
+ if (((bitField0_ & 0x00000004) == 0x00000004)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(3, getDeclaringClassProtocolNameBytes());
+ }
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeUInt64Size(4, clientProtocolVersion_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -166,6 +257,16 @@ public final class HadoopRpcProtos {
result = result && getRequest()
.equals(other.getRequest());
}
+ result = result && (hasDeclaringClassProtocolName() == other.hasDeclaringClassProtocolName());
+ if (hasDeclaringClassProtocolName()) {
+ result = result && getDeclaringClassProtocolName()
+ .equals(other.getDeclaringClassProtocolName());
+ }
+ result = result && (hasClientProtocolVersion() == other.hasClientProtocolVersion());
+ if (hasClientProtocolVersion()) {
+ result = result && (getClientProtocolVersion()
+ == other.getClientProtocolVersion());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -183,6 +284,14 @@ public final class HadoopRpcProtos {
hash = (37 * hash) + REQUEST_FIELD_NUMBER;
hash = (53 * hash) + getRequest().hashCode();
}
+ if (hasDeclaringClassProtocolName()) {
+ hash = (37 * hash) + DECLARINGCLASSPROTOCOLNAME_FIELD_NUMBER;
+ hash = (53 * hash) + getDeclaringClassProtocolName().hashCode();
+ }
+ if (hasClientProtocolVersion()) {
+ hash = (37 * hash) + CLIENTPROTOCOLVERSION_FIELD_NUMBER;
+ hash = (53 * hash) + hashLong(getClientProtocolVersion());
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
return hash;
}
@@ -303,6 +412,10 @@ public final class HadoopRpcProtos {
bitField0_ = (bitField0_ & ~0x00000001);
request_ = com.google.protobuf.ByteString.EMPTY;
bitField0_ = (bitField0_ & ~0x00000002);
+ declaringClassProtocolName_ = "";
+ bitField0_ = (bitField0_ & ~0x00000004);
+ clientProtocolVersion_ = 0L;
+ bitField0_ = (bitField0_ & ~0x00000008);
return this;
}
@@ -349,6 +462,14 @@ public final class HadoopRpcProtos {
to_bitField0_ |= 0x00000002;
}
result.request_ = request_;
+ if (((from_bitField0_ & 0x00000004) == 0x00000004)) {
+ to_bitField0_ |= 0x00000004;
+ }
+ result.declaringClassProtocolName_ = declaringClassProtocolName_;
+ if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+ to_bitField0_ |= 0x00000008;
+ }
+ result.clientProtocolVersion_ = clientProtocolVersion_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -371,6 +492,12 @@ public final class HadoopRpcProtos {
if (other.hasRequest()) {
setRequest(other.getRequest());
}
+ if (other.hasDeclaringClassProtocolName()) {
+ setDeclaringClassProtocolName(other.getDeclaringClassProtocolName());
+ }
+ if (other.hasClientProtocolVersion()) {
+ setClientProtocolVersion(other.getClientProtocolVersion());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -380,6 +507,14 @@ public final class HadoopRpcProtos {
return false;
}
+ if (!hasDeclaringClassProtocolName()) {
+
+ return false;
+ }
+ if (!hasClientProtocolVersion()) {
+
+ return false;
+ }
return true;
}
@@ -416,6 +551,16 @@ public final class HadoopRpcProtos {
request_ = input.readBytes();
break;
}
+ case 26: {
+ bitField0_ |= 0x00000004;
+ declaringClassProtocolName_ = input.readBytes();
+ break;
+ }
+ case 32: {
+ bitField0_ |= 0x00000008;
+ clientProtocolVersion_ = input.readUInt64();
+ break;
+ }
}
}
}
@@ -482,6 +627,63 @@ public final class HadoopRpcProtos {
return this;
}
+ // required string declaringClassProtocolName = 3;
+ private java.lang.Object declaringClassProtocolName_ = "";
+ public boolean hasDeclaringClassProtocolName() {
+ return ((bitField0_ & 0x00000004) == 0x00000004);
+ }
+ public String getDeclaringClassProtocolName() {
+ java.lang.Object ref = declaringClassProtocolName_;
+ if (!(ref instanceof String)) {
+ String s = ((com.google.protobuf.ByteString) ref).toStringUtf8();
+ declaringClassProtocolName_ = s;
+ return s;
+ } else {
+ return (String) ref;
+ }
+ }
+ public Builder setDeclaringClassProtocolName(String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000004;
+ declaringClassProtocolName_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearDeclaringClassProtocolName() {
+ bitField0_ = (bitField0_ & ~0x00000004);
+ declaringClassProtocolName_ = getDefaultInstance().getDeclaringClassProtocolName();
+ onChanged();
+ return this;
+ }
+ void setDeclaringClassProtocolName(com.google.protobuf.ByteString value) {
+ bitField0_ |= 0x00000004;
+ declaringClassProtocolName_ = value;
+ onChanged();
+ }
+
+ // required uint64 clientProtocolVersion = 4;
+ private long clientProtocolVersion_ ;
+ public boolean hasClientProtocolVersion() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public long getClientProtocolVersion() {
+ return clientProtocolVersion_;
+ }
+ public Builder setClientProtocolVersion(long value) {
+ bitField0_ |= 0x00000008;
+ clientProtocolVersion_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearClientProtocolVersion() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ clientProtocolVersion_ = 0L;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:HadoopRpcRequestProto)
}
@@ -1706,16 +1908,18 @@ public final class HadoopRpcProtos {
descriptor;
static {
java.lang.String[] descriptorData = {
- "\n\020hadoop_rpc.proto\"<\n\025HadoopRpcRequestPr" +
+ "\n\020hadoop_rpc.proto\"\177\n\025HadoopRpcRequestPr" +
"oto\022\022\n\nmethodName\030\001 \002(\t\022\017\n\007request\030\002 \001(\014" +
- "\"D\n\027HadoopRpcExceptionProto\022\025\n\rexception" +
- "Name\030\001 \001(\t\022\022\n\nstackTrace\030\002 \001(\t\"\272\001\n\026Hadoo" +
- "pRpcResponseProto\0226\n\006status\030\001 \002(\0162&.Hado" +
- "opRpcResponseProto.ResponseStatus\022\020\n\010res" +
- "ponse\030\002 \001(\014\022+\n\texception\030\003 \001(\0132\030.HadoopR" +
- "pcExceptionProto\")\n\016ResponseStatus\022\013\n\007SU" +
- "CCESS\020\001\022\n\n\006ERRROR\020\002B4\n\036org.apache.hadoop" +
- ".ipc.protobufB\017HadoopRpcProtos\240\001\001"
+ "\022\"\n\032declaringClassProtocolName\030\003 \002(\t\022\035\n\025" +
+ "clientProtocolVersion\030\004 \002(\004\"D\n\027HadoopRpc" +
+ "ExceptionProto\022\025\n\rexceptionName\030\001 \001(\t\022\022\n" +
+ "\nstackTrace\030\002 \001(\t\"\272\001\n\026HadoopRpcResponseP" +
+ "roto\0226\n\006status\030\001 \002(\0162&.HadoopRpcResponse" +
+ "Proto.ResponseStatus\022\020\n\010response\030\002 \001(\014\022+" +
+ "\n\texception\030\003 \001(\0132\030.HadoopRpcExceptionPr" +
+ "oto\")\n\016ResponseStatus\022\013\n\007SUCCESS\020\001\022\n\n\006ER",
+ "RROR\020\002B4\n\036org.apache.hadoop.ipc.protobuf" +
+ "B\017HadoopRpcProtos\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -1727,7 +1931,7 @@ public final class HadoopRpcProtos {
internal_static_HadoopRpcRequestProto_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_HadoopRpcRequestProto_descriptor,
- new java.lang.String[] { "MethodName", "Request", },
+ new java.lang.String[] { "MethodName", "Request", "DeclaringClassProtocolName", "ClientProtocolVersion", },
org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto.class,
org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto.Builder.class);
internal_static_HadoopRpcExceptionProto_descriptor =
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto Thu Jan 12 02:54:01 2012
@@ -34,6 +34,12 @@ message HadoopRpcRequestProto {
/** Bytes corresponding to the client protobuf request */
optional bytes request = 2;
+
+ /** protocol name of class declaring the called method */
+ required string declaringClassProtocolName = 3;
+
+ /** protocol version of class declaring the called method */
+ required uint64 clientProtocolVersion = 4;
}
/**
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java Thu Jan 12 02:54:01 2012
@@ -34,6 +34,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.TestSaslRPC.CustomSecurityInfo;
import org.apache.hadoop.ipc.TestSaslRPC.TestTokenIdentifier;
import org.apache.hadoop.ipc.TestSaslRPC.TestTokenSecretManager;
@@ -101,7 +102,8 @@ public class TestAvroRpc extends TestCas
RPC.setProtocolEngine(conf, AvroTestProtocol.class, AvroRpcEngine.class);
RPC.Server server = RPC.getServer(EmptyProtocol.class, new EmptyImpl(),
ADDRESS, 0, 5, true, conf, sm);
- server.addProtocol(AvroTestProtocol.class, new TestImpl());
+ server.addProtocol(RpcKind.RPC_WRITABLE,
+ AvroTestProtocol.class, new TestImpl());
try {
server.start();
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java Thu Jan 12 02:54:01 2012
@@ -23,6 +23,7 @@ import org.apache.commons.logging.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.net.NetUtils;
@@ -96,8 +97,8 @@ public class TestIPC {
}
@Override
- public Writable call(String protocol, Writable param, long receiveTime)
- throws IOException {
+ public Writable call(RpcKind rpcKind, String protocol, Writable param,
+ long receiveTime) throws IOException {
if (sleep) {
// sleep a bit
try {
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java Thu Jan 12 02:54:01 2012
@@ -30,6 +30,7 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.NetUtils;
/**
@@ -72,8 +73,8 @@ public class TestIPCServerResponder exte
}
@Override
- public Writable call(String protocol, Writable param, long receiveTime)
- throws IOException {
+ public Writable call(RpcKind rpcKind, String protocol, Writable param,
+ long receiveTime) throws IOException {
if (sleep) {
try {
Thread.sleep(RANDOM.nextInt(20)); // sleep a bit
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java Thu Jan 12 02:54:01 2012
@@ -23,10 +23,15 @@ import java.net.InetSocketAddress;
import org.junit.Assert;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
+import org.apache.hadoop.ipc.TestProtoBufRpc.PBServerImpl;
+import org.apache.hadoop.ipc.TestProtoBufRpc.TestRpcService;
+import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpcProto;
import org.apache.hadoop.net.NetUtils;
import org.junit.Before;
import org.junit.After;
import org.junit.Test;
+import com.google.protobuf.BlockingService;
public class TestMultipleProtocolServer {
private static final String ADDRESS = "0.0.0.0";
@@ -173,9 +178,19 @@ public class TestMultipleProtocolServer
// create a server with two handlers
server = RPC.getServer(Foo0.class,
new Foo0Impl(), ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(Foo1.class, new Foo1Impl());
- server.addProtocol(Bar.class, new BarImpl());
- server.addProtocol(Mixin.class, new BarImpl());
+ server.addProtocol(RpcKind.RPC_WRITABLE, Foo1.class, new Foo1Impl());
+ server.addProtocol(RpcKind.RPC_WRITABLE, Bar.class, new BarImpl());
+ server.addProtocol(RpcKind.RPC_WRITABLE, Mixin.class, new BarImpl());
+
+
+ // Add Protobuf server
+ // Create server side implementation
+ PBServerImpl pbServerImpl =
+ new PBServerImpl();
+ BlockingService service = TestProtobufRpcProto
+ .newReflectiveBlockingService(pbServerImpl);
+ server.addProtocol(RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService.class,
+ service);
server.start();
addr = NetUtils.getConnectAddress(server);
}
@@ -251,5 +266,16 @@ public class TestMultipleProtocolServer
public void testIncorrectServerCreation() throws IOException {
RPC.getServer(Foo1.class,
new Foo0Impl(), ADDRESS, 0, 2, false, conf, null);
+ }
+
+ // Now test a PB service - a server hosts both PB and Writable Rpcs.
+ @Test
+ public void testPBService() throws Exception {
+ // Set RPC engine to protobuf RPC engine
+ Configuration conf2 = new Configuration();
+ RPC.setProtocolEngine(conf2, TestRpcService.class,
+ ProtobufRpcEngine.class);
+ TestRpcService client = RPC.getProxy(TestRpcService.class, 0, addr, conf2);
+ TestProtoBufRpc.testProtoBufRpc(client);
}
}
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java Thu Jan 12 02:54:01 2012
@@ -21,14 +21,18 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto;
import org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto;
import org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto;
import org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto;
import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpcProto;
-import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface;
+import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpc2Proto;
+import org.apache.hadoop.net.NetUtils;
import org.junit.Assert;
import org.junit.Test;
+import org.junit.Before;
+import org.junit.After;
import com.google.protobuf.BlockingService;
import com.google.protobuf.RpcController;
@@ -42,8 +46,21 @@ import com.google.protobuf.ServiceExcept
public class TestProtoBufRpc {
public final static String ADDRESS = "0.0.0.0";
public final static int PORT = 0;
+ private static InetSocketAddress addr;
+ private static Configuration conf;
+ private static RPC.Server server;
+
+ @ProtocolInfo(protocolName = "testProto", protocolVersion = 1)
+ public interface TestRpcService
+ extends TestProtobufRpcProto.BlockingInterface {
+ }
+
+ @ProtocolInfo(protocolName = "testProto2", protocolVersion = 1)
+ public interface TestRpcService2 extends
+ TestProtobufRpc2Proto.BlockingInterface {
+ }
- public static class ServerImpl implements BlockingInterface {
+ public static class PBServerImpl implements TestRpcService {
@Override
public EmptyResponseProto ping(RpcController unused,
@@ -64,40 +81,78 @@ public class TestProtoBufRpc {
throw new ServiceException("error", new RpcServerException("error"));
}
}
+
+ public static class PBServer2Impl implements TestRpcService2 {
- private static RPC.Server startRPCServer(Configuration conf)
- throws IOException {
+ @Override
+ public EmptyResponseProto ping2(RpcController unused,
+ EmptyRequestProto request) throws ServiceException {
+ return EmptyResponseProto.newBuilder().build();
+ }
+
+ @Override
+ public EchoResponseProto echo2(RpcController unused, EchoRequestProto request)
+ throws ServiceException {
+ return EchoResponseProto.newBuilder().setMessage(request.getMessage())
+ .build();
+ }
+ }
+
+ @Before
+ public void setUp() throws IOException { // Setup server for both protocols
+ conf = new Configuration();
// Set RPC engine to protobuf RPC engine
- RPC.setProtocolEngine(conf, BlockingService.class, ProtobufRpcEngine.class);
+ RPC.setProtocolEngine(conf, TestRpcService.class, ProtobufRpcEngine.class);
// Create server side implementation
- ServerImpl serverImpl = new ServerImpl();
+ PBServerImpl serverImpl = new PBServerImpl();
BlockingService service = TestProtobufRpcProto
.newReflectiveBlockingService(serverImpl);
- // Get RPC server for serer side implementation
- RPC.Server server = RPC.getServer(BlockingService.class, service, ADDRESS,
- PORT, conf);
+ // Get RPC server for server side implementation
+ server = RPC.getServer(TestRpcService.class, service, ADDRESS, PORT, conf);
+ addr = NetUtils.getConnectAddress(server);
+
+ // now the second protocol
+ PBServer2Impl server2Impl = new PBServer2Impl();
+ BlockingService service2 = TestProtobufRpc2Proto
+ .newReflectiveBlockingService(server2Impl);
+
+ server.addProtocol(RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService2.class,
+ service2);
server.start();
- return server;
+ }
+
+
+ @After
+ public void tearDown() throws Exception {
+ server.stop();
}
- private static BlockingInterface getClient(Configuration conf,
- InetSocketAddress addr) throws IOException {
+ private static TestRpcService getClient() throws IOException {
+ // Set RPC engine to protobuf RPC engine
+ RPC.setProtocolEngine(conf, TestRpcService.class,
+ ProtobufRpcEngine.class);
+ return RPC.getProxy(TestRpcService.class, 0, addr,
+ conf);
+ }
+
+ private static TestRpcService2 getClient2() throws IOException {
// Set RPC engine to protobuf RPC engine
- RPC.setProtocolEngine(conf, BlockingInterface.class,
+ RPC.setProtocolEngine(conf, TestRpcService2.class,
ProtobufRpcEngine.class);
- BlockingInterface client = RPC.getProxy(BlockingInterface.class, 0, addr,
+ return RPC.getProxy(TestRpcService2.class, 0, addr,
conf);
- return client;
}
@Test
public void testProtoBufRpc() throws Exception {
- Configuration conf = new Configuration();
- RPC.Server server = startRPCServer(conf);
- BlockingInterface client = getClient(conf, server.getListenerAddress());
-
+ TestRpcService client = getClient();
+ testProtoBufRpc(client);
+ }
+
+ // separated test out so that other tests can call it.
+ public static void testProtoBufRpc(TestRpcService client) throws Exception {
// Test ping method
EmptyRequestProto emptyRequest = EmptyRequestProto.newBuilder().build();
client.ping(null, emptyRequest);
@@ -108,16 +163,29 @@ public class TestProtoBufRpc {
EchoResponseProto echoResponse = client.echo(null, echoRequest);
Assert.assertEquals(echoResponse.getMessage(), "hello");
- // Test error method - it should be thrown as RemoteException
+ // Test error method - error should be thrown as RemoteException
try {
client.error(null, emptyRequest);
Assert.fail("Expected exception is not thrown");
} catch (ServiceException e) {
RemoteException re = (RemoteException)e.getCause();
- re.printStackTrace();
RpcServerException rse = (RpcServerException) re
.unwrapRemoteException(RpcServerException.class);
- rse.printStackTrace();
}
}
+
+ @Test
+ public void testProtoBufRpc2() throws Exception {
+ TestRpcService2 client = getClient2();
+
+ // Test ping method
+ EmptyRequestProto emptyRequest = EmptyRequestProto.newBuilder().build();
+ client.ping2(null, emptyRequest);
+
+ // Test echo method
+ EchoRequestProto echoRequest = EchoRequestProto.newBuilder()
+ .setMessage("hello").build();
+ EchoResponseProto echoResponse = client.echo2(null, echoRequest);
+ Assert.assertEquals(echoResponse.getMessage(), "hello");
+ }
}
\ No newline at end of file
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java Thu Jan 12 02:54:01 2012
@@ -31,6 +31,7 @@ import junit.framework.Assert;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.NetUtils;
import org.junit.After;
import org.junit.Test;
@@ -56,6 +57,8 @@ public class TestRPCCompatibility {
String echo(String value) throws IOException;
}
+
+ // TestProtocol2 is a compatible impl of TestProtocol1 - hence use its name
@ProtocolInfo(protocolName=
"org.apache.hadoop.ipc.TestRPCCompatibility$TestProtocol1")
public interface TestProtocol2 extends TestProtocol1 {
@@ -114,9 +117,11 @@ public class TestRPCCompatibility {
public void tearDown() throws IOException {
if (proxy != null) {
RPC.stopProxy(proxy.getProxy());
+ proxy = null;
}
if (server != null) {
server.stop();
+ server = null;
}
}
@@ -126,7 +131,7 @@ public class TestRPCCompatibility {
TestImpl1 impl = new TestImpl1();
server = RPC.getServer(TestProtocol1.class,
impl, ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(TestProtocol0.class, impl);
+ server.addProtocol(RpcKind.RPC_WRITABLE, TestProtocol0.class, impl);
server.start();
addr = NetUtils.getConnectAddress(server);
@@ -170,8 +175,10 @@ public class TestRPCCompatibility {
public int echo(int value) throws IOException, NumberFormatException {
if (serverInfo.isMethodSupported("echo", int.class)) {
+System.out.println("echo int is supported");
return -value; // use version 3 echo long
} else { // server is version 2
+System.out.println("echo int is NOT supported");
return Integer.parseInt(proxy2.echo(String.valueOf(value)));
}
}
@@ -191,7 +198,7 @@ public class TestRPCCompatibility {
TestImpl1 impl = new TestImpl1();
server = RPC.getServer(TestProtocol1.class,
impl, ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(TestProtocol0.class, impl);
+ server.addProtocol(RpcKind.RPC_WRITABLE, TestProtocol0.class, impl);
server.start();
addr = NetUtils.getConnectAddress(server);
@@ -207,11 +214,12 @@ public class TestRPCCompatibility {
@Test // equal version client and server
public void testVersion2ClientVersion2Server() throws Exception {
+ ProtocolSignature.resetCache();
// create a server with two handlers
TestImpl2 impl = new TestImpl2();
server = RPC.getServer(TestProtocol2.class,
impl, ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(TestProtocol0.class, impl);
+ server.addProtocol(RpcKind.RPC_WRITABLE, TestProtocol0.class, impl);
server.start();
addr = NetUtils.getConnectAddress(server);
Modified: hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto?rev=1230378&r1=1230377&r2=1230378&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto (original)
+++ hadoop/common/branches/branch-0.23-PB/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto Thu Jan 12 02:54:01 2012
@@ -31,3 +31,8 @@ service TestProtobufRpcProto {
rpc echo(EchoRequestProto) returns (EchoResponseProto);
rpc error(EmptyRequestProto) returns (EmptyResponseProto);
}
+
+service TestProtobufRpc2Proto {
+ rpc ping2(EmptyRequestProto) returns (EmptyResponseProto);
+ rpc echo2(EchoRequestProto) returns (EchoResponseProto);
+}