From 4abf1ed941b2b0bbdc6e6931436fe89cd3b02f92 Mon Sep 17 00:00:00 2001 From: imhao183 Date: Fri, 8 Jan 2021 09:08:38 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=A4=9A=E7=AB=AF=E8=B8=A2?= =?UTF-8?q?=E4=B8=8B=E7=BA=BF=E7=9A=84=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../persistence/MemorySessionStore.java | 2 +- .../moquette/spi/impl/MessagesPublisher.java | 19 ++++++++++++ .../impl/PersistentQueueMessageSender.java | 4 +++ .../moquette/spi/impl/ProtocolProcessor.java | 31 ++++++++++++++----- .../src/main/java/win/liyufan/im/IMTopic.java | 2 ++ 5 files changed, 49 insertions(+), 9 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index e715d166..30a62cb3 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -407,8 +407,8 @@ public class MemorySessionStore implements ISessionsStore { if (remove) { sessions.remove(c); - mServer.getProcessor().kickoffSession(s); it.remove(); + mServer.getProcessor().kickoffSession(s); } } } diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index ae3faf80..a7efca08 100755 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -373,6 +373,25 @@ public class MessagesPublisher { } } } + + public boolean sendOfflineNotify(String clientId) { + boolean targetIsActive = this.connectionDescriptors.isConnected(clientId); + if (targetIsActive) { + ByteBuf payload = Unpooled.buffer(); + payload.ensureWritable(1).writeBytes("1".getBytes()); + MqttPublishMessage publishMsg; + publishMsg = notRetainedPublish(IMTopic.NotifyOffline, MqttQoS.AT_MOST_ONCE, payload); + + boolean sent = this.messageSender.sendPublish(clientId, publishMsg); + if (sent) { + return true; + } + } else { + LOG.info("the target {} is not active", clientId); + } + + return false; + } private void publishTransparentMessage2Receivers(long messageHead, Collection receivers, int pullType, String exceptClientId) { WFCMessage.Message message = m_messagesStore.getMessage(messageHead); diff --git a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java index 5aa8c28e..7e638640 100755 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -54,6 +54,10 @@ class PersistentQueueMessageSender { boolean sendPublish(ClientSession clientsession, MqttPublishMessage pubMessage) { String clientId = clientsession.clientID; + return sendPublish(clientId, pubMessage); + } + + boolean sendPublish(String clientId, MqttPublishMessage pubMessage) { final int messageId = pubMessage.variableHeader().packetId(); final String topicName = pubMessage.variableHeader().topicName(); if (LOG.isDebugEnabled()) { 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 c6295682..8b7f7c9f 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -72,15 +72,30 @@ public class ProtocolProcessor { public void kickoffSession(final MemorySessionStore.Session session) { mServer.getImBusinessScheduler().execute(()->{ - ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID()); - try { - if (descriptor != null) { - processDisconnect(descriptor.getChannel(), true, false); + new Thread(() -> { + messagesPublisher.sendOfflineNotify(session.getClientID()); + try { + Thread.sleep(3000); + } catch (InterruptedException e) { + e.printStackTrace(); } - } catch (InterruptedException e) { - e.printStackTrace(); - Utility.printExecption(LOG, e); - } + + mServer.getImBusinessScheduler().execute(()->{ + ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID()); + try { + if (descriptor != null) { + processDisconnect(descriptor.getChannel(), true, false); + } + } catch (InterruptedException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } catch (Exception e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + m_messagesStore.updateUserOnlineSetting(session, false); + }); + }).start(); }); } diff --git a/broker/src/main/java/win/liyufan/im/IMTopic.java b/broker/src/main/java/win/liyufan/im/IMTopic.java index 101993ae..98a74114 100755 --- a/broker/src/main/java/win/liyufan/im/IMTopic.java +++ b/broker/src/main/java/win/liyufan/im/IMTopic.java @@ -16,6 +16,8 @@ public interface IMTopic { String PullMessageTopic = "MP"; String NotifyMessageTopic = "MN"; String NotifyRecallMessageTopic = "RMN"; + String NotifyOffline = "ROFL"; + String BroadcastMessageTopic = "MBC"; String GetUserSettingTopic = "UG";