diff --git a/broker/migrate/h2/V16__add_friend_blocked.sql b/broker/migrate/h2/V16__add_friend_blocked.sql new file mode 100755 index 00000000..f352633f --- /dev/null +++ b/broker/migrate/h2/V16__add_friend_blocked.sql @@ -0,0 +1,2 @@ +alter table t_friend add column `_blacked` tinyint DEFAULT 0; +update t_friend set `_blacked` = 1 where `_state` = 2; diff --git a/broker/migrate/mysql/V21__add_friend_blocked.sql b/broker/migrate/mysql/V21__add_friend_blocked.sql new file mode 100755 index 00000000..da5911e4 --- /dev/null +++ b/broker/migrate/mysql/V21__add_friend_blocked.sql @@ -0,0 +1,2 @@ +alter table t_friend add column `_blacked` tinyint DEFAULT 0 COMMENT "0, normal; 1, blacked"; +update t_friend set `_blacked` = 1 where `_state` = 2; diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationGetAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationGetAction.java index c771bb0d..7c240d21 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationGetAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationGetAction.java @@ -36,7 +36,7 @@ public class FriendRelationGetAction extends AdminAction { public boolean action(Request request, Response response) { if (request.getNettyRequest() instanceof FullHttpRequest) { InputGetFriendList inputGetFriendList = getRequestBody(request.getNettyRequest(), InputGetFriendList.class); - List dataList = messagesStore.getFriendList(inputGetFriendList.getUserId(), 0); + List dataList = messagesStore.getFriendList(inputGetFriendList.getUserId(), null, 0); List list = new ArrayList<>(); for (FriendData data : dataList) { if (data.getState() == inputGetFriendList.getStatus()) { diff --git a/broker/src/main/java/com/xiaoleilu/loServer/model/FriendData.java b/broker/src/main/java/com/xiaoleilu/loServer/model/FriendData.java index c684572c..c2b60ee6 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/model/FriendData.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/model/FriendData.java @@ -15,14 +15,16 @@ public class FriendData implements Serializable { private String friendUid; private String alias; private int state; + private int blacked; private long timestamp; - public FriendData(String userId, String friendUid, String alias, int state, long timestamp) { + public FriendData(String userId, String friendUid, String alias, int state, int blacked, long timestamp) { this.userId = userId; this.friendUid = friendUid; this.alias = alias; this.state = state; + this.blacked = blacked; this.timestamp = timestamp; } @@ -61,6 +63,14 @@ public class FriendData implements Serializable { this.state = state; } + public int getBlacked() { + return blacked; + } + + public void setBlacked(int blacked) { + this.blacked = blacked; + } + public long getTimestamp() { return timestamp; } diff --git a/broker/src/main/java/io/moquette/imhandler/FriendPullHandler.java b/broker/src/main/java/io/moquette/imhandler/FriendPullHandler.java index 0927eb00..1a312096 100644 --- a/broker/src/main/java/io/moquette/imhandler/FriendPullHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/FriendPullHandler.java @@ -22,7 +22,7 @@ import java.util.List; public class FriendPullHandler extends IMHandler { @Override public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.Version request, Qos1PublishHandler.IMCallback callback) { - List friendDatas = m_messagesStore.getFriendList(fromUser, request.getVersion()); + List friendDatas = m_messagesStore.getFriendList(fromUser, clientID, request.getVersion()); WFCMessage.GetFriendsResult.Builder builder = WFCMessage.GetFriendsResult.newBuilder(); for (FriendData data : friendDatas ) { diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 94b2c47a..cf5cb2af 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -350,7 +350,7 @@ public class DatabaseStore { try { connection = DBUtil.getConnection(); - String sql = "select `_uid`, `_friend_uid`, `_alias`, `_state`, `_dt` from t_friend"; + String sql = "select `_uid`, `_friend_uid`, `_alias`, `_state`, `_blacked`, `_dt` from t_friend"; statement = connection.prepareStatement(sql); int index; @@ -375,6 +375,9 @@ public class DatabaseStore { int intvalue = rs.getInt(index++); builder.setState(intvalue); + intvalue = rs.getInt(index++); + builder.setBlacked(intvalue); + long longvalue = rs.getLong(index++); builder.setTimestamp(longvalue); @@ -2139,7 +2142,7 @@ public class DatabaseStore { ResultSet rs = null; try { connection = DBUtil.getConnection(); - String sql = "select `_friend_uid`, `_alias`, `_state`, `_dt` from t_friend where `_uid` = ?"; + String sql = "select `_friend_uid`, `_alias`, `_state`, `_blacked`, `_dt` from t_friend where `_uid` = ?"; statement = connection.prepareStatement(sql); @@ -2153,9 +2156,10 @@ public class DatabaseStore { String uid = rs.getString(1); String alias = rs.getString(2); int state = rs.getInt(3); - long timestamp = rs.getLong(4); + int blacked = rs.getInt(4); + long timestamp = rs.getLong(5); - FriendData data = new FriendData(userId, uid, alias, state, timestamp); + FriendData data = new FriendData(userId, uid, alias, state, blacked, timestamp); out.add(data); } return out; @@ -2305,10 +2309,11 @@ public class DatabaseStore { PreparedStatement statement = null; try { connection = DBUtil.getConnection(); - String sql = "insert into t_friend (`_uid`, `_friend_uid`, `_alias`, `_state`, `_dt`) values(?, ?, ?, ?, ?)" + + String sql = "insert into t_friend (`_uid`, `_friend_uid`, `_alias`, `_state`, `_blacked`, `_dt`) values(?, ?, ?, ?, ?, ?)" + " ON DUPLICATE KEY UPDATE " + "`_alias` = ?," + "`_state` = ?," + + "`_blacked` = ?," + "`_dt` = ?"; @@ -2318,9 +2323,11 @@ public class DatabaseStore { statement.setString(index++, request.getFriendUid()); statement.setString(index++, request.getAlias()); statement.setInt(index++, request.getState()); + statement.setInt(index++, request.getBlacked()); statement.setLong(index++, request.getTimestamp()); statement.setString(index++, request.getAlias()); statement.setInt(index++, request.getState()); + statement.setInt(index++, request.getBlacked()); statement.setLong(index++, request.getTimestamp()); int count = statement.executeUpdate(); LOG.info("Update rows {}", count); diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 758b92e4..72efe862 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -30,6 +30,7 @@ import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; import io.moquette.spi.security.Tokenor; import io.moquette.spi.impl.subscriptions.Topic; +import io.netty.handler.codec.mqtt.MqttVersion; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import win.liyufan.im.*; @@ -1652,7 +1653,7 @@ public class MemoryMessagesStore implements IMessagesStore { } @Override - public List getFriendList(String userId, long version) { + public List getFriendList(String userId, String clientId, long version) { List out = new ArrayList(); HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); @@ -1662,9 +1663,26 @@ public class MemoryMessagesStore implements IMessagesStore { friends = loadFriend(friendsMap, userId); } + boolean needFriendMigrate = false; + + if (!StringUtil.isNullOrEmpty(clientId)) { + MemorySessionStore.Session session = m_Server.getStore().sessionsStore().getSession(clientId); + if(session != null && session.getMqttVersion() != null && session.getMqttVersion().protocolLevel() < MqttVersion.Wildfire_2.protocolLevel()) { + needFriendMigrate = true; + } + } + for (FriendData friend : friends) { if (friend.getTimestamp() > version) { - out.add(friend); + if (needFriendMigrate) { + if (friend.getBlacked() > 0) { + friend.setState(2); + } + out.add(friend); + } else { + out.add(friend); + } + } } @@ -1719,6 +1737,24 @@ public class MemoryMessagesStore implements IMessagesStore { return ErrorCode.ERROR_CODE_FRIEND_ALREADY_REQUEST; } } + + MultiMap friendsMap = hzInstance.getMultiMap(USER_FRIENDS); + + FriendData friendData1 = null; + Collection friendDatas = friendsMap.get(userId); + if (friendDatas == null || friendDatas.size() == 0) { + friendDatas = loadFriend(friendsMap, userId); + } + for (FriendData fd : friendDatas) { + if (fd.getFriendUid().equals(request.getTargetUid())) { + friendData1 = fd; + break; + } + } + if (friendData1 != null && friendData1.getBlacked() > 0) { + return ErrorCode.ERROR_CODE_IN_BLACK_LIST; + } + WFCMessage.FriendRequest newRequest = WFCMessage.FriendRequest .newBuilder() .setFromUid(userId) @@ -1770,7 +1806,7 @@ public class MemoryMessagesStore implements IMessagesStore { } } if (friendData1 == null) { - friendData1 = new FriendData(userId, request.getTargetUid(), "", request.getStatus(), System.currentTimeMillis()); + friendData1 = new FriendData(userId, request.getTargetUid(), "", request.getStatus(), 0, System.currentTimeMillis()); } else { friendData1.setState(request.getStatus()); friendData1.setTimestamp(System.currentTimeMillis()); @@ -1792,7 +1828,7 @@ public class MemoryMessagesStore implements IMessagesStore { } } if (friendData2 == null) { - friendData2 = new FriendData(request.getTargetUid(), userId, "", request.getStatus(), friendData1.getTimestamp()); + friendData2 = new FriendData(request.getTargetUid(), userId, "", request.getStatus(), 0, friendData1.getTimestamp()); } else { friendsMap.remove(request.getTargetUid(), friendData2); friendData2.setState(request.getStatus()); @@ -1833,10 +1869,10 @@ public class MemoryMessagesStore implements IMessagesStore { databaseStore.persistOrUpdateFriendRequest(existRequest); MultiMap friendsMap = hzInstance.getMultiMap(USER_FRIENDS); - FriendData friendData1 = new FriendData(userId, request.getTargetUid(), "", 0, System.currentTimeMillis()); + FriendData friendData1 = new FriendData(userId, request.getTargetUid(), "", 0, 0, System.currentTimeMillis()); databaseStore.persistOrUpdateFriendData(friendData1); - FriendData friendData2 = new FriendData(request.getTargetUid(), userId, "", 0, friendData1.getTimestamp()); + FriendData friendData2 = new FriendData(request.getTargetUid(), userId, "", 0, 0, friendData1.getTimestamp()); databaseStore.persistOrUpdateFriendData(friendData2); requestMap.remove(userId); @@ -1858,6 +1894,9 @@ public class MemoryMessagesStore implements IMessagesStore { @Override public ErrorCode blackUserRequest(String fromUser, String targetUserId, int state, long[] heads) { + if (state == 2) { + state = 1; + } HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); MultiMap friendsMap = hzInstance.getMultiMap(USER_FRIENDS); @@ -1875,9 +1914,9 @@ public class MemoryMessagesStore implements IMessagesStore { } if (friendData == null) { - friendData = new FriendData(fromUser, targetUserId, "", state, System.currentTimeMillis()); + friendData = new FriendData(fromUser, targetUserId, "", 1, state, System.currentTimeMillis()); } - friendData.setState(state); + friendData.setBlacked(state); friendData.setTimestamp(System.currentTimeMillis()); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 70cfbd19..12231d0c 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -153,7 +153,7 @@ public interface IMessagesStore { ErrorCode verifyToken(String userId, String token, List serverIPs, List ports); ErrorCode login(String name, String password, List userIdRet); - List getFriendList(String userId, long version); + List getFriendList(String userId, String clientId, long version); List getFriendRequestList(String userId, long version); ErrorCode saveAddFriendRequest(String userId, WFCMessage.AddFriendRequest request, long[] head); 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 7340b419..4ed5eea9 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -142,9 +142,8 @@ public class ProtocolProcessor { String clientId = payload.clientIdentifier(); LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName()); - if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel() - && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel() - && msg.variableHeader().version() != MqttVersion.Wildfire_1.protocolLevel()) { + if (msg.variableHeader().version() <= MqttVersion.MQTT_3_1_1.protocolLevel() || + msg.variableHeader().version() >= MqttVersion.Wildfire_Max.protocolLevel()) { MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION); LOG.error("MQTT protocol version is not valid. CId={}", clientId); 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 a1aea010..19806b42 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -178,7 +178,7 @@ public class Qos1PublishHandler extends QosPublishHandler { } else { MemorySessionStore.Session session = m_sessionStore.getSession(clientID); if (session != null && session.getUsername().equals(fromUser)) { - if (data.length > 7*1024 && session.getMqttVersion() == MqttVersion.Wildfire_1) { + if (data.length > 7*1024 && session.getMqttVersion().protocolLevel() >= MqttVersion.Wildfire_1.protocolLevel()) { ByteArrayOutputStream out = new ByteArrayOutputStream(); GZIPOutputStream gzip; try { diff --git a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java index 2c31db07..f5093201 100755 --- a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java +++ b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java @@ -43,7 +43,7 @@ final class MqttCodecUtil { return clientId != null && clientId.length() >= MIN_CLIENT_ID_LENGTH && clientId.length() <= MAX_CLIENT_ID_LENGTH; } - if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.Wildfire_1) { + if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) { // In 3.1.3.1 Client Identifier of MQTT 3.1.1 specification, The Server MAY allow ClientId’s // that contain more than 23 encoded bytes. And, The Server MAY allow zero-length ClientId. return clientId != null; diff --git a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java index ff66cdea..0be6d71a 100755 --- a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java +++ b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java @@ -223,7 +223,7 @@ public final class MqttDecoder extends ReplayingDecoder { final int willQos = (b1 & 0x18) >> 3; final boolean willFlag = (b1 & 0x04) == 0x04; final boolean cleanSession = (b1 & 0x02) == 0x02; - if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.Wildfire_1) { + if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) { final boolean zeroReservedFlag = (b1 & 0x01) == 0x0; if (!zeroReservedFlag) { // MQTT v3.1.1: The Server MUST validate that the reserved flag in the CONNECT Control Packet is diff --git a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java index cd7f6519..9d5b7ed6 100755 --- a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java +++ b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java @@ -25,7 +25,9 @@ import io.netty.util.internal.ObjectUtil; public enum MqttVersion { MQTT_3_1("MQIsdp", (byte) 3), MQTT_3_1_1("MQTT", (byte) 4), - Wildfire_1("MQTT", (byte) 5); + Wildfire_1("MQTT", (byte) 5), + Wildfire_2("MQTT", (byte) 6), + Wildfire_Max("MQTT", (byte) 7); private final String name; private final byte level;