suggestions (#1224)

Author: snichme

spec/exchange_spec.cr

@@ -72,7 +72,7 @@ describe LavinMQ::Exchange do
             amqp_queue = s.vhosts["/"].queues["mqtt-messages"]
 
             mqtt_exchange.bind(amqp_queue, "home/lights", nil)
-            mqtt_exchange.publish(mqtt_publish("home/lights", "ON"), Set(LavinMQ::Queue).new, Set(LavinMQ::Exchange).new).should eq 1
+            mqtt_exchange.publish(mqtt_publish("home/lights", "ON")).should eq 1
             amqp_queue.message_count.should eq 1
           end
         end
@@ -89,7 +89,7 @@ describe LavinMQ::Exchange do
 
             mqtt_exchange.bind(sensor_queue, "sensor/+/temperature", nil)
             mqtt_exchange.bind(all_queue, "#", nil)
-            mqtt_exchange.publish(mqtt_publish("sensor/bedroom/temperature", "22.5"), Set(LavinMQ::Queue).new, Set(LavinMQ::Exchange).new).should eq 2
+            mqtt_exchange.publish(mqtt_publish("sensor/bedroom/temperature", "22.5")).should eq 2
             sensor_queue.message_count.should eq 1
             all_queue.message_count.should eq 1
           end

src/lavinmq/amqp/exchange/exchange.cr

@@ -218,7 +218,14 @@ module LavinMQ
                       queues : Set(LavinMQ::Queue) = Set(LavinMQ::Queue).new,
                       exchanges : Set(LavinMQ::Exchange) = Set(LavinMQ::Exchange).new) : Nil
         return unless exchanges.add? self
-        find_queues_internal(routing_key, headers, queues, exchanges)
+        each_destination(routing_key, headers) do |d|
+          case d
+          in LavinMQ::Queue
+            queues.add(d)
+          in LavinMQ::Exchange
+            d.find_queues(routing_key, headers, queues, exchanges)
+          end
+        end
 
         if hdrs = headers
           find_cc_queues(hdrs, "CC", queues)
@@ -233,17 +240,6 @@ module LavinMQ
         end
       end
 
-      protected def find_queues_internal(routing_key, headers, queues, exchanges)
-        each_destination(routing_key, headers) do |d|
-          case d
-          in LavinMQ::Queue
-            queues.add(d)
-          in LavinMQ::Exchange
-            d.find_queues(routing_key, headers, queues, exchanges)
-          end
-        end
-      end
-
       private def find_cc_queues(headers, key, queues)
         return unless cc = headers[key]?
         cc = cc.as?(Array(AMQP::Field))

src/lavinmq/mqtt/broker.cr

@@ -70,8 +70,8 @@ module LavinMQ
         @vhost.rm_connection(client)
       end
 
-      def publish(packet : MQTT::Publish, queues : Set(LavinMQ::Queue), exchanges : Set(LavinMQ::Exchange))
-        @exchange.publish(packet, queues, exchanges)
+      def publish(packet : MQTT::Publish)
+        @exchange.publish(packet)
       end
 
       def subscribe(client, topics)

src/lavinmq/mqtt/client.cr

@@ -40,8 +40,6 @@ module LavinMQ
         metadata = ::Log::Metadata.new(nil, {vhost: @broker.vhost.name, address: @connection_info.remote_address.to_s, client_id: client_id})
         @log = Logger.new(Log, metadata)
         @log.info { "Connection established for user=#{@user.name}" }
-        @queues = Set(LavinMQ::Queue).new
-        @exchanges = Set(LavinMQ::Exchange).new
         spawn read_loop, name: "MQTT read_loop #{@connection_info.remote_address}"
       end
 
@@ -129,7 +127,7 @@ module LavinMQ
       end
 
       def recieve_publish(packet : MQTT::Publish)
-        @broker.publish(packet, @queues, @exchanges)
+        @broker.publish(packet)
         vhost.event_tick(EventType::ClientPublish)
         # Ok to not send anything if qos = 0 (fire and forget)
         if packet.qos > 0 && (packet_id = packet.packet_id)
@@ -188,7 +186,7 @@ module LavinMQ
             qos: will.qos,
             retain: will.retain?,
             dup: false,
-          ), @queues, @exchanges
+          )
         end
       rescue ex
         @log.warn { "Failed to publish will: #{ex.message}" }

src/lavinmq/mqtt/exchange.cr

@@ -37,6 +37,12 @@ module LavinMQ
 
       def initialize(vhost : VHost, name : String, @retain_store : MQTT::RetainStore)
         super(vhost, name, false, false, true)
+        @queues = Set(LavinMQ::Queue).new
+        @exchanges = Set(LavinMQ::Exchange).new
+      end
+
+      def publish(packet : MQTT::Publish) : UInt32
+        publish(packet, @queues, @exchanges)
       end
 
       def publish(packet : MQTT::Publish, queues : Set(LavinMQ::Queue), exchanges : Set(LavinMQ::Exchange)) : UInt32
@@ -93,17 +99,6 @@ module LavinMQ
         # Use only the subscription tree for all destinations (MQTT and AMQP)
       end
 
-      protected def find_queues_internal(routing_key, headers, queues, exchanges)
-        @tree.each_entry(routing_key) do |destination, _|
-          case destination
-          in LavinMQ::Queue
-            queues.add(destination)
-          in LavinMQ::Exchange
-            destination.find_queues(routing_key, headers, queues, exchanges)
-          end
-        end
-      end
-
       def bind(destination : Destination, routing_key : String, arguments = nil) : Bool
         qos = arguments.try { |h| h[QOS_HEADER]?.try(&.as(UInt8)) } || 0u8
         binding_key = BindingKey.new(routing_key, arguments)