Publish to exchange (#786)

Move the logic of publishing a message to an exchange from Vhost to the exchange. --------- Co-authored-by: Carl Hörberg <[email protected]>
cloudamqp · Oct 14, 2024 · 02e9aab · 02e9aab
1 parent a05fb6b
commit 02e9aab
Show file tree

Hide file tree

Showing 20 changed files with 559 additions and 429 deletions.
diff --git a/spec/api/bindings_spec.cr b/spec/api/bindings_spec.cr
@@ -65,7 +65,7 @@ describe LavinMQ::HTTP::BindingsController do
         response = http.post("/api/bindings/%2f/e/be1/q/bindings_q1", body: body)
         response.status_code.should eq 201
         response.headers["Location"].should eq "bindings_q1/rk"
-        s.vhosts["/"].exchanges["be1"].queue_bindings.last_key.first.should eq "rk"
+        s.vhosts["/"].exchanges["be1"].bindings_details.first.routing_key.should eq "rk"
       end
     end
 
@@ -127,7 +127,7 @@ describe LavinMQ::HTTP::BindingsController do
         props = binding[0]["properties_key"].as_s
         response = http.delete("/api/bindings/%2f/e/be1/q/bindings_q1/#{props}")
         response.status_code.should eq 204
-        s.vhosts["/"].exchanges["be1"].queue_bindings.empty?.should be_true
+        s.vhosts["/"].exchanges["be1"].bindings_details.empty?.should be_true
       end
     end
   end

diff --git a/spec/api/definitions_spec.cr b/spec/api/definitions_spec.cr
@@ -106,13 +106,20 @@ describe LavinMQ::HTTP::Server do
       ]})
         response = http.post("/api/definitions", body: body)
         response.status_code.should eq 200
-        matches = [] of String
         ex = s.vhosts["/"].exchanges["import_x1"]
-        ex.do_exchange_matches("r.k2", nil) { |e| matches << e.name }
-        matches.includes?("import_x2").should be_true
-        matches.clear
-        ex.do_queue_matches("rk", nil) { |e| matches << e.name }
-        matches.includes?("import_q1").should be_true
+        qs = Set(LavinMQ::Queue).new
+        es = Set(LavinMQ::Exchange).new
+        ex.find_queues("r.k2", nil, qs, es)
+        res = Set(LavinMQ::Exchange).new
+        res << s.vhosts["/"].exchanges["import_x1"]
+        res << s.vhosts["/"].exchanges["import_x2"]
+        es.should eq res
+        qs = Set(LavinMQ::Queue).new
+        es = Set(LavinMQ::Exchange).new
+        ex.find_queues("rk", nil, qs, es)
+        res = Set(LavinMQ::Queue).new
+        res << s.vhosts["/"].queues["import_q1"]
+        qs.should eq res
       end
     end
 
@@ -470,13 +477,20 @@ describe LavinMQ::HTTP::Server do
       ]})
         response = http.post("/api/definitions/%2f", body: body)
         response.status_code.should eq 200
-        matches = [] of String
         ex = s.vhosts["/"].exchanges["import_x1"]
-        ex.do_exchange_matches("r.k2", nil) { |e| matches << e.name }
-        matches.includes?("import_x2").should be_true
-        matches.clear
-        ex.do_queue_matches("rk", nil) { |e| matches << e.name }
-        matches.includes?("import_q1").should be_true
+        qs = Set(LavinMQ::Queue).new
+        es = Set(LavinMQ::Exchange).new
+        ex.find_queues("r.k2", nil, qs, es)
+        res = Set(LavinMQ::Exchange).new
+        res << s.vhosts["/"].exchanges["import_x1"]
+        res << s.vhosts["/"].exchanges["import_x2"]
+        es.should eq res
+        qs = Set(LavinMQ::Queue).new
+        es = Set(LavinMQ::Exchange).new
+        ex.find_queues("rk", nil, qs, es)
+        res = Set(LavinMQ::Queue).new
+        res << s.vhosts["/"].queues["import_q1"]
+        qs.should eq res
       end
     end
 

diff --git a/spec/exchange_spec.cr b/spec/exchange_spec.cr
@@ -124,4 +124,50 @@ describe LavinMQ::Exchange do
       end
     end
   end
