记录客户端来源IP地址

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;

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);

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);

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);

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());