lib/embulk/input/zendesk/client.rb in embulk-input-zendesk-0.1.7 vs lib/embulk/input/zendesk/client.rb in embulk-input-zendesk-0.1.8

@@ -18,10 +18,11 @@
           @config = config
         end
 
         def httpclient
           httpclient = HTTPClient.new
+          httpclient.connect_timeout = 240 # default:60 is not enough for huge data
           # httpclient.debug_dev = STDOUT
           return set_auth(httpclient)
         end
 
         def validate_config
@@ -160,38 +161,42 @@
         end
 
         def incremental_export(path, key, start_time = 0, known_ids = [], partial = true, &block)
           if partial
             records = request_partial(path, {start_time: start_time}).first(5)
-          else
+            records.uniq{|r| r["id"]}.each do |record|
+              block.call record
+            end
+            return
+          end
+
+          loop do
+            start_fetching = Time.now
             response = request(path, {start_time: start_time})
             begin
               data = JSON.parse(response.body)
             rescue => e
               raise Embulk::DataError.new(e)
             end
-            Embulk.logger.info "Fetched records from #{start_time} (#{Time.at(start_time)})"
+            actual_fetched = 0
             records = data[key]
-          end
+            records.each do |record|
+              # de-duplicated records.
+              # https://developer.zendesk.com/rest_api/docs/core/incremental_export#usage-notes
+              # https://github.com/zendesk/zendesk_api_client_rb/issues/251
+              next if known_ids.include?(record["id"])
 
-          records.each do |record|
-            # de-duplicated records.
-            # https://developer.zendesk.com/rest_api/docs/core/incremental_export#usage-notes
-            # https://github.com/zendesk/zendesk_api_client_rb/issues/251
-            next if known_ids.include?(record["id"])
+              known_ids << record["id"]
+              block.call record
+              actual_fetched += 1
+            end
+            Embulk.logger.info "Fetched #{actual_fetched} records from start_time:#{start_time} (#{Time.at(start_time)}) within #{Time.now.to_i - start_fetching.to_i} seconds"
+            start_time = data["end_time"]
 
-            known_ids << record["id"]
-            block.call record
-          end
-          return if partial
-
-          # NOTE: If count is less than 1000, then stop paginating.
-          #       Otherwise, use the next_page URL to get the next page of results.
-          #       https://developer.zendesk.com/rest_api/docs/core/incremental_export#pagination
-          if data["count"] == 1000
-            incremental_export(path, key, data["end_time"], known_ids, partial, &block)
-          else
-            data
+            # NOTE: If count is less than 1000, then stop paginating.
+            #       Otherwise, use the next_page URL to get the next page of results.
+            #       https://developer.zendesk.com/rest_api/docs/core/incremental_export#pagination
+            break data if data["count"] < 1000
           end
         end
 
         def extract_records_from_response(response, key)
           begin