Merge branch 'release/0.7.0'

Author: ledermann

.env.test

@@ -3,9 +3,9 @@ SENEC_HOST=10.0.1.123
 SENEC_INTERVAL=5
 
 # Example of an InfluxDB host
-INFLUX_HOST=influx.example.com
-INFLUX_SCHEMA=https
-INFLUX_PORT=443
-INFLUX_TOKEN=this.is.just.an.example
-INFLUX_ORG=solectrus
-INFLUX_BUCKET=SENEC
+INFLUX_HOST=localhost
+INFLUX_SCHEMA=http
+INFLUX_PORT=8086
+INFLUX_TOKEN=my-token
+INFLUX_ORG=my-org
+INFLUX_BUCKET=my-bucket

.rubocop.yml

@@ -1,6 +1,7 @@
 require:
   - rubocop-minitest
   - rubocop-rake
+  - rubocop-performance
 
 AllCops:
   TargetRubyVersion: 3.2
@@ -16,7 +17,7 @@ Style/FrozenStringLiteralComment:
   Enabled: false
 
 Metrics/MethodLength:
-  Max: 15
+  Max: 20
 
 Minitest/MultipleAssertions:
   Max: 10

.ruby-version

@@ -1 +1 @@
-3.2.0
+3.2.1

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+  "ruby.lint": {
+    "rubocop": {
+      "forceExclusion": true
+    }
+  }
+}

Dockerfile

@@ -1,4 +1,4 @@
-FROM ruby:3.2.0-alpine AS Builder
+FROM ruby:3.2.1-alpine AS Builder
 RUN apk add --no-cache build-base
 
 WORKDIR /senec-collector
@@ -8,7 +8,7 @@ RUN bundle config --local frozen 1 && \
     bundle install -j4 --retry 3 && \
     bundle clean --force
 
-FROM ruby:3.2.0-alpine
+FROM ruby:3.2.1-alpine
 LABEL maintainer="georg@ledermann.dev"
 
 # Decrease memory usage

Gemfile

@@ -28,6 +28,9 @@ group :development, :test do
   # A RuboCop plugin for Rake (https://github.com/rubocop/rubocop-rake)
   gem 'rubocop-rake'
 
+  # Automatic performance checking tool for Ruby code. (https://github.com/rubocop/rubocop-performance)
+  gem 'rubocop-performance'
+
   # Record your test suite's HTTP interactions and replay them during future test runs for fast, deterministic, accurate tests. (https://relishapp.com/vcr/vcr/docs)
   gem 'vcr'
 

Gemfile.lock

@@ -6,59 +6,47 @@ GEM
     ast (2.4.2)
     crack (0.4.5)
       rexml
-    domain_name (0.5.20190701)
-      unf (>= 0.0.5, < 1.0.0)
     dotenv (2.8.1)
-    ffi (1.15.5)
-    ffi-compiler (1.0.1)
-      ffi (>= 1.0.0)
-      rake
     hashdiff (1.0.1)
-    http (5.1.1)
-      addressable (~> 2.8)
-      http-cookie (~> 1.0)
-      http-form_data (~> 2.2)
-      llhttp-ffi (~> 0.4.0)
-    http-cookie (1.0.5)
-      domain_name (~> 0.5)
-    http-form_data (2.3.0)
+    httparty (0.21.0)
+      mini_mime (>= 1.0.0)
+      multi_xml (>= 0.5.2)
     influxdb-client (2.9.0)
     json (2.6.3)
-    llhttp-ffi (0.4.0)
-      ffi-compiler (~> 1.0)
-      rake (~> 13.0)
-    minitest (5.17.0)
-    oj (3.14.1)
+    mini_mime (1.1.2)
+    minitest (5.18.0)
+    multi_xml (0.6.0)
+    oj (3.14.2)
     parallel (1.22.1)
-    parser (3.2.0.0)
+    parser (3.2.1.1)
       ast (~> 2.4.1)
     public_suffix (5.0.1)
     rainbow (3.1.1)
     rake (13.0.6)
-    regexp_parser (2.6.2)
+    regexp_parser (2.7.0)
     rexml (3.2.5)
-    rubocop (1.44.1)
+    rubocop (1.48.1)
       json (~> 2.3)
       parallel (~> 1.10)
       parser (>= 3.2.0.0)
       rainbow (>= 2.2.2, < 4.0)
       regexp_parser (>= 1.8, < 3.0)
       rexml (>= 3.2.5, < 4.0)
-      rubocop-ast (>= 1.24.1, < 2.0)
+      rubocop-ast (>= 1.26.0, < 2.0)
       ruby-progressbar (~> 1.7)
       unicode-display_width (>= 2.4.0, < 3.0)
-    rubocop-ast (1.24.1)
-      parser (>= 3.1.1.0)
-    rubocop-minitest (0.27.0)
-      rubocop (>= 0.90, < 2.0)
+    rubocop-ast (1.27.0)
+      parser (>= 3.2.1.0)
+    rubocop-minitest (0.29.0)
+      rubocop (>= 1.39, < 2.0)
+    rubocop-performance (1.16.0)
+      rubocop (>= 1.7.0, < 2.0)
+      rubocop-ast (>= 0.4.0)
     rubocop-rake (0.6.0)
       rubocop (~> 1.0)
