specs that checks connection count and upstream count

shard.lock

@@ -2,5 +2,5 @@ version: 1.0
 shards:
   amqp:
     github: cloudamqp/amqp.cr
-    commit: 692200ba05bb5e6479ec1e3eeaed38d67c61b3a4
+    commit: e5388a5dd6dd892be00a2ebf8b0f38e8e7221cea
 

spec/amqproxy_spec.cr

@@ -1,9 +1,17 @@
 require "./spec_helper"
 
-describe AMQProxy do
-  # TODO: Write tests
-
-  it "works" do
-    false.should eq(true)
+describe AMQProxy::Server do
+  it "keeps connections open" do
+    s = AMQProxy::Server.new("127.0.0.1", 5672, false)
+    spawn { s.listen("127.0.0.1", 5673) }
+    sleep 0.001
+    AMQP::Connection.start(AMQP::Config.new(port: 5673)) do |conn|
+      conn.channel
+      s.client_connections.should eq(1)
+      s.upstream_connections.should eq(1)
+    end
+    s.client_connections.should eq(0)
+    s.upstream_connections.should eq(1)
+    s.close
   end
 end

spec/spec_helper.cr

@@ -1,2 +1,4 @@
 require "spec"
-require "../src/amqproxy"
+require "../src/amqproxy/server"
+require "../src/amqproxy/version"
+require "amqp"

src/amqproxy/pool.cr

@@ -1,19 +1,24 @@
 module AMQProxy
   class Pool
+    getter :size
     def initialize(@host : String, @port : Int32, @tls : Bool)
       @pools = {} of String => Deque(Upstream)
+      @size = 0
     end
 
     def borrow(user : String, password : String, vhost : String, &block : (Upstream | Nil) -> _)
       q = @pools[[user, password, vhost].join] ||= Deque(Upstream).new
       u = q.shift do
+        @size += 1
         Upstream.new(@host, @port, @tls).connect(user, password, vhost)
       end
       block.call u
     ensure
       if u.nil?
+        @size -= 1
         print "Upstream connection could not be established\n"
       elsif u.closed?
+        @size -= 1
         print "Upstream connection closed when returned\n"
       elsif !q.nil?
         q.push u

src/amqproxy/server.cr

@@ -14,6 +14,13 @@ module AMQProxy
       print "TLS" if upstream_tls
       print "\n"
       @pool = Pool.new(upstream_host, upstream_port, upstream_tls)
+      @client_connections = 0
+    end
+
+    getter :client_connections
+
+    def upstream_connections
+      @pool.size
     end
 
     def listen(address, port)
@@ -64,6 +71,7 @@ module AMQProxy
     end
 
     def handle_connection(socket, remote_address)
+      @client_connections += 1
       c = Client.new(socket)
       print "Client connection accepted from ", remote_address, "\n"
       @pool.borrow(c.user, c.password, c.vhost) do |u|
@@ -97,6 +105,7 @@ module AMQProxy
     ensure
       print "Client connection closed from ", remote_address, "\n"
       socket.close
+      @client_connections -= 1
     end
   end
 end