+
+  describe "in_use?" do
+    it "should not be in use when just created" do
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          ch.exchange("e1", "topic", auto_delete: true)
+          ch.exchange("e2", "topic", auto_delete: true)
+          s.vhosts["/"].exchanges["e1"].in_use?.should be_false
+          s.vhosts["/"].exchanges["e2"].in_use?.should be_false
+        end
+      end
+    end
+    it "should be in use when it has bindings" do
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          x1 = ch.exchange("e1", "topic", auto_delete: true)
+          x2 = ch.exchange("e2", "topic", auto_delete: true)
+          x2.bind(x1.name, "#")
+          s.vhosts["/"].exchanges["e2"].in_use?.should be_true
+        end
+      end
+    end
+    it "should be in use when other exchange has binding to it" do
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          x1 = ch.exchange("e1", "topic", auto_delete: true)
+          x2 = ch.exchange("e2", "topic", auto_delete: true)
+          x2.bind(x1.name, "#")
+          s.vhosts["/"].exchanges["e1"].in_use?.should be_true
+        end
+      end
+    end
+
+    it "should be in use when it has bindings" do
+      with_amqp_server do |s|
+        with_channel(s) do |ch|
+          x1 = ch.exchange("e1", "topic")
+          x2 = ch.exchange("e2", "topic", auto_delete: true)
+          x2.bind(x1.name, "#")
+          s.vhosts["/"].exchanges["e1"].in_use?.should be_true
+          x2.unbind(x1.name, "#")
+          s.vhosts["/"].exchanges["e1"].in_use?.should be_false
+        end
+      end
+    end
+  end
 end
diff --git a/spec/message_routing_spec.cr b/spec/message_routing_spec.cr
@@ -5,8 +5,10 @@ module LavinMQ
     # Monkey patch for backward compability and easier testing
     def matches(routing_key, headers = nil) : Set(Queue | Exchange)
       s = Set(Queue | Exchange).new
-      queue_matches(routing_key, headers) { |q| s << q }
-      exchange_matches(routing_key, headers) { |x| s << x }
+      qs = Set(Queue).new
+      es = Set(Exchange).new
+      find_queues(routing_key, headers, qs, es)
+      qs.each { |q| s << q }
       s
     end
   end
@@ -19,15 +21,20 @@ describe LavinMQ::DirectExchange do
       q1 = LavinMQ::Queue.new(vhost, "q1")
       x = LavinMQ::DirectExchange.new(vhost, "")
       x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
-      x.matches("q1").should eq(Set{q1})
+      found_queues = Set(LavinMQ::Queue).new
+      x.find_queues("q1", nil, found_queues, Set(LavinMQ::Exchange).new)
+      found_queues.should eq(Set{q1})
     end
   end
 
   it "matches no rk" do
     with_amqp_server do |s|
       vhost = s.vhosts.create("x")
       x = LavinMQ::DirectExchange.new(vhost, "")
-      x.matches("q1").should be_empty
+
+      found_queues = Set(LavinMQ::Queue).new
+      x.find_queues("q1", nil, found_queues, Set(LavinMQ::Exchange).new)
+      found_queues.should be_empty
     end
   end
 end
@@ -39,15 +46,20 @@ describe LavinMQ::FanoutExchange do
       q1 = LavinMQ::Queue.new(vhost, "q1")
       x = LavinMQ::FanoutExchange.new(vhost, "")
       x.bind(q1, "")
-      x.matches("any").should eq(Set{q1})
+
+      found_queues = Set(LavinMQ::Queue).new
+      x.find_queues("q1", nil, found_queues, Set(LavinMQ::Exchange).new)
+      found_queues.should eq(Set{q1})
     end
   end
 
   it "matches no rk" do
     with_amqp_server do |s|
       vhost = s.vhosts.create("x")
       x = LavinMQ::FanoutExchange.new(vhost, "")
-      x.matches("q1").should be_empty
+      found_queues = Set(LavinMQ::Queue).new
+      x.find_queues("q1", nil, found_queues, Set(LavinMQ::Exchange).new)
+      found_queues.should be_empty
     end
   end
 end
@@ -307,3 +319,105 @@ describe LavinMQ::HeadersExchange do
     end
   end
 end