-    ruby-progressbar (1.11.0)
-    senec (0.7.0)
-      http (~> 5)
-    unf (0.1.4)
-      unf_ext
-    unf_ext (0.0.8.2)
+    ruby-progressbar (1.13.0)
+    senec (0.7.1)
+      httparty
     unicode-display_width (2.4.2)
     vcr (6.1.0)
     webmock (3.18.1)
@@ -77,10 +65,11 @@ DEPENDENCIES
   rake
   rubocop
   rubocop-minitest
+  rubocop-performance
   rubocop-rake
   senec
   vcr
   webmock
 
 BUNDLED WITH
-   2.4.6
+   2.4.9

src/flux_writer.rb

@@ -7,10 +7,6 @@ def initialize(config)
 
   attr_reader :config
 
-  def self.push(config:, record:)
-    new(config).push(record)
-  end
-
   def push(record)
     return unless record
 

src/influx_push.rb

@@ -2,42 +2,44 @@
 
 class InfluxPush
   def initialize(config:, queue:)
-    @queue = queue
     @config = config
-    @count = 0
+    @queue = queue
+    @flux_writer = FluxWriter.new(config)
   end
 
-  attr_reader :config, :queue
+  attr_reader :config, :queue, :flux_writer
 
   def run
-    until queue.empty?
+    until queue.closed?
+      # Wait for a record to be added to the queue
       record = queue.pop
 
+      # Stop if the queue has been closed
+      next unless record
+
       begin
-        FluxWriter.push(config:, record:)
-        @count += 1
-      rescue StandardError
-        # Put the record back into the queue
-        queue << record
+        flux_writer.push(record)
+        puts "Successfully pushed record ##{record.id} to InfluxDB"
+      rescue StandardError => e
+        error_handling(record, e)
 
-        raise
+        # Wait a bit before trying again
+        sleep(5)
       end
     end
   end
 
-  def success_message
-    return unless @count.positive?
+  private
 
-    "Successfully pushed #{pluralize(@count, 'record', 'records')} to InfluxDB"
-  end
+  def error_handling(record, error)
+    # Log the error
+    puts "Error while pushing record ##{record.id} to InfluxDB: #{error.message}"
 
-  def failure_message(error)
-    "Error while pushing #{pluralize(queue.length, 'record', 'records')} to InfluxDB. #{error.class}: #{error.message}"
-  end
+    return if queue.closed?
 
-  private
+    # Put the record back into the queue
+    queue << record
 
-  def pluralize(count, singular, plural)
-    count == 1 ? singular : "#{count} #{plural}"
+    puts "The record has been queued. Will retry to push #{queue.size} records later."
   end
 end

src/loop.rb

@@ -12,52 +12,66 @@ def initialize(config:, max_count:)
   end
 
   attr_reader :config, :max_count
-  attr_accessor :queue, :count, :push_thread
+  attr_accessor :queue
 
   def start
     self.queue = Queue.new
-    self.count = 0
 
-    loop do
-      self.count += 1
+    pull_thread = Thread.new { pull_loop }
+    push_thread = Thread.new { push_loop }
 
-      pull_from_senec
-      push_to_influx
+    # Wait for the pull thread to finish (will happen if max_count is set)
+    pull_thread.join
 
-      break if max_count && count >= max_count
+    # Push any remaining records to InfluxDB
+    close_queue
 
-      sleep config.senec_interval
-    end
+    # Wait for the push thread to finish (will happen because queue is closed)
+    push_thread.join
+  rescue SystemExit, Interrupt
+    puts 'Exiting...'
+
+    # Stop pulling data from SENEC
+    pull_thread.exit
+
+    # Push any remaining records to InfluxDB (can take a while)
+    close_queue
 
-    # Wait for the push thread (if there is one) to finish
-    push_thread&.join
+    # Stop pushing data to InfluxDB
+    push_thread.exit
   end
 
   private
 
   # Pull data from SENEC and add to queue
-  def pull_from_senec
+  def pull_loop
     pull = SenecPull.new(config:, queue:)
-    pull.run
-    puts pull.success_message(count)
-  rescue StandardError => e
-    puts pull.failure_message(e)
-  end
 
-  # Push data from queue to InfluxDB
-  # Do this in a separate thread so that the main thread is not blocked
-  def push_to_influx
-    # Do nothing if there is already a push thread running
-    return if push_thread&.status
-
-    # Create new thread and push to InfluxDB
-    self.push_thread =
-      Thread.new do
-        push = InfluxPush.new(config:, queue:)
-        push.run
-        puts push.success_message
+    loop do
+      begin
+        pull.next
+        puts pull.success_message
       rescue StandardError => e
-        puts push.failure_message(e)
+        puts pull.failure_message(e)
       end
+
+      break if max_count && pull.count >= max_count
+
+      sleep config.senec_interval
+    end
+  end
+
+  # Push data from queue to InfluxDB
+  def push_loop
+    InfluxPush.new(config:, queue:).run
+  end
+
+  def close_queue
+    until queue.empty?
+      puts "Waiting for #{queue.size} records to be pushed to InfluxDB"
+      sleep 1
+    end
+
+    queue.close
   end
 end