pass more specs and handle session_present

spec/mqtt/integrations/connect_spec.cr

@@ -25,7 +25,7 @@ module MqttSpecs
             with_client_io(server) do |io|
               connect(io, clean_session: false)
 
-              # Myra won't save sessions without subscriptions
+              # LavinMQ won't save sessions without subscriptions
               subscribe(io,
                 topic_filters: [subtopic("a/topic", 0u8)],
                 packet_id: 1u16

src/lavinmq/mqtt/broker.cr

@@ -8,16 +8,29 @@ module LavinMQ
         @sessions = Hash(String, Session).new
       end
 
-      def start_session(client : Client)
-        client_id = client.client_id
-        session = MQTT::Session.new(self, client_id)
+      def clean_session?(client_id : String) : Bool
+        session = @sessions[client_id]?
+        return false if session.nil?
+        session.set_clean_session
+      end
+
+      def session_present?(client_id : String, clean_session) : Bool
+        session = @sessions[client_id]?
+        clean_session?(client_id)
+        return false if session.nil? || clean_session
+        true
+      end
+
+      def start_session(client_id, clean_session)
+        session = MQTT::Session.new(@vhost, client_id)
+        session.clean_session if clean_session
         @sessions[client_id] = session
         @queues[client_id] = session
       end
 
-      def clear_session(client : Client)
-        @sessions.delete client.client_id
-        @queues.delete client.client_id
+      def clear_session(client_id)
+        @sessions.delete client_id
+        @queues.delete client_id
       end
 
       # def connected(client) : MQTT::Session

src/lavinmq/mqtt/client.cr

@@ -57,7 +57,7 @@ module LavinMQ
         Log.info { "eof #{ex.inspect}" }
       ensure
         publish_will if @will
-        disconnect_session(self) if @clean_session
+        @broker.clear_session(client_id) if @clean_session
         @socket.close
         @broker.vhost.rm_connection(self)
       end
@@ -114,6 +114,7 @@ module LavinMQ
         auto_delete = true
         tbl = AMQP::Table.new
         # TODO: declare Session instead
+        @broker.start_session(@client_id, @clean_session)
         q = @broker.vhost.declare_queue(name, durable, auto_delete, tbl)
         qos = Array(MQTT::SubAck::ReturnCode).new
         packet.topic_filters.each do |tf|
@@ -145,21 +146,20 @@ module LavinMQ
 
       def start_session(client) : MQTT::Session
         if @clean_session
-          pp "clear session"
+           Log.trace { "clear session" }
           @broker.clear_session(client)
         end
         @broker.start_session(client)
       end
 
       def disconnect_session(client)
-        pp "disconnect session"
+        Log.trace { "disconnect session" }
         @broker.clear_session(client)
       end
 
       # TODO: actually publish will to session
       private def publish_will
         if will = @will
-          pp "Publish will to session"
         end
       rescue ex
         Log.warn { "Failed to publish will: #{ex.message}" }

src/lavinmq/mqtt/connection_factory.cr

@@ -20,7 +20,9 @@ module LavinMQ
         if packet = MQTT::Packet.from_io(socket).as?(MQTT::Connect)
           Log.trace { "recv #{packet.inspect}" }
           if user = authenticate(io, packet)
-            MQTT::Connack.new(false, MQTT::Connack::ReturnCode::Accepted).to_io(io)
+            session_present = @broker.session_present?(packet.client_id, packet.clean_session?)
+            pp "in connection_factory: #{session_present}"
+            MQTT::Connack.new(session_present, MQTT::Connack::ReturnCode::Accepted).to_io(io)
             io.flush
             return LavinMQ::MQTT::Client.new(socket, connection_info, user, @vhost, @broker, packet.client_id, packet.clean_session?, packet.will)
           end

src/lavinmq/mqtt/session.cr

@@ -1,6 +1,8 @@
 module LavinMQ
   module MQTT
     class Session < Queue
+       @clean_session : Bool = false
+       getter clean_session
       def initialize(@vhost : VHost,
                      @name : String,
                      @exclusive = true,
@@ -9,6 +11,9 @@ module LavinMQ
         super
       end
 
+      def set_clean_session
+        @clean_session = true
+      end
 
       #TODO: implement subscribers array and session_present? and send instead of false
       def connect(client)