Merge pull request #338 from fablabbcn/forwarding-in-job

Forwarding in job

app/controllers/v0/readings_controller.rb

@@ -38,14 +38,12 @@ def create
     def legacy_create
 
       if request.headers['X-SmartCitizenData']
-        MQTTClientFactory.create_client({clean_session: true, client_id: nil}) do |mqtt_client|
-          storer = RawStorer.new(mqtt_client)
-          JSON.parse(request.headers['X-SmartCitizenData']).each do |raw_reading|
-            mac = request.headers['X-SmartCitizenMacADDR']
-            version = request.headers['X-SmartCitizenVersion']
-            ip = (request.headers['X-SmartCitizenIP'] || request.remote_ip)
-            storer.store(raw_reading,mac,version,ip)
-          end
+        storer = RawStorer.new
+        JSON.parse(request.headers['X-SmartCitizenData']).each do |raw_reading|
+          mac = request.headers['X-SmartCitizenMacADDR']
+          version = request.headers['X-SmartCitizenVersion']
+          ip = (request.headers['X-SmartCitizenIP'] || request.remote_ip)
+          storer.store(raw_reading,mac,version,ip)
         end
       end
 

app/jobs/mqtt_forwarding_job.rb

@@ -0,0 +1,43 @@
+class MQTTForwardingJob < ApplicationJob
+
+  queue_as :mqtt_forward
+
+  def perform(device_id, reading)
+    begin
+      device = Device.find(device_id)
+      forwarder = MQTTForwarder.new(mqtt_client)
+      payload = payload_for(device, reading)
+      forwarder.forward_reading(device.forwarding_token, device.id, payload)
+    ensure
+      disconnect_mqtt!
+    end
+  end
+
+  private
+
+  def payload_for(device, reading)
+    renderer.render(
+      partial: "v0/devices/device",
+      locals: {
+        device: device.reload,
+        current_user: nil,
+        slim_owner: true
+      }
+    )
+  end
+
+  def mqtt_client
+    @mqtt_client ||= MQTTClientFactory.create_client({
+      clean_session: true, client_id: nil
+    })
+  end
+
+  def renderer
+    @renderer ||= ActionController::Base.new.view_context
+  end
+
+  def disconnect_mqtt!
+    @mqtt_client&.disconnect
+  end
+end
+

app/jobs/retry_mqtt_message_job.rb

@@ -19,26 +19,14 @@ class RetryMessageHandlerError < RuntimeError
   end
 
   def perform(topic, message)
-    begin
-      result = handler.handle_topic(topic, message, false)
-      raise RetryMessageHandlerError if result.nil?
-    ensure
-      disconnect_mqtt
-    end
+    result = handler.handle_topic(topic, message, false)
+    raise RetryMessageHandlerError if result.nil?
   end
 
   private
 
   def handler
-    @handler ||= MqttMessagesHandler.new(mqtt_client)
-  end
-
-  def mqtt_client
-    @mqtt_client ||= MQTTClientFactory.create_client({clean_session: true, client_id: nil })
-  end
-
-  def disconnect_mqtt
-    @mqtt_client&.disconnect
+    @handler ||= MqttMessagesHandler.new
   end
 end
 

app/jobs/send_to_datastore_job.rb

@@ -2,28 +2,16 @@ class SendToDatastoreJob < ApplicationJob
   queue_as :default
 
   def perform(data_param, device_id)
-    begin
-      @device = Device.includes(:components).find(device_id)
-      the_data = JSON.parse(data_param)
-      the_data.sort_by {|a| a['recorded_at']}.reverse.each_with_index do |reading, index|
-        # move to async method call
-        do_update = index == 0
-        storer.store(@device, reading, do_update)
-      end
-    ensure
-      disconnect_mqtt
+    @device = Device.includes(:components).find(device_id)
+    the_data = JSON.parse(data_param)
+    the_data.sort_by {|a| a['recorded_at']}.reverse.each_with_index do |reading, index|
+      # move to async method call
+      do_update = index == 0
+      storer.store(@device, reading, do_update)
     end
   end
 
   def storer
-    @storer ||= Storer.new(mqtt_client)
-  end
-
-  def mqtt_client
-    @mqtt_client ||= MQTTClientFactory.create_client({clean_session: true, client_id: nil })
-  end
-
-  def disconnect_mqtt
-    @mqtt_client&.disconnect
+    @storer ||= Storer.new
   end
 end

