param, to the IPC server running at
* address, returning the value. Throws exceptions if there are
* network problems or if the remote code threw an exception.
- * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, Writable,
+ * @deprecated Use {@link #call(RPC.RpcKind, Writable,
* ConnectionId)} instead
*/
@Deprecated
- public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress address)
+ public Writable call(RPC.RpcKind rpcKind, Writable param, InetSocketAddress address)
throws InterruptedException, IOException {
return call(rpcKind, param, address, null);
}
@@ -1047,11 +1050,11 @@ public class Client {
* the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, Writable,
+ * @deprecated Use {@link #call(RPC.RpcKind, Writable,
* ConnectionId)} instead
*/
@Deprecated
- public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
+ public Writable call(RPC.RpcKind rpcKind, Writable param, InetSocketAddress addr,
UserGroupInformation ticket)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, null, ticket, 0,
@@ -1065,11 +1068,11 @@ public class Client {
* timeout, returning the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, Writable,
+ * @deprecated Use {@link #call(RPC.RpcKind, Writable,
* ConnectionId)} instead
*/
@Deprecated
- public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
+ public Writable call(RPC.RpcKind rpcKind, Writable param, InetSocketAddress addr,
Class protocol, UserGroupInformation ticket,
int rpcTimeout)
throws InterruptedException, IOException {
@@ -1080,7 +1083,7 @@ public class Client {
/**
- * Same as {@link #call(RpcPayloadHeader.RpcKind, Writable, InetSocketAddress,
+ * Same as {@link #call(RPC.RpcKind, Writable, InetSocketAddress,
* Class, UserGroupInformation, int, Configuration)}
* except that rpcKind is writable.
*/
@@ -1090,7 +1093,7 @@ public class Client {
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
ticket, rpcTimeout, conf);
- return call(RpcKind.RPC_BUILTIN, param, remoteId);
+ return call(RPC.RpcKind.RPC_BUILTIN, param, remoteId);
}
/**
@@ -1101,7 +1104,7 @@ public class Client {
* value. Throws exceptions if there are network problems or if the remote
* code threw an exception.
*/
- public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
+ public Writable call(RPC.RpcKind rpcKind, Writable param, InetSocketAddress addr,
Class protocol, UserGroupInformation ticket,
int rpcTimeout, Configuration conf)
throws InterruptedException, IOException {
@@ -1111,12 +1114,12 @@ public class Client {
}
/**
- * Same as {link {@link #call(RpcPayloadHeader.RpcKind, Writable, ConnectionId)}
+ * Same as {link {@link #call(RPC.RpcKind, Writable, ConnectionId)}
* except the rpcKind is RPC_BUILTIN
*/
public Writable call(Writable param, ConnectionId remoteId)
throws InterruptedException, IOException {
- return call(RpcKind.RPC_BUILTIN, param, remoteId);
+ return call(RPC.RpcKind.RPC_BUILTIN, param, remoteId);
}
/**
@@ -1130,7 +1133,7 @@ public class Client {
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
*/
- public Writable call(RpcKind rpcKind, Writable rpcRequest,
+ public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
ConnectionId remoteId) throws InterruptedException, IOException {
Call call = new Call(rpcKind, rpcRequest);
Connection connection = getConnection(remoteId, call);
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
index befc8f70e03..2d3f91e5e4c 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
@@ -38,7 +38,6 @@ import org.apache.hadoop.io.DataOutputOutputStream;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.ipc.Client.ConnectionId;
import org.apache.hadoop.ipc.RPC.RpcInvoker;
-import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto;
import org.apache.hadoop.security.UserGroupInformation;
@@ -61,7 +60,7 @@ public class ProtobufRpcEngine implements RpcEngine {
static { // Register the rpcRequest deserializer for WritableRpcEngine
org.apache.hadoop.ipc.Server.registerProtocolEngine(
- RpcKind.RPC_PROTOCOL_BUFFER, RpcRequestWritable.class,
+ RPC.RpcKind.RPC_PROTOCOL_BUFFER, RpcRequestWritable.class,
new Server.ProtoBufRpcInvoker());
}
@@ -182,7 +181,7 @@ public class ProtobufRpcEngine implements RpcEngine {
HadoopRpcRequestProto rpcRequest = constructRpcRequest(method, args);
RpcResponseWritable val = null;
try {
- val = (RpcResponseWritable) client.call(RpcKind.RPC_PROTOCOL_BUFFER,
+ val = (RpcResponseWritable) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
new RpcRequestWritable(rpcRequest), remoteId);
} catch (Throwable e) {
throw new ServiceException(e);
@@ -351,7 +350,7 @@ public class ProtobufRpcEngine implements RpcEngine {
numReaders, queueSizePerHandler, conf, classNameBase(protocolImpl
.getClass().getName()), secretManager, portRangeConfig);
this.verbose = verbose;
- registerProtocolAndImpl(RpcKind.RPC_PROTOCOL_BUFFER, protocolClass,
+ registerProtocolAndImpl(RPC.RpcKind.RPC_PROTOCOL_BUFFER, protocolClass,
protocolImpl);
}
@@ -363,10 +362,10 @@ public class ProtobufRpcEngine implements RpcEngine {
String protoName, long version) throws IOException {
ProtoNameVer pv = new ProtoNameVer(protoName, version);
ProtoClassProtoImpl impl =
- server.getProtocolImplMap(RpcKind.RPC_PROTOCOL_BUFFER).get(pv);
+ server.getProtocolImplMap(RPC.RpcKind.RPC_PROTOCOL_BUFFER).get(pv);
if (impl == null) { // no match for Protocol AND Version
VerProtocolImpl highest =
- server.getHighestSupportedProtocol(RpcKind.RPC_PROTOCOL_BUFFER,
+ server.getHighestSupportedProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
protoName);
if (highest == null) {
throw new IOException("Unknown protocol: " + protoName);
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtocolMetaInfoServerSideTranslatorPB.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtocolMetaInfoServerSideTranslatorPB.java
index aaf71f8a4e7..d9d80a84d86 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtocolMetaInfoServerSideTranslatorPB.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtocolMetaInfoServerSideTranslatorPB.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.ipc;
import org.apache.hadoop.ipc.RPC.Server.VerProtocolImpl;
-import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.ProtocolInfoProtos.GetProtocolSignatureRequestProto;
import org.apache.hadoop.ipc.protobuf.ProtocolInfoProtos.GetProtocolSignatureResponseProto;
import org.apache.hadoop.ipc.protobuf.ProtocolInfoProtos.GetProtocolVersionsRequestProto;
@@ -49,7 +48,7 @@ public class ProtocolMetaInfoServerSideTranslatorPB implements
String protocol = request.getProtocol();
GetProtocolVersionsResponseProto.Builder builder =
GetProtocolVersionsResponseProto.newBuilder();
- for (RpcKind r : RpcKind.values()) {
+ for (RPC.RpcKind r : RPC.RpcKind.values()) {
long[] versions;
try {
versions = getProtocolVersionForRpcKind(r, protocol);
@@ -78,7 +77,7 @@ public class ProtocolMetaInfoServerSideTranslatorPB implements
String rpcKind = request.getRpcKind();
long[] versions;
try {
- versions = getProtocolVersionForRpcKind(RpcKind.valueOf(rpcKind),
+ versions = getProtocolVersionForRpcKind(RPC.RpcKind.valueOf(rpcKind),
protocol);
} catch (ClassNotFoundException e1) {
throw new ServiceException(e1);
@@ -104,7 +103,7 @@ public class ProtocolMetaInfoServerSideTranslatorPB implements
return builder.build();
}
- private long[] getProtocolVersionForRpcKind(RpcKind rpcKind,
+ private long[] getProtocolVersionForRpcKind(RPC.RpcKind rpcKind,
String protocol) throws ClassNotFoundException {
Class protocolClass = Class.forName(protocol);
String protocolName = RPC.getProtocolName(protocolClass);
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
index d0f268ec5d2..56fbd7d5a1b 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
@@ -42,7 +42,6 @@ import org.apache.commons.logging.*;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.io.*;
import org.apache.hadoop.ipc.Client.ConnectionId;
-import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.ProtocolInfoProtos.ProtocolInfoService;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.SaslRpcServer;
@@ -73,6 +72,18 @@ import com.google.protobuf.BlockingService;
* the protocol instance is transmitted.
*/
public class RPC {
+ public enum RpcKind {
+ RPC_BUILTIN ((short) 1), // Used for built in calls by tests
+ RPC_WRITABLE ((short) 2), // Use WritableRpcEngine
+ RPC_PROTOCOL_BUFFER ((short) 3); // Use ProtobufRpcEngine
+ final static short MAX_INDEX = RPC_PROTOCOL_BUFFER.value; // used for array size
+ private static final short FIRST_INDEX = RPC_BUILTIN.value;
+ public final short value; //TODO make it private
+
+ RpcKind(short val) {
+ this.value = val;
+ }
+ }
interface RpcInvoker {
/**
@@ -777,7 +788,7 @@ public class RPC {
ArrayList
- * The format of RPC call is as follows:
- * +---------------------------------------------------+
- * | Rpc length in bytes (header + payload length) |
- * +---------------------------------------------------+
- * | Rpc Header | Rpc Payload |
- * +---------------------------------------------------+
- *
- * The format of Rpc Header is:
- * +----------------------------------+
- * | RpcKind (1 bytes) |
- * +----------------------------------+
- * | RpcPayloadOperation (1 bytes) |
- * +----------------------------------+
- * | Call ID (4 bytes) |
- * +----------------------------------+
- *
- * {@link RpcKind} determines the type of serialization used for Rpc Payload.
- *
- *
- * Note this header does NOT have its own version number,
- * it used the version number from the connection header.
- */
-public class RpcPayloadHeader implements Writable {
- public enum RpcPayloadOperation {
- RPC_FINAL_PAYLOAD ((short)1),
- RPC_CONTINUATION_PAYLOAD ((short)2), // not implemented yet
- RPC_CLOSE_CONNECTION ((short)3); // close the rpc connection
-
- private final short code;
- private static final short FIRST_INDEX = RPC_FINAL_PAYLOAD.code;
- RpcPayloadOperation(short val) {
- this.code = val;
- }
-
- public void write(DataOutput out) throws IOException {
- out.writeByte(code);
- }
-
- static RpcPayloadOperation readFields(DataInput in) throws IOException {
- short inValue = in.readByte();
- return RpcPayloadOperation.values()[inValue - FIRST_INDEX];
- }
- }
-
- public enum RpcKind {
- RPC_BUILTIN ((short) 1), // Used for built in calls by tests
- RPC_WRITABLE ((short) 2), // Use WritableRpcEngine
- RPC_PROTOCOL_BUFFER ((short) 3); // Use ProtobufRpcEngine
- final static short MAX_INDEX = RPC_PROTOCOL_BUFFER.value; // used for array size
- private static final short FIRST_INDEX = RPC_BUILTIN.value;
- private final short value;
-
- RpcKind(short val) {
- this.value = val;
- }
-
- public void write(DataOutput out) throws IOException {
- out.writeByte(value);
- }
-
- static RpcKind readFields(DataInput in) throws IOException {
- short inValue = in.readByte();
- return RpcKind.values()[inValue - FIRST_INDEX];
- }
- }
-
- private RpcKind kind;
- private RpcPayloadOperation operation;
- private int callId;
-
- public RpcPayloadHeader() {
- kind = RpcKind.RPC_WRITABLE;
- operation = RpcPayloadOperation.RPC_CLOSE_CONNECTION;
- }
-
- public RpcPayloadHeader(RpcKind kind, RpcPayloadOperation op, int callId) {
- this.kind = kind;
- this.operation = op;
- this.callId = callId;
- }
-
- int getCallId() {
- return callId;
- }
-
- RpcKind getkind() {
- return kind;
- }
-
- RpcPayloadOperation getOperation() {
- return operation;
- }
-
- @Override
- public void write(DataOutput out) throws IOException {
- kind.write(out);
- operation.write(out);
- out.writeInt(callId);
- }
-
- @Override
- public void readFields(DataInput in) throws IOException {
- kind = RpcKind.readFields(in);
- operation = RpcPayloadOperation.readFields(in);
- this.callId = in.readInt();
- }
-}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
index 8628bb0c485..24c2f7beaa7 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
@@ -72,11 +72,10 @@ import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.ipc.RPC.RpcInvoker;
import org.apache.hadoop.ipc.RPC.VersionMismatch;
-import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
-import org.apache.hadoop.ipc.RpcPayloadHeader.RpcPayloadOperation;
import org.apache.hadoop.ipc.metrics.RpcDetailedMetrics;
import org.apache.hadoop.ipc.metrics.RpcMetrics;
import org.apache.hadoop.ipc.protobuf.IpcConnectionContextProtos.IpcConnectionContextProto;
+import org.apache.hadoop.ipc.protobuf.RpcPayloadHeaderProtos.*;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.SaslRpcServer;
@@ -170,8 +169,8 @@ public abstract class Server {
this.rpcRequestWrapperClass = rpcRequestWrapperClass;
}
}
- static Map