close listening socket in #stop not to accept new connections anymore

lib/fluent/plugin/in_forward.rb

@@ -147,17 +147,19 @@ def configure(conf)
     def start
       super
 
+      shared_socket = false
+
       server_create_connection(
         :in_forward_server, @port,
         bind: @bind,
-        shared: false,
+        shared: shared_socket,
         resolve_name: @resolve_hostname,
         linger_timeout: @linger_timeout,
         backlog: @backlog,
         &method(:handle_connection)
       )
 
-      server_create(:in_forward_server_udp_heartbeat, @port, shared: false, proto: :udp, bind: @bind, resolve_name: @resolve_hostname, max_bytes: 128) do |data, sock|
+      server_create(:in_forward_server_udp_heartbeat, @port, shared: shared_socket, proto: :udp, bind: @bind, resolve_name: @resolve_hostname, max_bytes: 128) do |data, sock|
         log.trace "heartbeat udp data arrived", host: sock.remote_host, port: sock.remote_port, data: data
         begin
           sock.write HEARTBEAT_UDP_PAYLOAD

lib/fluent/plugin_helper/server.rb

@@ -205,28 +205,23 @@ def initialize
         @_server_mutex = Mutex.new
       end
 
-      def shutdown
-        @_server_connections.each do |conn|
-          conn.close rescue nil
-        end
+      def stop
         @_server_mutex.synchronize do
           @_servers.each do |si|
             si.server.detach if si.server.attached?
+            # to refuse more connections: (connected sockets are still alive here)
+            si.server.close rescue nil
           end
         end
 
         super
       end
 
-      def close
+      def shutdown
         @_server_connections.each do |conn|
           conn.close rescue nil
         end
-        @_server_mutex.synchronize do
-          @_servers.each do |si|
-            si.server.close rescue nil
-          end
-        end
+
         super
       end
 

test/plugin_helper/test_server.rb

@@ -897,6 +897,84 @@ class Dummy < Fluent::Plugin::TestBase
       assert_equal 0, @d.instance_eval{ @_server_connections.size }
     end
 
+    data(protocols)
+    test 'will refuse more connect requests after stop, but read data from sockets already connected, in non-shared server' do |(proto, kwargs)|
+      connected = false
+      begin
+        TCPSocket.open("127.0.0.1", PORT) do |sock|
+          # expected behavior is connection refused...
+          connected = true
+        end
+      rescue
+      end
+
+      assert_false connected
+
+      received = ""
+      @d.server_create_connection(:s, PORT, proto: proto, shared: false, **kwargs) do |conn|
+        conn.on(:data) do |data|
+          received << data
+          conn.write "ack\n"
+        end
+      end
+
+      th0 = Thread.new do
+        TCPSocket.open("127.0.0.1", PORT) do |sock|
+          sock.puts "yay"
+          sock.readline
+        end
+      end
+
+      value0 = waiting(5){ th0.value }
+      assert_equal "ack\n", value0
+
+      # TODO: change how to create clients by proto
+
+      stopped = false
+      sleeping = false
+      ending = false
+
+      th1 = Thread.new do
+        TCPSocket.open("127.0.0.1", PORT) do |sock|
+          sleeping = true
+          sleep 0.1 until stopped
+          sock.puts "yay"
+          res = sock.readline
+          ending = true
+          res
+        end
+      end
+
+      sleep 0.1 until sleeping
+
+      @d.stop
+      assert @d.stopped?
+      stopped = true
+
+      sleep 0.1 until ending
+
+      @d.before_shutdown
+      @d.shutdown
+
+      th2 = Thread.new do
+        begin
+          TCPSocket.open("127.0.0.1", PORT) do |sock|
+            sock.puts "foo"
+          end
+          false # failed
+        rescue => e
+          true # success
+        end
+      end
+
+      value1 = waiting(5){ th1.value }
+      value2 = waiting(5){ th2.value }
+
+      assert_equal "yay\nyay\n", received
+      assert_equal "ack\n", value1
+      assert value2, "should be truthy value to show connection was correctly refused"
+    end
+
     test 'can keep connections alive for tcp/tls if keepalive specified' do
       # pend "not implemented yet"
     end