rpc-darpc/src/main/java/org/apache/crail/namenode/rpc/darpc/DaRPCNameNodeConnection.java - incubator-crail - Git at Google

 /*
  *
  * 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.
  */

 package org.apache.crail.namenode.rpc.darpc;

 import java.io.IOException;
 import java.net.InetAddress;

 import org.apache.crail.CrailNodeType;
 import org.apache.crail.conf.CrailConstants;
 import org.apache.crail.metadata.BlockInfo;
 import org.apache.crail.metadata.DataNodeInfo;
 import org.apache.crail.metadata.FileInfo;
 import org.apache.crail.metadata.FileName;
 import org.apache.crail.rpc.RpcConnection;
 import org.apache.crail.rpc.RpcCreateFile;
 import org.apache.crail.rpc.RpcDeleteFile;
 import org.apache.crail.rpc.RpcFuture;
 import org.apache.crail.rpc.RpcGetBlock;
 import org.apache.crail.rpc.RpcGetDataNode;
 import org.apache.crail.rpc.RpcGetFile;
 import org.apache.crail.rpc.RpcGetLocation;
 import org.apache.crail.rpc.RpcPing;
 import org.apache.crail.rpc.RpcRemoveDataNode;
 import org.apache.crail.rpc.RpcProtocol;
 import org.apache.crail.rpc.RpcRenameFile;
 import org.apache.crail.rpc.RpcRequestMessage;
 import org.apache.crail.rpc.RpcResponseMessage;
 import org.apache.crail.rpc.RpcVoid;
 import org.apache.crail.utils.CrailUtils;
 import org.slf4j.Logger;

 import com.ibm.darpc.DaRPCClientEndpoint;
 import com.ibm.darpc.DaRPCFuture;
 import com.ibm.darpc.DaRPCStream;

 public class DaRPCNameNodeConnection implements RpcConnection {
 	private static final Logger LOG = CrailUtils.getLogger();

 	private DaRPCClientEndpoint<DaRPCNameNodeRequest, DaRPCNameNodeResponse> rpcEndpoint;
 	private DaRPCStream<DaRPCNameNodeRequest, DaRPCNameNodeResponse> stream;

 	public DaRPCNameNodeConnection(DaRPCClientEndpoint<DaRPCNameNodeRequest, DaRPCNameNodeResponse> endpoint) throws IOException {
 		this.rpcEndpoint = endpoint;
 		this.stream = endpoint.createStream();
 	}

 	@Override
 	public RpcFuture<RpcCreateFile> createFile(FileName filename, CrailNodeType type, int storageClass, int locationClass, boolean enumerable) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: createFile, fileType " + type + ", storageClass " + storageClass + ", locationClass " + locationClass);
 		}

 		RpcRequestMessage.CreateFileReq createFileReq = new RpcRequestMessage.CreateFileReq(filename, type, storageClass, locationClass, enumerable);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(createFileReq);
 		request.setCommand(RpcProtocol.CMD_CREATE_FILE);

 		RpcResponseMessage.CreateFileRes fileRes = new RpcResponseMessage.CreateFileRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(fileRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcCreateFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcCreateFile>(future, fileRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public RpcFuture<RpcGetFile> getFile(FileName filename, boolean writeable) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: getFile, writeable " + writeable);
 		}

 		RpcRequestMessage.GetFileReq getFileReq = new RpcRequestMessage.GetFileReq(filename, writeable);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getFileReq);
 		request.setCommand(RpcProtocol.CMD_GET_FILE);

 		RpcResponseMessage.GetFileRes fileRes = new RpcResponseMessage.GetFileRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(fileRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcGetFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetFile>(future, fileRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcVoid> setFile(FileInfo fileInfo, boolean close) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: setFile, id " + fileInfo.getFd() + ", close " + close);
 		}

 		RpcRequestMessage.SetFileReq setFileReq = new RpcRequestMessage.SetFileReq(fileInfo, close);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(setFileReq);
 		request.setCommand(RpcProtocol.CMD_SET_FILE);

 		RpcResponseMessage.VoidRes voidRes = new RpcResponseMessage.VoidRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(voidRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcVoid> nameNodeFuture = new DaRPCNameNodeFuture<RpcVoid>(future, voidRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcDeleteFile> removeFile(FileName filename, boolean recursive) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: removeFile");
 		}

 		RpcRequestMessage.RemoveFileReq removeReq = new RpcRequestMessage.RemoveFileReq(filename, recursive);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(removeReq);
 		request.setCommand(RpcProtocol.CMD_REMOVE_FILE);

 		RpcResponseMessage.DeleteFileRes fileRes = new RpcResponseMessage.DeleteFileRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(fileRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcDeleteFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcDeleteFile>(future, fileRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcRenameFile> renameFile(FileName srcHash, FileName dstHash) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: renameFile");
 		}

 		RpcRequestMessage.RenameFileReq renameReq = new RpcRequestMessage.RenameFileReq(srcHash, dstHash);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(renameReq);
 		request.setCommand(RpcProtocol.CMD_RENAME_FILE);

 		RpcResponseMessage.RenameRes renameRes = new RpcResponseMessage.RenameRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(renameRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcRenameFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcRenameFile>(future, renameRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcGetBlock> getBlock(long fd, long token, long position, long capacity) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: getBlock, fd " + fd + ", token " + token + ", position " + position + ", capacity " + capacity);
 		}

 		RpcRequestMessage.GetBlockReq getBlockReq = new RpcRequestMessage.GetBlockReq(fd, token, position, capacity);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getBlockReq);
 		request.setCommand(RpcProtocol.CMD_GET_BLOCK);

 		RpcResponseMessage.GetBlockRes getBlockRes = new RpcResponseMessage.GetBlockRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(getBlockRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcGetBlock> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetBlock>(future, getBlockRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcGetLocation> getLocation(FileName fileName, long position) throws IOException {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: getLocation, position " + position);
 		}

 		RpcRequestMessage.GetLocationReq getLocationReq = new RpcRequestMessage.GetLocationReq(fileName, position);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getLocationReq);
 		request.setCommand(RpcProtocol.CMD_GET_LOCATION);

 		RpcResponseMessage.GetLocationRes getLocationRes = new RpcResponseMessage.GetLocationRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(getLocationRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcGetLocation> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetLocation>(future, getLocationRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcVoid> setBlock(BlockInfo blockInfo) throws Exception {
 		if (CrailConstants.DEBUG){
 			LOG.debug("RPC: setBlock, ");
 		}

 		RpcRequestMessage.SetBlockReq setBlockReq = new RpcRequestMessage.SetBlockReq(blockInfo);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(setBlockReq);
 		request.setCommand(RpcProtocol.CMD_SET_BLOCK);

 		RpcResponseMessage.VoidRes voidRes = new RpcResponseMessage.VoidRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(voidRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcVoid> nameNodeFuture = new DaRPCNameNodeFuture<RpcVoid>(future, voidRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcGetDataNode> getDataNode(DataNodeInfo dnInfo) throws Exception {
 		RpcRequestMessage.GetDataNodeReq getDataNodeReq = new RpcRequestMessage.GetDataNodeReq(dnInfo);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getDataNodeReq);
 		request.setCommand(RpcProtocol.CMD_GET_DATANODE);

 		RpcResponseMessage.GetDataNodeRes getDataNodeRes = new RpcResponseMessage.GetDataNodeRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(getDataNodeRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcGetDataNode> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetDataNode>(future, getDataNodeRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcVoid> dumpNameNode() throws Exception {


 		RpcRequestMessage.DumpNameNodeReq dumpNameNodeReq = new RpcRequestMessage.DumpNameNodeReq();
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(dumpNameNodeReq);
 		request.setCommand(RpcProtocol.CMD_DUMP_NAMENODE);

 		RpcResponseMessage.VoidRes voidRes = new RpcResponseMessage.VoidRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(voidRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcVoid> nameNodeFuture = new DaRPCNameNodeFuture<RpcVoid>(future, voidRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public DaRPCNameNodeFuture<RpcPing> pingNameNode() throws Exception {

 		RpcRequestMessage.PingNameNodeReq pingReq = new RpcRequestMessage.PingNameNodeReq();
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(pingReq);
 		request.setCommand(RpcProtocol.CMD_PING_NAMENODE);

 		RpcResponseMessage.PingNameNodeRes pingRes = new RpcResponseMessage.PingNameNodeRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(pingRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcPing> nameNodeFuture = new DaRPCNameNodeFuture<RpcPing>(future, pingRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public RpcFuture<RpcRemoveDataNode> removeDataNode(InetAddress ipaddr, int port) throws Exception {

 		RpcRequestMessage.RemoveDataNodeReq removeDataNodeReq = new RpcRequestMessage.RemoveDataNodeReq(ipaddr, port);
 		DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(removeDataNodeReq);
 		request.setCommand(RpcProtocol.CMD_REMOVE_DATANODE);

 		RpcResponseMessage.RemoveDataNodeRes removeDataNodeRes = new RpcResponseMessage.RemoveDataNodeRes();
 		DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(removeDataNodeRes);

 		DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

 		DaRPCNameNodeFuture<RpcRemoveDataNode> nameNodeFuture = new DaRPCNameNodeFuture<RpcRemoveDataNode>(future, removeDataNodeRes);

 		return nameNodeFuture;
 	}

 	@Override
 	public void close() throws Exception {
 		if (rpcEndpoint != null){
 			rpcEndpoint.close();
 			rpcEndpoint = null;
 		}
 	}

 	private DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> issueRPC(DaRPCNameNodeRequest request, DaRPCNameNodeResponse response) throws IOException{
 		try {
 			DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = stream.request(request, response, false);
 			return future;
 		} catch(IOException e){
 			LOG.info("ERROR: RPC failed, messagesSend " + rpcEndpoint.getMessagesSent() + ", messagesReceived " + rpcEndpoint.getMessagesReceived() + ", isConnected " + rpcEndpoint.isConnected() + ", qpNum " + rpcEndpoint.getQp().getQp_num());
 			throw e;
 		}
 	}

 	@Override
 	public String toString() {
 		try {
 			return rpcEndpoint.getDstAddr().toString();
 		} catch(Exception e){
 			return "Unknown";
 		}
 	}
 }
	/*
	*
	* 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.
	*/

	package org.apache.crail.namenode.rpc.darpc;

	import java.io.IOException;
	import java.net.InetAddress;

	import org.apache.crail.CrailNodeType;
	import org.apache.crail.conf.CrailConstants;
	import org.apache.crail.metadata.BlockInfo;
	import org.apache.crail.metadata.DataNodeInfo;
	import org.apache.crail.metadata.FileInfo;
	import org.apache.crail.metadata.FileName;
	import org.apache.crail.rpc.RpcConnection;
	import org.apache.crail.rpc.RpcCreateFile;
	import org.apache.crail.rpc.RpcDeleteFile;
	import org.apache.crail.rpc.RpcFuture;
	import org.apache.crail.rpc.RpcGetBlock;
	import org.apache.crail.rpc.RpcGetDataNode;
	import org.apache.crail.rpc.RpcGetFile;
	import org.apache.crail.rpc.RpcGetLocation;
	import org.apache.crail.rpc.RpcPing;
	import org.apache.crail.rpc.RpcRemoveDataNode;
	import org.apache.crail.rpc.RpcProtocol;
	import org.apache.crail.rpc.RpcRenameFile;
	import org.apache.crail.rpc.RpcRequestMessage;
	import org.apache.crail.rpc.RpcResponseMessage;
	import org.apache.crail.rpc.RpcVoid;
	import org.apache.crail.utils.CrailUtils;
	import org.slf4j.Logger;

	import com.ibm.darpc.DaRPCClientEndpoint;
	import com.ibm.darpc.DaRPCFuture;
	import com.ibm.darpc.DaRPCStream;

	public class DaRPCNameNodeConnection implements RpcConnection {
	private static final Logger LOG = CrailUtils.getLogger();

	private DaRPCClientEndpoint<DaRPCNameNodeRequest, DaRPCNameNodeResponse> rpcEndpoint;
	private DaRPCStream<DaRPCNameNodeRequest, DaRPCNameNodeResponse> stream;

	public DaRPCNameNodeConnection(DaRPCClientEndpoint<DaRPCNameNodeRequest, DaRPCNameNodeResponse> endpoint) throws IOException {
	this.rpcEndpoint = endpoint;
	this.stream = endpoint.createStream();
	}

	@Override
	public RpcFuture<RpcCreateFile> createFile(FileName filename, CrailNodeType type, int storageClass, int locationClass, boolean enumerable) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: createFile, fileType " + type + ", storageClass " + storageClass + ", locationClass " + locationClass);
	}

	RpcRequestMessage.CreateFileReq createFileReq = new RpcRequestMessage.CreateFileReq(filename, type, storageClass, locationClass, enumerable);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(createFileReq);
	request.setCommand(RpcProtocol.CMD_CREATE_FILE);

	RpcResponseMessage.CreateFileRes fileRes = new RpcResponseMessage.CreateFileRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(fileRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcCreateFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcCreateFile>(future, fileRes);

	return nameNodeFuture;
	}

	@Override
	public RpcFuture<RpcGetFile> getFile(FileName filename, boolean writeable) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: getFile, writeable " + writeable);
	}

	RpcRequestMessage.GetFileReq getFileReq = new RpcRequestMessage.GetFileReq(filename, writeable);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getFileReq);
	request.setCommand(RpcProtocol.CMD_GET_FILE);

	RpcResponseMessage.GetFileRes fileRes = new RpcResponseMessage.GetFileRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(fileRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcGetFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetFile>(future, fileRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcVoid> setFile(FileInfo fileInfo, boolean close) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: setFile, id " + fileInfo.getFd() + ", close " + close);
	}

	RpcRequestMessage.SetFileReq setFileReq = new RpcRequestMessage.SetFileReq(fileInfo, close);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(setFileReq);
	request.setCommand(RpcProtocol.CMD_SET_FILE);

	RpcResponseMessage.VoidRes voidRes = new RpcResponseMessage.VoidRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(voidRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcVoid> nameNodeFuture = new DaRPCNameNodeFuture<RpcVoid>(future, voidRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcDeleteFile> removeFile(FileName filename, boolean recursive) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: removeFile");
	}

	RpcRequestMessage.RemoveFileReq removeReq = new RpcRequestMessage.RemoveFileReq(filename, recursive);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(removeReq);
	request.setCommand(RpcProtocol.CMD_REMOVE_FILE);

	RpcResponseMessage.DeleteFileRes fileRes = new RpcResponseMessage.DeleteFileRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(fileRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcDeleteFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcDeleteFile>(future, fileRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcRenameFile> renameFile(FileName srcHash, FileName dstHash) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: renameFile");
	}

	RpcRequestMessage.RenameFileReq renameReq = new RpcRequestMessage.RenameFileReq(srcHash, dstHash);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(renameReq);
	request.setCommand(RpcProtocol.CMD_RENAME_FILE);

	RpcResponseMessage.RenameRes renameRes = new RpcResponseMessage.RenameRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(renameRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcRenameFile> nameNodeFuture = new DaRPCNameNodeFuture<RpcRenameFile>(future, renameRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcGetBlock> getBlock(long fd, long token, long position, long capacity) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: getBlock, fd " + fd + ", token " + token + ", position " + position + ", capacity " + capacity);
	}

	RpcRequestMessage.GetBlockReq getBlockReq = new RpcRequestMessage.GetBlockReq(fd, token, position, capacity);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getBlockReq);
	request.setCommand(RpcProtocol.CMD_GET_BLOCK);

	RpcResponseMessage.GetBlockRes getBlockRes = new RpcResponseMessage.GetBlockRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(getBlockRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcGetBlock> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetBlock>(future, getBlockRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcGetLocation> getLocation(FileName fileName, long position) throws IOException {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: getLocation, position " + position);
	}

	RpcRequestMessage.GetLocationReq getLocationReq = new RpcRequestMessage.GetLocationReq(fileName, position);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getLocationReq);
	request.setCommand(RpcProtocol.CMD_GET_LOCATION);

	RpcResponseMessage.GetLocationRes getLocationRes = new RpcResponseMessage.GetLocationRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(getLocationRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcGetLocation> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetLocation>(future, getLocationRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcVoid> setBlock(BlockInfo blockInfo) throws Exception {
	if (CrailConstants.DEBUG){
	LOG.debug("RPC: setBlock, ");
	}

	RpcRequestMessage.SetBlockReq setBlockReq = new RpcRequestMessage.SetBlockReq(blockInfo);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(setBlockReq);
	request.setCommand(RpcProtocol.CMD_SET_BLOCK);

	RpcResponseMessage.VoidRes voidRes = new RpcResponseMessage.VoidRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(voidRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcVoid> nameNodeFuture = new DaRPCNameNodeFuture<RpcVoid>(future, voidRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcGetDataNode> getDataNode(DataNodeInfo dnInfo) throws Exception {
	RpcRequestMessage.GetDataNodeReq getDataNodeReq = new RpcRequestMessage.GetDataNodeReq(dnInfo);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(getDataNodeReq);
	request.setCommand(RpcProtocol.CMD_GET_DATANODE);

	RpcResponseMessage.GetDataNodeRes getDataNodeRes = new RpcResponseMessage.GetDataNodeRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(getDataNodeRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcGetDataNode> nameNodeFuture = new DaRPCNameNodeFuture<RpcGetDataNode>(future, getDataNodeRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcVoid> dumpNameNode() throws Exception {


	RpcRequestMessage.DumpNameNodeReq dumpNameNodeReq = new RpcRequestMessage.DumpNameNodeReq();
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(dumpNameNodeReq);
	request.setCommand(RpcProtocol.CMD_DUMP_NAMENODE);

	RpcResponseMessage.VoidRes voidRes = new RpcResponseMessage.VoidRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(voidRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcVoid> nameNodeFuture = new DaRPCNameNodeFuture<RpcVoid>(future, voidRes);

	return nameNodeFuture;
	}

	@Override
	public DaRPCNameNodeFuture<RpcPing> pingNameNode() throws Exception {

	RpcRequestMessage.PingNameNodeReq pingReq = new RpcRequestMessage.PingNameNodeReq();
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(pingReq);
	request.setCommand(RpcProtocol.CMD_PING_NAMENODE);

	RpcResponseMessage.PingNameNodeRes pingRes = new RpcResponseMessage.PingNameNodeRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(pingRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcPing> nameNodeFuture = new DaRPCNameNodeFuture<RpcPing>(future, pingRes);

	return nameNodeFuture;
	}

	@Override
	public RpcFuture<RpcRemoveDataNode> removeDataNode(InetAddress ipaddr, int port) throws Exception {

	RpcRequestMessage.RemoveDataNodeReq removeDataNodeReq = new RpcRequestMessage.RemoveDataNodeReq(ipaddr, port);
	DaRPCNameNodeRequest request = new DaRPCNameNodeRequest(removeDataNodeReq);
	request.setCommand(RpcProtocol.CMD_REMOVE_DATANODE);

	RpcResponseMessage.RemoveDataNodeRes removeDataNodeRes = new RpcResponseMessage.RemoveDataNodeRes();
	DaRPCNameNodeResponse response = new DaRPCNameNodeResponse(removeDataNodeRes);

	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = issueRPC(request, response);

	DaRPCNameNodeFuture<RpcRemoveDataNode> nameNodeFuture = new DaRPCNameNodeFuture<RpcRemoveDataNode>(future, removeDataNodeRes);

	return nameNodeFuture;
	}

	@Override
	public void close() throws Exception {
	if (rpcEndpoint != null){
	rpcEndpoint.close();
	rpcEndpoint = null;
	}
	}

	private DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> issueRPC(DaRPCNameNodeRequest request, DaRPCNameNodeResponse response) throws IOException{
	try {
	DaRPCFuture<DaRPCNameNodeRequest, DaRPCNameNodeResponse> future = stream.request(request, response, false);
	return future;
	} catch(IOException e){
	LOG.info("ERROR: RPC failed, messagesSend " + rpcEndpoint.getMessagesSent() + ", messagesReceived " + rpcEndpoint.getMessagesReceived() + ", isConnected " + rpcEndpoint.isConnected() + ", qpNum " + rpcEndpoint.getQp().getQp_num());
	throw e;
	}
	}

	@Override
	public String toString() {
	try {
	return rpcEndpoint.getDstAddr().toString();
	} catch(Exception e){
	return "Unknown";
	}
	}
	}