diff --git a/broker/migrate/h2/V53__session_add_ip_column.sql b/broker/migrate/h2/V53__session_add_ip_column.sql new file mode 100755 index 00000000..e194cbb3 --- /dev/null +++ b/broker/migrate/h2/V53__session_add_ip_column.sql @@ -0,0 +1 @@ +ALTER TABLE `t_user_session` ADD COLUMN `_ip` varchar(40) DEFAULT ''; diff --git a/broker/migrate/mysql/V61__session_add_ip_column.sql b/broker/migrate/mysql/V61__session_add_ip_column.sql new file mode 100755 index 00000000..e194cbb3 --- /dev/null +++ b/broker/migrate/mysql/V61__session_add_ip_column.sql @@ -0,0 +1 @@ +ALTER TABLE `t_user_session` ADD COLUMN `_ip` varchar(40) DEFAULT ''; diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java index 056481cf..acd8fb05 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java @@ -11,6 +11,7 @@ package com.xiaoleilu.loServer.action; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import com.google.protobuf.InvalidProtocolBufferException; +import com.hazelcast.util.StringUtil; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; @@ -28,7 +29,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import cn.wildfirechat.common.ErrorCode; +import java.net.InetSocketAddress; import java.util.Base64; +import java.util.Map; import java.util.concurrent.Executor; @Route("/route") @@ -111,6 +114,12 @@ public class RouteAction extends Action { WFCMessage.IMHttpWrapper wrapper = WFCMessage.IMHttpWrapper.parseFrom(bytes); String token = wrapper.getToken(); String userId = Tokenor.getUserId(token.getBytes()); + + String remote = ((InetSocketAddress) ctx.channel().remoteAddress()) + .getAddress().getHostAddress(); + String ip = computeRealIp(request.getHeaders(), remote); + + LOG.info("RouteAction token={}, userId={}", token, userId); if (userId == null) { sendResponse(response, ErrorCode.ERROR_CODE_TOKEN_ERROR, null); @@ -119,6 +128,8 @@ public class RouteAction extends Action { @Override public void onSuccess(byte[] result) { sendResponse(response, null, result); + //save ip after success response. + sessionsStore.updateSessionIp(userId, wrapper.getClientId(), ip); } @Override @@ -147,6 +158,20 @@ public class RouteAction extends Action { return true; } + private String computeRealIp(Map h, String remote) { + String xff = h.get("X-Forwarded-For"); + if (xff != null && !xff.isEmpty() && !"unknown".equalsIgnoreCase(xff)) { + int idx = xff.indexOf(','); + String first = (idx > 0 ? xff.substring(0, idx) : xff).trim(); + return first; + } + String xri = h.get("X-Real-IP"); + if (xri != null && !xri.isEmpty() && !"unknown".equalsIgnoreCase(xri)) { + return xri; + } + return remote; + } + private void sendResponse(Response response, ErrorCode errorCode, byte[] contents) { response.setStatus(HttpResponseStatus.OK); if (contents == null) { diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 01e85728..3ec27540 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -1832,6 +1832,31 @@ public class DatabaseStore { }); } + void updateSessionIp(String uid, String cid, String ip) { + mScheduler.execute(()->{ + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + String sql = "update t_user_session set _ip = ? where `_uid` = ? and `_cid` = ?"; + statement = connection.prepareStatement(sql); + int index = 1; + statement.setString(index++, ip); + statement.setString(index++, uid); + statement.setString(index++, cid); + + int c = statement.executeUpdate(); + LOG.info("Update rows {}", c); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement); + } + }); + } + void clearMultiEndpoint(String uid, String clientId, int platform) { LOG.info("clearMultiEndpoint {}, {}", uid, clientId); Connection connection = null; diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 3b44b7d9..39dad7b8 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -61,6 +61,7 @@ public class MemorySessionStore implements ISessionsStore { private int deleted; private boolean pullHistoryMsg; + private String ip; public int getDeleted() { return deleted; @@ -159,6 +160,14 @@ public class MemorySessionStore implements ISessionsStore { } } + public String getIp() { + return ip; + } + + public void setIp(String ip) { + this.ip = ip; + } + private int pushType; private int platform; @@ -544,6 +553,15 @@ public class MemorySessionStore implements ISessionsStore { return session.clientSession; } + @Override + public void updateSessionIp(String username, String clientID, String ip) { + Session session = getSession(clientID); + if(session != null && session.getDeleted() == 0) { + session.setIp(ip); + databaseStore.updateSessionIp(username, clientID, ip); + } + } + @Override public Session sessionForClientAndUser(String username, String clientID) { Session session = sessions.get(clientID); diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 4e6cdf7b..7749ef5f 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -43,7 +43,7 @@ public interface ISessionsStore { ErrorCode loadActiveSession(String username, String clientID); ClientSession updateExistSession(String username, String clientID, WFCMessage.RouteRequest endpoint, boolean cleanSession); - + void updateSessionIp(String username, String clientID, String ip); Session getSession(String clientID); void cleanDuplatedToken(String cid, int pushType, String token, boolean isVoip, String packageName); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index f5491aea..939f0ce3 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -284,7 +284,7 @@ public class ProtocolProcessor { sessionstatus = 0; } - userOnlineStatus.sessions.add(new OutputCheckUserOnline.Session(userId, session.getClientID(), session.getPlatform(), sessionstatus, session.getLastActiveTime(), session.getAppName())); + userOnlineStatus.sessions.add(new OutputCheckUserOnline.Session(userId, session.getClientID(), session.getPlatform(), sessionstatus, session.getLastActiveTime(), session.getAppName(), session.getIp())); } HttpUtils.httpJsonPost(mUserOnlineStatusCallback, GsonUtil.gson.toJson(userOnlineStatus), HttpUtils.HttpPostType.POST_TYPE_User_Online_Event_Callback); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 63b4d9c1..59fd7986 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -141,7 +141,7 @@ public class Qos1PublishHandler extends QosPublishHandler { status = 0; } - out.addSession(userId, session.getClientID(), session.getPlatform(), status, session.getLastActiveTime(), session.getAppName()); + out.addSession(userId, session.getClientID(), session.getPlatform(), status, session.getLastActiveTime(), session.getAppName(), session.getIp()); } callback.onRouteHandled(GsonUtil.gson.toJson(out).getBytes()); diff --git a/common/src/main/java/cn/wildfirechat/pojos/OutputCheckUserOnline.java b/common/src/main/java/cn/wildfirechat/pojos/OutputCheckUserOnline.java index 735f7a5d..bdb1d89b 100644 --- a/common/src/main/java/cn/wildfirechat/pojos/OutputCheckUserOnline.java +++ b/common/src/main/java/cn/wildfirechat/pojos/OutputCheckUserOnline.java @@ -20,22 +20,24 @@ public class OutputCheckUserOnline { public int status; //0 online, 1 have session offline public long lastSeen; public String packageName; + public String ip; public Session() { } - public Session(String clientId, String userId, int platform, int status, long lastSeen, String packageName) { + public Session(String clientId, String userId, int platform, int status, long lastSeen, String packageName, String ip) { this.clientId = clientId; this.userId = userId; this.platform = platform; this.status = status; this.lastSeen = lastSeen; this.packageName = packageName; + this.ip = ip; } } - public void addSession(String userId, String clientId, int platform, int status, long lastSeen, String packageName) { - Session session = new Session(clientId, userId, platform, status, lastSeen, packageName); + public void addSession(String userId, String clientId, int platform, int status, long lastSeen, String packageName, String ip) { + Session session = new Session(clientId, userId, platform, status, lastSeen, packageName, ip); sessions.add(session); }