Merge pull request #4619 from Watson1978/async-2.x

http_server: Ready to support Async 2.0 gem

fluentd.gemspec

@@ -63,7 +63,6 @@ Gem::Specification.new do |gem|
   gem.add_development_dependency("test-unit", ["~> 3.3"])
   gem.add_development_dependency("test-unit-rr", ["~> 1.0"])
   gem.add_development_dependency("oj", [">= 2.14", "< 4"])
-  gem.add_development_dependency("async", "~> 1.23")
   gem.add_development_dependency("async-http", ">= 0.50.0")
   gem.add_development_dependency("aws-sigv4", ["~> 1.8"])
   gem.add_development_dependency("aws-sdk-core", ["~> 3.191"])

lib/fluent/plugin_helper/http_server/server.rb

@@ -39,7 +39,8 @@ def initialize(addr:, port:, logger:, default_app: nil, tls_context: nil)
           scheme = tls_context ? 'https' : 'http'
           @uri = URI("#{scheme}://#{@addr}:#{@port}").to_s
           @router = Router.new(default_app)
-          @reactor = Async::Reactor.new(nil, logger: Fluent::Log::ConsoleAdapter.wrap(@logger))
+          @server_task = nil
+          Console.logger = Fluent::Log::ConsoleAdapter.wrap(@logger)
 
           opts = if tls_context
                    { ssl_context: tls_context }
@@ -54,24 +55,35 @@ def initialize(addr:, port:, logger:, default_app: nil, tls_context: nil)
         end
 
         def start(notify = nil)
+          Console.logger = Fluent::Log::ConsoleAdapter.wrap(@logger)
           @logger.debug("Start async HTTP server listening #{@uri}")
-          task = @reactor.run do
-            @server.run
 
+          Async do |task|
+            Console.logger = Fluent::Log::ConsoleAdapter.wrap(@logger)
+            @server_task = task.async do
+              Console.logger = Fluent::Log::ConsoleAdapter.wrap(@logger)
+              @server.run
+            end
             if notify
               notify.push(:ready)
             end
+
+            if async_v2?
+              @server_task_queue = ::Thread::Queue.new
+              @server_task_queue.pop
+              @server_task&.stop
+            end
           end
 
-          task.stop
           @logger.debug('Finished HTTP server')
         end
 
         def stop
           @logger.debug('closing HTTP server')
-
-          if @reactor
-            @reactor.stop
+          if async_v2?
+            @server_task_queue&.push(:stop)
+          else
+            @server_task&.stop
           end
         end
 
@@ -88,6 +100,10 @@ def stop
             @router.mount(name, path, app || block)
           end
         end
+
+        private def async_v2?
+          Gem::Version.new(Async::VERSION) >= Gem::Version.new('2.0')
+        end
       end
     end
   end

test/plugin_helper/test_http_server_helper.rb

@@ -127,20 +127,21 @@ def start_https_request(addr, port, verify: true, cert_path: nil, selfsigned: tr
     end
 
     client = Async::HTTP::Client.new(Async::HTTP::Endpoint.parse("https://#{addr}:#{port}", ssl_context: context))
-    reactor = Async::Reactor.new(nil, logger: Fluent::Log::ConsoleAdapter.wrap(NULL_LOGGER))
+    Console.logger = Fluent::Log::ConsoleAdapter.wrap(NULL_LOGGER)
 
     resp = nil
     error = nil
 
-    reactor.run do
+    Sync do
+      Console.logger = Fluent::Log::ConsoleAdapter.wrap(NULL_LOGGER)
       begin
         response = yield(client)
       rescue => e               # Async::Reactor rescue all error. handle it by myself
         error = e
       end
 
       if response
-        resp = Response.new(response.status.to_s, response.body.read, response.headers)
+        resp = Response.new(response.status.to_s, response.read, response.headers)
       end
     end