From afee0466e7f50d21c4156aa78c865c8ff2bc141a Mon Sep 17 00:00:00 2001 From: heavyrian2012 Date: Thu, 13 Apr 2023 09:18:43 +0800 Subject: [PATCH] =?UTF-8?q?=E7=94=A8=E6=88=B7=E7=8A=B6=E6=80=81=E5=92=8C?= =?UTF-8?q?=E9=A2=91=E9=81=93=E8=AE=A2=E9=98=85=E7=8A=B6=E6=80=81=E4=BF=AE?= =?UTF-8?q?=E6=94=B9=E6=94=B9=E6=88=90=E5=90=8C=E6=AD=A5=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=EF=BC=8C=E9=98=B2=E6=AD=A2=E9=AB=98=E5=B9=B6=E5=8F=91=E6=97=B6?= =?UTF-8?q?=E5=87=BA=E7=8E=B0=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../moquette/persistence/DatabaseStore.java | 162 +++++++++--------- 1 file changed, 78 insertions(+), 84 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 130d6167..d175ec92 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -2411,59 +2411,55 @@ public class DatabaseStore { } void deleteUserStatus(String userId) { - mScheduler.execute(()->{ - Connection connection = null; - PreparedStatement statement = null; - try { - connection = DBUtil.getConnection(); + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + String sql = "delete from t_user_status where _uid = ?"; + statement = connection.prepareStatement(sql); + statement.setString(1, userId); + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement); + } + } + + void updateUserStatus(String userId, int status) { + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + if (status == 0) { String sql = "delete from t_user_status where _uid = ?"; statement = connection.prepareStatement(sql); statement.setString(1, userId); int count = statement.executeUpdate(); LOG.info("Update rows {}", count); - } catch (SQLException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - Utility.printExecption(LOG, e, RDBS_Exception); - } finally { - DBUtil.closeDB(connection, statement); + } else { + String sql = "insert into t_user_status (`_uid`, `_status`, `_dt`) values(?,?,?) ON DUPLICATE KEY UPDATE `_status` = ?"; + + statement = connection.prepareStatement(sql); + int index = 1; + statement.setString(index++, userId); + statement.setInt(index++, status); + statement.setLong(index++, System.currentTimeMillis()); + statement.setInt(index++, status); + + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); } - }); - } - - void updateUserStatus(String userId, int status) { - mScheduler.execute(()->{ - Connection connection = null; - PreparedStatement statement = null; - try { - connection = DBUtil.getConnection(); - if (status == 0) { - String sql = "delete from t_user_status where _uid = ?"; - statement = connection.prepareStatement(sql); - statement.setString(1, userId); - int count = statement.executeUpdate(); - LOG.info("Update rows {}", count); - } else { - String sql = "insert into t_user_status (`_uid`, `_status`, `_dt`) values(?,?,?) ON DUPLICATE KEY UPDATE `_status` = ?"; - - statement = connection.prepareStatement(sql); - int index = 1; - statement.setString(index++, userId); - statement.setInt(index++, status); - statement.setLong(index++, System.currentTimeMillis()); - statement.setInt(index++, status); - - int count = statement.executeUpdate(); - LOG.info("Update rows {}", count); - } - } catch (SQLException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - Utility.printExecption(LOG, e, RDBS_Exception); - } finally { - DBUtil.closeDB(connection, statement); - } - }); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement); + } } int getGeneratedId() { @@ -3342,47 +3338,45 @@ public class DatabaseStore { } void updateChannelListener(final String channelId, final String listener, final boolean listen) { - LOG.info("Database remove channel {}", channelId); - mScheduler.execute(()->{ - Connection connection = null; - PreparedStatement statement = null; - try { - connection = DBUtil.getConnection(); - String sql; - if (listen) { - sql = "insert into t_channel_listener (`_cid`" + - ", `_mid`" + - ", `_dt`) values(?, ?, ?)" + - " ON DUPLICATE KEY UPDATE " + - "`_dt` = ?"; - statement = connection.prepareStatement(sql); - int index = 1; - statement.setString(index++, channelId); - statement.setString(index++, listener); - statement.setLong(index++, System.currentTimeMillis()); - statement.setLong(index++, System.currentTimeMillis()); + LOG.info("updateChannelListener channel {}, listener {}, listen {}", channelId, listener, listen); + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + String sql; + if (listen) { + sql = "insert into t_channel_listener (`_cid`" + + ", `_mid`" + + ", `_dt`) values(?, ?, ?)" + + " ON DUPLICATE KEY UPDATE " + + "`_dt` = ?"; + statement = connection.prepareStatement(sql); + int index = 1; + statement.setString(index++, channelId); + statement.setString(index++, listener); + statement.setLong(index++, System.currentTimeMillis()); + statement.setLong(index++, System.currentTimeMillis()); - int count = statement.executeUpdate(); - LOG.info("Update rows {}", count); - } else { - sql = "delete from t_channel_listener where `_cid`=? and `_mid`=?"; + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); + } else { + sql = "delete from t_channel_listener where `_cid`=? and `_mid`=?"; - statement = connection.prepareStatement(sql); - int index = 1; - statement.setString(index++, channelId); - statement.setString(index++, listener); + statement = connection.prepareStatement(sql); + int index = 1; + statement.setString(index++, channelId); + statement.setString(index++, listener); - int count = statement.executeUpdate(); - LOG.info("Update rows {}", count); - } - } catch (SQLException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - Utility.printExecption(LOG, e, RDBS_Exception); - } finally { - DBUtil.closeDB(connection, statement); + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); } - }); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement); + } } List getUserChannels(String userId) {