app/lib/mqtt_messages_handler.rb

@@ -1,9 +1,5 @@
 class MqttMessagesHandler
 
-  def initialize(mqtt_client)
-    @mqtt_client = mqtt_client
-  end
-
   def handle_topic(topic, message, retry_on_nil_device=true)
     Sentry.set_tags('mqtt-topic': topic)
 
@@ -138,10 +134,7 @@ def data(message)
 
   private
 
-  attr_reader :mqtt_client
-
-
   def storer
-    @storer ||= Storer.new(mqtt_client)
+    @storer ||= Storer.new
   end
 end

app/models/concerns/message_forwarding.rb

@@ -4,30 +4,8 @@ module MessageForwarding
 
   def forward_reading(device, reading)
     if device.forward_readings?
-      forwarder = MQTTForwarder.new(mqtt_client)
-      payload = payload_for(device, reading)
-      forwarder.forward_reading(device.forwarding_token, device.id, payload)
+      MQTTForwardingJob.perform_later(device.id, reading)
     end
   end
 
-  def payload_for(device, reading)
-    renderer.render(
-      partial: "v0/devices/device",
-      locals: {
-        device: device.reload,
-        current_user: nil,
-        slim_owner: true
-      }
-    )
-  end
-
-  private
-
-  def mqtt_client
-    raise NotImplementedError
-  end
-
-  def renderer
-    raise NotImplementedError
-  end
 end

app/models/raw_storer.rb

@@ -5,11 +5,6 @@
 class RawStorer
   include MessageForwarding
 
-  def initialize(mqtt_client, renderer=nil)
-    @mqtt_client = mqtt_client
-    @renderer = renderer || ActionController::Base.new.view_context
-  end
-
   def store data, mac, version, ip, raise_errors=false
     success = true
 
@@ -81,9 +76,4 @@ def store data, mac, version, ip, raise_errors=false
       end
     end
   end
-
-  private
-
-  attr_reader :mqtt_client, :renderer
-
 end

app/models/storer.rb

@@ -2,11 +2,6 @@ class Storer
   include DataParser::Storer
   include MessageForwarding
 
-  def initialize(mqtt_client, renderer=nil)
-    @mqtt_client = mqtt_client
-    @renderer = renderer || ActionController::Base.new.view_context
-  end
-
   def store device, reading, do_update = true
     begin
       parsed_reading = Storer.parse_reading(device, reading)
@@ -47,9 +42,4 @@ def kairos_publish(reading_data)
     #NOTE: If you want to use the Telnet port below, make sure it is open!
     Redis.current.publish('telnet_queue', reading_data.to_json)
   end
-
-  private
-
-  attr_reader :mqtt_client, :renderer
-
 end

config/sidekiq.yml

@@ -5,6 +5,7 @@
   - default
   - mailers
   - mqtt_retry
+  - mqtt_forward
 
 production:
   :concurrency: 25

lib/tasks/mqtt_subscriber.rake

@@ -31,7 +31,7 @@ namespace :mqtt do
         mqtt_log.info "Connected to #{client.host}"
         mqtt_log.info "Using clean_session setting: #{client.clean_session}"
 
