socketchannelhelper.java

用java写的ftp服务器程序

		}
		if (totalDataLength > 0 && (totalReaded != maxLength)) {
			throw new IOException("read data from client failed,readed cnt=:" + totalReaded + ",want cnt="
					+ maxLength + " is not equal");
		}
		buffer.clear();
		ByteBuffer retBuffer = ByteBuffer.allocate(totalReaded);
		for (Iterator itor = listData.iterator(); itor.hasNext();) {
			byte[] data = (byte[]) itor.next();
			retBuffer.put(data);
		}
		listData.clear();
		retBuffer.position(0);
		retBuffer.limit(totalReaded);
		return retBuffer;
	}

//	/**
//	 * 
//	 * @param socketChannel
//	 * @param dataLength
//	 *          if <=0 will read all data back max data length = 1MB(1048576)
//	 * @param timeOut
//	 *          time in ms
//	 * @return ByteBuffer's position=0 ByteBuffer's limit=data that want readed
//	 */
//	public final ByteBuffer readFixPackage(final ClientSocketConnectedIn socketChannel,
//			final int totalDataLength, final long timeOut, ByteBuffer buffer) throws IOException,
//			InterruptedException
//	{
//		final int maxLength = ((totalDataLength <= 0) ? 1048576 : totalDataLength);
//		final int bufferLength = (maxLength > 16384) ? 16384 : maxLength;
//		long currTime = System.currentTimeMillis();
//		if (buffer == null || buffer.capacity() < bufferLength)
//			buffer = ByteBuffer.allocate(bufferLength);
//		else
//			buffer.clear();
//		// contain byte[]
//		List listData = new ArrayList(maxLength / bufferLength + 1);
//		int currReaded = 0;
//		int byteRemain = maxLength, totalReaded = 0;
//		buffer.limit(bufferLength);
//		boolean isTimeOut = false;
//		while (true) {
//			int readed = socketChannel.getChannel().read(buffer);
//			if (readed < 0) {
//				if (currReaded > 0) {
//					buffer.position(0);
//					buffer.limit(currReaded);
//					byte[] datas = new byte[currReaded];
//					buffer.get(datas);
//					listData.add(datas);
//					buffer.clear();
//				}
//				break;
//			}
//			else if (readed > 0) {
//				socketChannel.markActive();
//				currReaded += readed;
//				totalReaded += readed;
//				if (currReaded == bufferLength) {
//					// one buffer readed
//					buffer.position(0);
//					buffer.limit(bufferLength);
//					byte[] datas = new byte[bufferLength];
//					buffer.get(datas);
//					listData.add(datas);
//					buffer.clear();
//					byteRemain -= bufferLength;
//					if (byteRemain == 0)
//						break;
//					if (byteRemain >= bufferLength)
//						buffer.limit(bufferLength);
//					else
//						buffer.limit(byteRemain);
//					currReaded = 0;
//				}
//				currTime = System.currentTimeMillis();
//			}
//			else {
//				if (totalReaded >= maxLength) {
//					if (currReaded > 0) {
//						buffer.position(0);
//						buffer.limit(currReaded);
//						byte[] datas = new byte[currReaded];
//						buffer.get(datas);
//						listData.add(datas);
//						buffer.clear();
//					}
//					break;
//				}
//				Thread.sleep(3);
//				if (System.currentTimeMillis() - currTime > timeOut) {
//					isTimeOut = true;
//					break;
//				}
//			}
//		}
//		if (isTimeOut) {
//			String strErr = "read data from destination is time out,remote:" + socketChannel;
//			throw new IOException(strErr);
//		}
//		if (totalDataLength > 0 && (totalReaded != maxLength)) {
//			throw new IOException("read data from client failed,readed cnt=:" + totalReaded + ",want cnt="
//					+ maxLength + " is not equal");
//		}
//		buffer.clear();
//		ByteBuffer retBuffer = ByteBuffer.allocate(totalReaded);
//		for (Iterator itor = listData.iterator(); itor.hasNext();) {
//			byte[] data = (byte[]) itor.next();
//			retBuffer.put(data);
//		}
//		listData.clear();
//		retBuffer.position(0);
//		retBuffer.limit(totalReaded);
//		return retBuffer;
//	}

	/**
	 * 获得长度定义字节 ，长度的格式是 前补0的16进制，举例: 100 = 0x64, buffLen = 6 在byte[]中存放的结果就是字符串:'000064' 对应的字节(UTF-8)编码
	 * 
	 * @param pkgLen
	 * @param formatBuffLen
	 * @return
	 */
	public final byte[] getLengthFormatByte(int pkgLen, int formatBuffLen) throws UnsupportedEncodingException {
		byte[] data = new byte[formatBuffLen];
		Arrays.fill(data, (byte) '0');
		byte[] lenByte = (Integer.toString(pkgLen, 16)).getBytes("UTF-8");
		System.arraycopy(lenByte, 0, data, formatBuffLen - lenByte.length, lenByte.length);
		return data;
	}

	/**
	 * 从ByteBuffer中拆析出长度，长度的格式是 前补0的16进制，举例: 100 = 0x64, buffLen = 6 在ByteBuffer中存放的结果就是字符串'000064'对应的UTF-8字节编码
	 * 
	 * @param buffer
	 * @param buffLen
	 * @return
	 * @throws NumberFormatException
	 */
	public final int unParseFormatLengthByte(ByteBuffer buffer, int buffLen) throws NumberFormatException {
		Charset charset = Charset.forName("UTF-8");
		CharsetDecoder decoder = charset.newDecoder();
		CharBuffer charBuff = CharBuffer.allocate(buffLen);
		decoder.decode(buffer, charBuff, true);
		charBuff.flip();
		return Integer.parseInt(charBuff.toString(), 16);
	}

	/**
	 * 替换在ByteBuffer中的数据，dest中的偏移值=@param position 替换的数据在@param data,@param offset,@param len中定义 函数处理以后，
	 * 当替换的数据超出原先的limit时，position将自动扩充 否则，position＝原先的position 当时limit可能会发生变化（当替换的数据超出原先的buffer的长度的时候)
	 * 
	 * @param dest
	 * @param position
	 * @param data
	 * @param offset
	 * @param len
	 * @return
	 */
	public final ByteBuffer replaceBuffer(ByteBuffer dest, int position, byte[] data, int offset, int len) {
		if (dest == null) {
			return ByteBuffer.wrap(data, offset, len);
		}
		int prePosition = dest.position();
		if ((position + len) >= dest.limit()) {
			// 需要重新分配
			ByteBuffer newBuffer = ByteBuffer.allocate(dest.limit() + len);
			newBuffer.put(dest);
			newBuffer.position(position);
			newBuffer.put(data, offset, len);
			newBuffer.position(position + len);
			return newBuffer;
		}
		else {
			dest.position(position);
			dest.put(data, offset, len);
			dest.position(prePosition);
			return dest;
		}
	}

	/**
	 * put data to dest buffer,if dest is over flow, will new buffer and return it back
	 * 
	 * @param dest
	 * @param data
	 * @return
	 */
	public final ByteBuffer appendBuffer(ByteBuffer dest, byte[] data, int offset, int length) {
		if (dest == null) {
			dest = ByteBuffer.wrap(data, offset, length);
			return dest;
		}
		if (dest.remaining() < length) {
			ByteBuffer newBuff = ByteBuffer.allocate(dest.limit() + length + 64);
			dest.flip();
			newBuff.put(dest);
			newBuff.put(data, offset, length);
			return newBuff;
		}
		else {
			return dest.put(data, offset, length);
		}
	}

	/**
	 * put data to dest buffer,if dest is over flow, will new buffer and return it back
	 * 
	 * @param dest
	 * @param data
	 * @return
	 */
	public final ByteBuffer appendBuffer(ByteBuffer dest, byte data) {
		if (dest == null) {
			dest = ByteBuffer.allocate(64);
			dest.put(data);
			return dest;
		}
		if (dest.remaining() < 1) {
			ByteBuffer newBuff = ByteBuffer.allocate(dest.limit() + 64);
			dest.flip();
			newBuff.put(dest);
			newBuff.put(data);
			return newBuff;
		}
		else {
			return dest.put(data);
		}
	}

	public SocketChannel open(String host, int port, boolean isBlockingConnect) throws IOException {
		SocketChannel channel = SocketChannel.open();
		channel.configureBlocking(isBlockingConnect);
		channel.connect(new InetSocketAddress(InetAddress.getByName(host), port));
		return channel;
	}

	public void close(SocketChannel channel) {
		if (channel != null) {
			try {
				channel.close();
			}
			catch (Exception e) {
			}
		}
	}
	
	public boolean isSocketAlive(SocketChannel channel)
	{
		if(channel == null)
			return false;
		Socket socket = channel.socket();
		if(socket == null)
			return false;
		boolean isOpen = 
			socket.isConnected() && (!socket.isInputShutdown() || !socket.isOutputShutdown());
		return isOpen;		
	}
}