Fix async persistent connections

With the addition of persistent connections we need to make a separate
InfluxDB::Client per worker as the persistent connection is attached to
the client.

Here this is done by copying the InfluxDB::Config object and creating a
new host queue (so threads don't have to communicate to share hosts) and
creating a new client without async enabled (so it will use HTTP or UDP
write methods).

Author: drbrain

lib/influxdb/client.rb

@@ -49,9 +49,12 @@ class Client
     # +:ssl_ca_cert+:: ssl CA certificate, chainfile or CA path.
     #                  The system CA path is automatically included
     # +:retry+:: number of times a failed request should be retried. Defaults to infinite.
-    def initialize(database = nil, **opts)
+    def initialize(database = nil, config = nil, **opts)
+      raise ArgumentError, "provide config or opts, not both" if
+        config && !opts.empty?
+
       opts[:database] = database if database.is_a? String
-      @config = InfluxDB::Config.new(**opts)
+      @config = config || InfluxDB::Config.new(**opts)
       @stopped = false
       @writer = find_writer
 
@@ -82,7 +85,9 @@ def now
 
     def find_writer
       if config.async?
-        InfluxDB::Writer::Async.new(self, config.async)
+        client = InfluxDB::Client.new nil, config.writer_config
+
+        InfluxDB::Writer::Async.new(client, config.async)
       elsif config.udp.is_a?(Hash)
         InfluxDB::Writer::UDP.new(self, **config.udp)
       elsif config.udp?

lib/influxdb/config.rb

@@ -83,6 +83,12 @@ def initialize(url: nil, **opts)
       configure_hosts! opts[:hosts] || opts[:host] || "localhost".freeze
     end
 
+    def initialize_copy(source)
+      super
+
+      configure_hosts! source.hosts
+    end
+
     def udp?
       udp != false
     end
@@ -105,7 +111,15 @@ def hosts
       end
     end
 
-    private
+    def writer_config
+      writer_config = dup
+
+      writer_config.set_ivar! :async, false
+
+      writer_config
+    end
+
+    protected
 
     def set_ivar!(name, value)
       case name
@@ -118,6 +132,8 @@ def set_ivar!(name, value)
       instance_variable_set "@#{name}", value
     end
 
+    private
+
     def normalize_retry_option(value)
       case value
       when Integer   then value

spec/influxdb/cases/async_client_spec.rb

@@ -6,7 +6,8 @@
   let(:client) { described_class.new(async: async_options) }
   let(:subject) { client }
   let(:stub_url) { "http://localhost:8086/write?db=&p=root&precision=s&u=root" }
-  let(:worker) { client.writer.worker }
+  let(:writer) { client.writer }
+  let(:worker) { writer.worker }
 
   specify { expect(subject.writer).to be_a(InfluxDB::Writer::Async) }
 
@@ -40,7 +41,7 @@
 
       it "writes aggregate payload to the client" do
         queue = Queue.new
-        allow(client).to receive(:write) do |*args|
+        allow_any_instance_of(InfluxDB::Client).to receive(:write) do |_, *args|
           queue.push(args)
         end
 
@@ -61,7 +62,7 @@
 
         it "writes separated payloads for each {precision, retention_policy, database} set" do
           queue = Queue.new
-          allow(client).to receive(:write) do |*args|
+          allow_any_instance_of(InfluxDB::Client).to receive(:write) do |_, *args|
             queue.push(args)
           end