-        message_handler = MqttMessagesHandler.new(client)
+        message_handler = MqttMessagesHandler.new
 
         client.subscribe(*mqtt_topics.flat_map { |topic|
           topic = topic == "" ? topic : topic + "/"

spec/jobs/mqtt_forwarding_job_spec.rb

@@ -0,0 +1,75 @@
+require 'rails_helper'
+
+RSpec.describe MQTTForwardingJob, type: :job do
+
+  let(:forwarding_token) { "abc123_forwarding_token" }
+
+  let(:device) { create(:device) }
+
+  let(:reading) { double(:reading) }
+
+  let(:mqtt_client) {
+    double(:mqtt_client).tap do |mqtt_client|
+      allow(mqtt_client).to receive(:disconnect)
+    end
+  }
+
+  let(:device_json) {
+    double(:device_json)
+  }
+
+  let(:renderer) {
+    double(:renderer).tap do |renderer|
+      allow(renderer).to receive(:render).and_return(device_json)
+    end
+  }
+
+  let(:forwarder) {
+    double(:forwarder).tap do |forwarder|
+      allow(forwarder).to receive(:forward_reading)
+    end
+  }
+
+  before do
+    allow(MQTTClientFactory).to receive(:create_client).and_return(mqtt_client)
+    allow_any_instance_of(ActionController::Base).to receive(:view_context).and_return(renderer)
+    allow(MQTTForwarder).to receive(:new).and_return(forwarder)
+    allow_any_instance_of(Device).to receive(:forwarding_token).and_return(forwarding_token)
+  end
+
+  it "creates an mqtt client with a clean session and no client id" do
+    MQTTForwardingJob.perform_now(device.id, reading)
+    expect(MQTTClientFactory).to have_received(:create_client).with({
+      clean_session: true,
+      client_id: nil
+    })
+  end
+
+  it "creates a forwarder with the mqtt client" do
+    MQTTForwardingJob.perform_now(device.id, reading)
+    expect(MQTTForwarder).to have_received(:new).with(mqtt_client)
+  end
+
+  it "renders the device json for the given device, as an unauthorized user" do
+    MQTTForwardingJob.perform_now(device.id, reading)
+    expect(renderer).to have_received(:render).with({
+      partial: "v0/devices/device",
+      locals: {
+        device: device.reload,
+        current_user: nil,
+        slim_owner: true
+      }
+    })
+  end
+
+  it "forwards using the device's id and forwarding token, with the rendered json payload" do
+    MQTTForwardingJob.perform_now(device.id, reading)
+    expect(forwarder).to have_received(:forward_reading).with(forwarding_token, device.id, device_json)
+  end
+
+  it "disconnects the MQTT client" do
+    MQTTForwardingJob.perform_now(device.id, reading)
+    expect(mqtt_client).to have_received(:disconnect)
+  end
+
+end

spec/jobs/retry_mqtt_message_job_spec.rb

@@ -3,12 +3,6 @@
 RSpec.describe RetryMQTTMessageJob, type: :job do
   include ActiveJob::TestHelper
 
-  let(:mqtt_client) {
-    double(:mqtt_client).tap do |mqtt_client|
-      allow(mqtt_client).to receive(:disconnect)
-    end
-  }
-
   let(:handler_results) { [true] }
 
   let(:mqtt_message_handler) {
@@ -22,30 +16,19 @@
   let(:message) { '{"foo": "bar", "test": "message"}' }
 
   before do
-    allow(MQTTClientFactory).to receive(:create_client).and_return(mqtt_client)
     allow(MqttMessagesHandler).to receive(:new).and_return(mqtt_message_handler)
   end
 
-  it "creates an MQTT client, overriding clean_session to true and the client_id to nil" do
-    RetryMQTTMessageJob.perform_now(topic, message)
-    expect(MQTTClientFactory).to have_received(:create_client).with({clean_session: true, client_id: nil })
-  end
-
-  it "creates an MQTTMessagesHandler, passing the client" do
+  it "creates an MQTTMessagesHandler" do
     RetryMQTTMessageJob.perform_now(topic, message)
-    expect(MqttMessagesHandler).to have_received(:new).with(mqtt_client)
+    expect(MqttMessagesHandler).to have_received(:new)
   end
 
   it "retries the mqtt ingest with the given topic and message, and with automatic retries disabled" do
     RetryMQTTMessageJob.perform_now(topic, message)
     expect(mqtt_message_handler).to have_received(:handle_topic).with(topic, message, false)
   end
 
-  it "disconnects the client when done" do
-    RetryMQTTMessageJob.perform_now(topic, message)
-    expect(mqtt_client).to have_received(:disconnect)
-  end
-
   context "when the handler returns nil" do
     let(:handler_results) { [nil, nil, true] }
 
@@ -54,10 +37,5 @@
         RetryMQTTMessageJob.perform_later(topic, message)
       end
     end
-
-    it "closes the mqtt connection" do
-      RetryMQTTMessageJob.perform_now(topic, message)
-      expect(mqtt_client).to have_received(:disconnect)
-    end
   end
 end

spec/lib/mqtt_messages_handler_spec.rb

@@ -8,14 +8,8 @@
   let(:device_inventory) { create(:device_inventory, report: '{"random_property": "random_result"}') }
 
 
-  let(:mqtt_client) {
-    double(:mqtt_client).tap do |mqtt_client|
-      allow(mqtt_client).to receive(:publish)
-    end
-  }
-
   subject(:message_handler) {
-    MqttMessagesHandler.new(mqtt_client)
+    MqttMessagesHandler.new
   }