+
+describe LavinMQ::Exchange do
+  it "should handle CC in header" do
+    with_amqp_server do |s|
+      vhost = s.vhosts.create("x")
+      q1 = LavinMQ::Queue.new(vhost, "q1")
+      q2 = LavinMQ::Queue.new(vhost, "q2")
+      x = LavinMQ::DirectExchange.new(vhost, "")
+      x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
+      x.bind(q2, "q2", LavinMQ::AMQP::Table.new)
+      found_queues = Set(LavinMQ::Queue).new
+      headers = LavinMQ::AMQP::Table.new
+      headers["CC"] = ["q2"]
+      x.find_queues("q1", headers, found_queues)
+      found_queues.should eq(Set{q1, q2})
+    end
+  end
+  it "should raise if CC header isn't array" do
+    with_amqp_server do |s|
+      vhost = s.vhosts.create("x")
+      q1 = LavinMQ::Queue.new(vhost, "q1")
+      q2 = LavinMQ::Queue.new(vhost, "q2")
+      x = LavinMQ::DirectExchange.new(vhost, "")
+      x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
+      x.bind(q2, "q2", LavinMQ::AMQP::Table.new)
+      found_queues = Set(LavinMQ::Queue).new
+      headers = LavinMQ::AMQP::Table.new
+      headers["CC"] = "q2"
+      expect_raises(LavinMQ::Error::PreconditionFailed) do
+        x.find_queues("q1", headers, found_queues)
+      end
+    end
+  end
+
+  it "should handle BCC in header" do
+    with_amqp_server do |s|
+      vhost = s.vhosts.create("x")
+      q1 = LavinMQ::Queue.new(vhost, "q1")
+      q2 = LavinMQ::Queue.new(vhost, "q2")
+      x = LavinMQ::DirectExchange.new(vhost, "")
+      x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
+      x.bind(q2, "q2", LavinMQ::AMQP::Table.new)
+      found_queues = Set(LavinMQ::Queue).new
+      headers = LavinMQ::AMQP::Table.new
+      headers["BCC"] = ["q2"]
+      x.find_queues("q1", headers, found_queues)
+      found_queues.should eq(Set{q1, q2})
+    end
+  end
+
+  it "should raise if BCC header isn't array" do
+    with_amqp_server do |s|
+      vhost = s.vhosts.create("x")
+      q1 = LavinMQ::Queue.new(vhost, "q1")
+      q2 = LavinMQ::Queue.new(vhost, "q2")
+      x = LavinMQ::DirectExchange.new(vhost, "")
+      x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
+      x.bind(q2, "q2", LavinMQ::AMQP::Table.new)
+      found_queues = Set(LavinMQ::Queue).new
+      headers = LavinMQ::AMQP::Table.new
+      headers["BCC"] = "q2"
+      expect_raises(LavinMQ::Error::PreconditionFailed) do
+        x.find_queues("q1", headers, found_queues)
+      end
+    end
+  end
+
+  it "should drop BCC from header" do
+    with_amqp_server do |s|
+      vhost = s.vhosts.create("x")
+      q1 = LavinMQ::Queue.new(vhost, "q1")
+      q2 = LavinMQ::Queue.new(vhost, "q2")
+      x = LavinMQ::DirectExchange.new(vhost, "")
+      x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
+      x.bind(q2, "q2", LavinMQ::AMQP::Table.new)
+      found_queues = Set(LavinMQ::Queue).new
+      headers = LavinMQ::AMQP::Table.new
+      headers["BCC"] = ["q2"]
+      x.find_queues("q1", headers, found_queues)
+      headers["BCC"]?.should be_nil
+    end
+  end
+
+  it "should read both CC and BCC" do
+    with_amqp_server do |s|
+      vhost = s.vhosts.create("x")
+      q1 = LavinMQ::Queue.new(vhost, "q1")
+      q2 = LavinMQ::Queue.new(vhost, "q2")
+      q3 = LavinMQ::Queue.new(vhost, "q3")
+      x = LavinMQ::DirectExchange.new(vhost, "")
+      x.bind(q1, "q1", LavinMQ::AMQP::Table.new)
+      x.bind(q2, "q2", LavinMQ::AMQP::Table.new)
+      x.bind(q3, "q3", LavinMQ::AMQP::Table.new)
+      found_queues = Set(LavinMQ::Queue).new
+      headers = LavinMQ::AMQP::Table.new
+      headers["CC"] = ["q2"]
+      headers["BCC"] = ["q3"]
+      x.find_queues("q1", headers, found_queues)
+      found_queues.should eq(Set{q1, q2, q3})
+    end
+  end
+end
diff --git a/spec/queue_spec.cr b/spec/queue_spec.cr
@@ -59,7 +59,7 @@ describe LavinMQ::Queue do
           x.publish_confirm "test message", q.name
           q.get(no_ack: true).try(&.body_io.to_s).should eq("test message")
 
