diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 15557d5f..ecadfe19 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -30,8 +30,6 @@ public interface IMessaging { void stop(); - void disconnect(ServerChannel session); - void lostConnection(String clientID); void handleProtocolMessage(ServerChannel session, AbstractMessage msg); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index df83a1b1..70c91d9b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -65,12 +65,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { private static SimpleMessaging INSTANCE; - private ProtocolProcessor m_processor = new ProtocolProcessor(); + private final ProtocolProcessor m_processor = new ProtocolProcessor(); + private final AnnotationHelper annotationHelper = new AnnotationHelper(); CountDownLatch m_stopLatch; - private AnnotationHelper annotationHelper = new AnnotationHelper(); - private SimpleMessaging() { } @@ -109,19 +108,17 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_ringBuffer.publish(sequence); } - - public void disconnect(ServerChannel session) { - disruptorPublish(new DisconnectEvent(session)); - } - + @Override public void lostConnection(String clientID) { disruptorPublish(new LostConnectionEvent(clientID)); } + @Override public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) { disruptorPublish(new ProtocolEvent(session, msg)); } + @Override public void stop() { m_stopLatch = new CountDownLatch(1); disruptorPublish(new StopEvent()); @@ -143,11 +140,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_processor.processPublish((PublishEvent) evt); } else if (evt instanceof StopEvent) { processStop(); - } else if (evt instanceof DisconnectEvent) { + } /*else if (evt instanceof DisconnectEvent) { DisconnectEvent disEvt = (DisconnectEvent) evt; String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); m_processor.processDisconnect(disEvt.getSession(), clientID, false); - } else if (evt instanceof ProtocolEvent) { + } */else if (evt instanceof ProtocolEvent) { ServerChannel session = ((ProtocolEvent) evt).getSession(); AbstractMessage message = ((ProtocolEvent) evt).getMessage(); if (message instanceof ConnectMessage || @@ -174,9 +171,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { throw new RuntimeException("Illegal message received " + message); } - } /*else if (evt instanceof InitEvent) { - processInit(((InitEvent) evt).getConfig()); - } */else if (evt instanceof LostConnectionEvent) { + } else if (evt instanceof LostConnectionEvent) { LostConnectionEvent lostEvt = (LostConnectionEvent) evt; m_processor.proccessConnectionLost(lostEvt.getClientID()); }