wyl/fastdfs-client-java
1
0
Fork 0
mirror of https://github.com/happyfish100/fastdfs-client-java.git synced 2026-09-17 09:20:46 +00:00
Code Issues Projects Releases Wiki Activity

connection broken message with address and port

Browse Source
This commit is contained in:
YuQing
2025-11-28 09:50:09 +08:00
parent cc95bc0244
commit 91a5cc2e36
5 changed files with 107 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HISTORY
+1 -1
View File
@@ -1,5 +1,5 @@
Version 1.38 2025-11-27
Version 1.38 2025-11-28
* bugfixed: loadStorageServersFromTracker correctly with rw option
* throw IOException with msg "connection broken" when in.read() < 0
src/main/java/org/csource/fastdfs/ProtoCommon.java
+69 -54
View File
@@ -15,6 +15,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.Socket;
import java.net.InetSocketAddress;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
@@ -174,39 +175,48 @@ public class ProtoCommon {
* @param expect_body_len expect response package body length
* @return RecvHeaderInfo: errno and pkg body length
*/
public static RecvHeaderInfo recvHeader(InputStream in, byte expect_cmd, long expect_body_len) throws IOException {
byte[] header;
int bytes;
long pkg_len;
public static RecvHeaderInfo recvHeader(InputStream in, InetSocketAddress socketAddr,
byte expect_cmd, long expect_body_len) throws IOException
{
byte[] header;
int bytes;
long pkg_len;
header = new byte[FDFS_PROTO_PKG_LEN_SIZE + 2];
bytes = in.read(header);
if (bytes != header.length) {
if (bytes < 0) {
throw new IOException("connection broken");
} else {
throw new IOException("recv package size " + bytes + " != " + header.length);
}
}
header = new byte[FDFS_PROTO_PKG_LEN_SIZE + 2];
bytes = in.read(header);
if (bytes != header.length) {
if (bytes < 0) {
throw new IOException("server " + socketAddr.getAddress().getHostAddress()
+ ":" + socketAddr.getPort() + ", connection broken");
} else {
throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":"
+ socketAddr.getPort() + ", recv package size " + bytes + " != " + header.length);
}
}
if (header[PROTO_HEADER_CMD_INDEX] != expect_cmd) {
throw new IOException("recv cmd: " + header[PROTO_HEADER_CMD_INDEX] + " is not correct, expect cmd: " + expect_cmd);
}
if (header[PROTO_HEADER_CMD_INDEX] != expect_cmd) {
throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":"
+ socketAddr.getPort() + ", recv cmd: " + header[PROTO_HEADER_CMD_INDEX]
+ " is not correct, expect cmd: " + expect_cmd);
}
if (header[PROTO_HEADER_STATUS_INDEX] != 0) {
return new RecvHeaderInfo(header[PROTO_HEADER_STATUS_INDEX], 0);
}
if (header[PROTO_HEADER_STATUS_INDEX] != 0) {
return new RecvHeaderInfo(header[PROTO_HEADER_STATUS_INDEX], 0);
}
pkg_len = ProtoCommon.buff2long(header, 0);
if (pkg_len < 0) {
throw new IOException("recv body length: " + pkg_len + " < 0!");
}
pkg_len = ProtoCommon.buff2long(header, 0);
if (pkg_len < 0) {
throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":"
+ socketAddr.getPort() + ", recv body length: " + pkg_len + " < 0!");
}
if (expect_body_len >= 0 && pkg_len != expect_body_len) {
throw new IOException("recv body length: " + pkg_len + " is not correct, expect length: " + expect_body_len);
}
if (expect_body_len >= 0 && pkg_len != expect_body_len) {
throw new IOException("server " + socketAddr.getAddress().getHostAddress() + ":"
+ socketAddr.getPort() + ", recv body length: " + pkg_len
+ " is not correct, expect length: " + expect_body_len);
}
return new RecvHeaderInfo((byte) 0, pkg_len);
return new RecvHeaderInfo((byte)0, pkg_len);
}
/**
@@ -217,36 +227,40 @@ public class ProtoCommon {
* @param expect_body_len expect response package body length
* @return RecvPackageInfo: errno and reponse body(byte buff)
*/
public static RecvPackageInfo recvPackage(InputStream in, byte expect_cmd, long expect_body_len) throws IOException {
RecvHeaderInfo header = recvHeader(in, expect_cmd, expect_body_len);
if (header.errno != 0) {
return new RecvPackageInfo(header.errno, null);
}
byte[] body = new byte[(int) header.body_len];
int totalBytes = 0;
int remainBytes = (int) header.body_len;
int bytes = 0;
while (totalBytes < header.body_len) {
if ((bytes = in.read(body, totalBytes, remainBytes)) < 0) {
break;
public static RecvPackageInfo recvPackage(InputStream in, InetSocketAddress socketAddr,
byte expect_cmd, long expect_body_len) throws IOException
{
RecvHeaderInfo header = recvHeader(in, socketAddr, expect_cmd, expect_body_len);
if (header.errno != 0) {
return new RecvPackageInfo(header.errno, null);
}
totalBytes += bytes;
remainBytes -= bytes;
}
byte[] body = new byte[(int) header.body_len];
int totalBytes = 0;
int remainBytes = (int) header.body_len;
int bytes = 0;
if (totalBytes != header.body_len) {
if (totalBytes == 0) {
throw new IOException("connection broken");
} else {
throw new IOException("connection broken, recv length: " + totalBytes
+ ", expect length: " + header.body_len);
}
}
while (totalBytes < header.body_len) {
if ((bytes = in.read(body, totalBytes, remainBytes)) < 0) {
break;
}
return new RecvPackageInfo((byte) 0, body);
totalBytes += bytes;
remainBytes -= bytes;
}
if (totalBytes != header.body_len) {
String msg = "server " + socketAddr.getAddress().getHostAddress()
+ ":" + socketAddr.getPort() + " connection broken";
if (totalBytes == 0) {
throw new IOException(msg);
} else {
throw new IOException(msg + ", recv length: " + totalBytes
+ ", expect length: " + header.body_len);
}
}
return new RecvPackageInfo((byte)0, body);
}
/**
@@ -329,7 +343,8 @@ public class ProtoCommon {
header = packHeader(FDFS_PROTO_CMD_ACTIVE_TEST, 0, (byte) 0);
sock.getOutputStream().write(header);
RecvHeaderInfo headerInfo = recvHeader(sock.getInputStream(), TRACKER_PROTO_CMD_RESP, 0);
InetSocketAddress socketAddr = new InetSocketAddress(sock.getInetAddress(), sock.getPort());
RecvHeaderInfo headerInfo = recvHeader(sock.getInputStream(), socketAddr, TRACKER_PROTO_CMD_RESP, 0);
return headerInfo.errno == 0 ? true : false;
}
src/main/java/org/csource/fastdfs/StorageClient.java
+12 -12
View File
@@ -656,7 +656,7 @@ public class StorageClient {
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -801,7 +801,7 @@ public class StorageClient {
}
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -907,7 +907,7 @@ public class StorageClient {
}
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return this.errno;
@@ -1004,7 +1004,7 @@ public class StorageClient {
}
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return this.errno;
@@ -1040,7 +1040,7 @@ public class StorageClient {
try {
this.send_package(ProtoCommon.STORAGE_PROTO_CMD_DELETE_FILE, group_name, remote_filename, connection);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
this.errno = pkgInfo.errno;
return pkgInfo.errno;
@@ -1122,7 +1122,7 @@ public class StorageClient {
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
this.errno = pkgInfo.errno;
return pkgInfo.errno;
} catch (IOException ex) {
@@ -1171,7 +1171,7 @@ public class StorageClient {
this.send_download_package(group_name, remote_filename, file_offset, download_bytes, connection);
pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
@@ -1232,7 +1232,7 @@ public class StorageClient {
this.send_download_package(group_name, remote_filename, file_offset, download_bytes, connection);
InputStream in = connection.getInputStream();
header = ProtoCommon.recvHeader(in, ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
header = ProtoCommon.recvHeader(in, connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = header.errno;
if (header.errno != 0) {
return header.errno;
@@ -1320,7 +1320,7 @@ public class StorageClient {
this.send_download_package(group_name, remote_filename, file_offset, download_bytes, connection);
InputStream in = connection.getInputStream();
header = ProtoCommon.recvHeader(in, ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
header = ProtoCommon.recvHeader(in, connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = header.errno;
if (header.errno != 0) {
return header.errno;
@@ -1376,7 +1376,7 @@ public class StorageClient {
this.send_package(ProtoCommon.STORAGE_PROTO_CMD_GET_METADATA, group_name, remote_filename, connection);
pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
@@ -1467,7 +1467,7 @@ public class StorageClient {
}
pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, 0);
this.errno = pkgInfo.errno;
return pkgInfo.errno;
@@ -1580,7 +1580,7 @@ public class StorageClient {
out.write(wholePkg);
pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.STORAGE_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
src/main/java/org/csource/fastdfs/TrackerClient.java
+12 -13
View File
@@ -118,19 +118,19 @@ public class TrackerClient {
}
return trackerServer.getConnection();
} catch (IOException e) {
System.err.println("fail over trackerServer get connection error, failOverCount:" + failOverCount + "," + e.getMessage());
System.err.println("fail over trackerServer get connection error, "
+ "failOverCount: " + failOverCount + ", " + e.getMessage());
if (failOverCount == length - 1) {
throw e;
}
} catch (MyException e) {
System.err.println("fail over trackerServer get connection error, failOverCount:" + failOverCount + ", " + e.getMessage());
System.err.println("fail over trackerServer get connection error, "
+ "failOverCount: " + failOverCount + ", " + e.getMessage());
if (failOverCount == length - 1) {
throw e;
}
}
}
return null;
}
@@ -183,7 +183,7 @@ public class TrackerClient {
}
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -275,7 +275,7 @@ public class TrackerClient {
}
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -439,7 +439,7 @@ public class TrackerClient {
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -560,7 +560,7 @@ public class TrackerClient {
out.write(header);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -668,7 +668,7 @@ public class TrackerClient {
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
@@ -749,7 +749,7 @@ public class TrackerClient {
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, 0);
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, 0);
this.errno = pkgInfo.errno;
return pkgInfo.errno == 0;
} catch (IOException e) {
@@ -893,9 +893,8 @@ public class TrackerClient {
System.arraycopy(bs, 0, wholePkg, header.length, bs.length);
out.write(wholePkg);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(
connection.getInputStream(),
ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
ProtoCommon.RecvPackageInfo pkgInfo = ProtoCommon.recvPackage(connection.getInputStream(),
connection.getInetSocketAddress(), ProtoCommon.TRACKER_PROTO_CMD_RESP, -1);
this.errno = pkgInfo.errno;
if (pkgInfo.errno != 0) {
return null;
src/main/java/org/csource/fastdfs/pool/ConnectionManager.java
+13 -5
View File
@@ -53,7 +53,9 @@ public class ConnectionManager {
if (freeCount.get() > 0) {
freeCount.decrementAndGet();
connection = freeConnections.poll();
if (!connection.isAvaliable() || (System.currentTimeMillis() - connection.getLastAccessTime()) > ClientGlobal.g_connection_pool_max_idle_time) {
if (!connection.isAvaliable() || (System.currentTimeMillis() - connection.getLastAccessTime()) >
ClientGlobal.g_connection_pool_max_idle_time)
{
closeConnection(connection);
continue;
}
@@ -62,7 +64,8 @@ public class ConnectionManager {
try {
isActive = connection.activeTest();
} catch (IOException e) {
System.err.println("send to server[" + inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort() + "] active test error, emsg: " + e.getMessage());
System.err.println("send to server " + inetSocketAddress.getAddress().getHostAddress()
+ ":" + inetSocketAddress.getPort() + " active test error, emsg: " + e.getMessage());
isActive = false;
}
if (!isActive) {
@@ -72,7 +75,9 @@ public class ConnectionManager {
connection.setNeedActiveTest(false);
}
}
} else if (ClientGlobal.g_connection_pool_max_count_per_entry == 0 || totalCount.get() < ClientGlobal.g_connection_pool_max_count_per_entry) {
} else if (ClientGlobal.g_connection_pool_max_count_per_entry == 0 || totalCount.get() <
ClientGlobal.g_connection_pool_max_count_per_entry)
{
connection = ConnectionFactory.create(this.inetSocketAddress);
totalCount.incrementAndGet();
} else {
@@ -81,10 +86,13 @@ public class ConnectionManager {
//wait single success
continue;
}
throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort() + " fail, wait_time > " + ClientGlobal.g_connection_pool_max_wait_time_in_ms + "ms");
throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress()
+ ":" + inetSocketAddress.getPort() + " fail, wait_time > "
+ ClientGlobal.g_connection_pool_max_wait_time_in_ms + "ms");
} catch (InterruptedException e) {
e.printStackTrace();
throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress() + ":" + inetSocketAddress.getPort() + " fail, emsg: " + e.getMessage());
throw new MyException("connect to server " + inetSocketAddress.getAddress().getHostAddress()
+ ":" + inetSocketAddress.getPort() + " fail, emsg: " + e.getMessage());
}
}
return connection;