Encodes an RPC and sends it downstream (to the wire). This method can be called from any thread so it needs to be thread-safe. @param rpc The RPC to send downstream. @return The buffer to write to the channel or null if there was an error and there's nothing to write.
(final HBaseRpc rpc)
| 1278 | * an error and there's nothing to write. |
| 1279 | */ |
| 1280 | private ChannelBuffer encode(final HBaseRpc rpc) { |
| 1281 | if (!rpc.hasDeferred()) { |
| 1282 | throw new AssertionError("Should never happen! rpc=" + rpc); |
| 1283 | } |
| 1284 | |
| 1285 | // TODO(tsuna): Add rate-limiting here. We don't want to send more than |
| 1286 | // N QPS to a given region server. |
| 1287 | // TODO(tsuna): Check the size() of rpcs_inflight. We don't want to have |
| 1288 | // more than M RPCs in flight at the same time, and we may be overwhelming |
| 1289 | // the server if we do. |
| 1290 | |
| 1291 | rpc.rpc_id = this.rpcid.incrementAndGet(); |
| 1292 | ChannelBuffer payload; |
| 1293 | try { |
| 1294 | payload = rpc.serialize(server_version); |
| 1295 | // We assume that payload has enough bytes at the beginning for us to |
| 1296 | // "fill in the blanks" and put the RPC header. This is accounted for |
| 1297 | // automatically by HBaseRpc#newBuffer. If someone creates their own |
| 1298 | // buffer without this extra space at the beginning, we're going to |
| 1299 | // corrupt the RPC at this point. |
| 1300 | final byte[] method = rpc.method(server_version); |
| 1301 | if (server_version >= SERVER_VERSION_095_OR_ABOVE) { |
| 1302 | final RPCPB.RequestHeader header = RPCPB.RequestHeader.newBuilder() |
| 1303 | .setCallId(rpc.rpc_id) // 1 + 1-to-5 bytes (vint) |
| 1304 | .setMethodNameBytes(Bytes.wrap(method)) // 1 + 1 + N bytes |
| 1305 | .setRequestParam(true) // 1 + 1 bytes |
| 1306 | .build(); |
| 1307 | final int pblen = header.getSerializedSize(); |
| 1308 | // In HBaseRpc.newBuffer() we reserved 19 bytes for the RPC header |
| 1309 | // (without counting the leading 4 bytes for the overall size). |
| 1310 | // Here the size is variable due to the nature of the protobuf |
| 1311 | // encoding, but the expected absolute maximum size is 17 bytes |
| 1312 | // if we ignore the method name. So we have to offset the header |
| 1313 | // by 2 to 13 bytes typically. Note that the "-1" is for the varint |
| 1314 | // that's at the beginning of the header that indicates how long the |
| 1315 | // header itself is. |
| 1316 | final int offset = 19 + method.length - pblen - 1; |
| 1317 | assert offset >= 0 : ("RPC header too big (" + pblen + " bytes): " |
| 1318 | + header); |
| 1319 | // Skip the few extraneous bytes we over-allocated for the header. |
| 1320 | payload.readerIndex(offset); |
| 1321 | // The first int is the size of the message, excluding the 4 bytes |
| 1322 | // needed for the size itself, hence the `-4'. |
| 1323 | payload.setInt(offset, payload.readableBytes() - 4); // 4 bytes |
| 1324 | try { |
| 1325 | final CodedOutputStream output = |
| 1326 | CodedOutputStream.newInstance(payload.array(), 4 + offset, |
| 1327 | 1 + pblen); |
| 1328 | output.writeRawByte(pblen); // varint but always on 1 byte here. |
| 1329 | header.writeTo(output); |
| 1330 | output.checkNoSpaceLeft(); |
| 1331 | } catch (IOException e) { |
| 1332 | throw new RuntimeException("Should never happen", e); |
| 1333 | } |
| 1334 | } else if (server_version >= SERVER_VERSION_092_OR_ABOVE) { |
| 1335 | // The first int is the size of the message, excluding the 4 bytes |
| 1336 | // needed for the size itself, hence the `-4'. |
| 1337 | payload.setInt(0, payload.readableBytes() - 4); // 4 bytes |