-          iq = s.vhosts["/"].exchanges[x_name].queue_bindings[{q.name, nil}].first
+          iq = s.vhosts["/"].queues[q.name]
           iq.pause!
 
           x.publish_confirm "test message 2", q.name
@@ -93,7 +93,7 @@ describe LavinMQ::Queue do
           x.publish_confirm "test message", q.name
           q.get(no_ack: true).try(&.body_io.to_s).should eq("test message")
 
-          iq = s.vhosts["/"].exchanges[x_name].queue_bindings[{q.name, nil}].first
+          iq = s.vhosts["/"].queues[q.name]
           iq.pause!
 
           x.publish_confirm "test message 2", q.name
@@ -126,7 +126,7 @@ describe LavinMQ::Queue do
           x.publish_confirm "test message", q.name
           q.get(no_ack: true).try(&.body_io.to_s).should eq("test message")
 
-          iq = s.vhosts["/"].exchanges[x_name].queue_bindings[{q.name, nil}].first
+          iq = s.vhosts["/"].queues[q.name]
           iq.pause!
 
           x.publish_confirm "test message 2", q.name
@@ -187,7 +187,7 @@ describe LavinMQ::Queue do
           x.publish_confirm "test message 3", q.name
           x.publish_confirm "test message 4", q.name
 
-          internal_queue = s.vhosts["/"].exchanges[x_name].queue_bindings[{q.name, nil}].first
+          internal_queue = s.vhosts["/"].queues[q.name]
           internal_queue.message_count.should eq 4
 
           response = http.delete("/api/queues/%2f/#{q_name}/contents")
@@ -209,7 +209,7 @@ describe LavinMQ::Queue do
           end
 
           vhost = s.vhosts["/"]
-          internal_queue = vhost.exchanges[x_name].queue_bindings[{q.name, nil}].first
+          internal_queue = vhost.queues[q.name]
           internal_queue.message_count.should eq 10
 
           response = http.delete("/api/queues/%2f/#{q_name}/contents?count=5")
@@ -235,8 +235,7 @@ describe LavinMQ::Queue do
             x.publish_confirm "test message #{i}", q.name
           end
 
-          internal_queue = s.vhosts["/"].exchanges[x_name].queue_bindings[{q.name, nil}].first
-
+          internal_queue = s.vhosts["/"].queues[q.name]
           internal_queue.message_count.should eq 10
 
           channel = Channel(String).new(1)

diff --git a/spec/upstream_spec.cr b/spec/upstream_spec.cr
@@ -432,18 +432,18 @@ describe LavinMQ::Federation::Upstream do
         wait_for { upstream.links.first?.try &.state.running? }
 
         upstream_q = upstream_vhost.queues.values.first
-        upstream_q.bindings.size.should eq queues.size
+        upstream_q.bindings.size.should eq queues.size + 1 # +1 for the default exchange
         # Assert setup is correct
         10.times do |i|
           downstream_q = downstream_ch.queue("")
           downstream_q.bind("downstream_ex", "after.link.#{i}")
           queues << downstream_q
         end
         sleep 0.1.seconds
-        upstream_q.bindings.size.should eq queues.size
+        upstream_q.bindings.size.should eq queues.size + 1
         queues.each &.delete
         sleep 10.milliseconds
-        upstream_q.bindings.size.should eq 0
+        upstream_q.bindings.size.should eq 1
       end
     end
   end

diff --git a/spec/vhost_spec.cr b/spec/vhost_spec.cr
@@ -71,7 +71,7 @@ describe LavinMQ::VHost do
       v.declare_queue("q", true, false)
       s.vhosts["test"].bind_queue("q", "e", "q")
       s.restart
-      s.vhosts["test"].exchanges["e"].queue_bindings[{"q", nil}].size.should eq 1
+      s.vhosts["test"].exchanges["e"].bindings_details.first.destination.name.should eq "q"
     end
   end