feat: generic publishing

Dockerfile

@@ -10,9 +10,9 @@ RUN shards install --production
 COPY src /app/src
 
 # Build App
-RUN crystal build --error-trace --release --debug -o bin/mqtt src/app.cr
+RUN crystal build --error-trace --release --debug -o source src/app.cr
 
 # Run the app binding on port 3000
 EXPOSE 3000
-HEALTHCHECK CMD ["/app/bin/mqtt", "-c", "http://127.0.0.1:3000/api/mqtt/v1/"]
-CMD ["/app/bin/mqtt", "-b", "0.0.0.0", "-p", "3000"]
+HEALTHCHECK CMD ["/app/source", "-c", "http://127.0.0.1:3000/api/source/v1/"]
+CMD ["/app/source", "-b", "0.0.0.0", "-p", "3000"]

README.md

@@ -1,6 +1,8 @@
-# PlaceOS MQTT Service
+# PlaceOS State Source Service
 
-MQTT Client service for PlaceOS
+Service that publishes module state.
+
+Currently MQTT and InfluxDB are supported backends.
 
 [![Build Status](https://travis-ci.com/PlaceOS/mqtt.svg?token=tDypQH5g9ptvYso68jVV&branch=master)](https://travis-ci.com/PlaceOS/mqtt)
 

shard.lock

@@ -6,36 +6,24 @@ shards:
 
   action-controller:
     git: https://github.com/spider-gazelle/action-controller.git
-    version: 3.3.0
+    version: 3.4.0
 
   active-model:
     git: https://github.com/spider-gazelle/active-model.git
-    version: 1.8.4
+    version: 2.0.1
 
   ameba:
     git: https://github.com/crystal-ameba/ameba.git
-    version: 0.13.0
+    version: 0.13.2
 
   bindata:
     git: https://github.com/spider-gazelle/bindata.git
-    version: 1.5.5
+    version: 1.5.6
 
   bisect:
     git: https://github.com/spider-gazelle/bisect.git
     version: 1.2.0
 
-  clustering:
-    git: https://github.com/place-labs/clustering.git
-    version: 2.0.0
-
-  connect-proxy:
-    git: https://github.com/spider-gazelle/connect-proxy.git
-    version: 1.0.0
-
-  crc16:
-    git: https://github.com/maiha/crc16.cr.git
-    version: 0.1.0
-
   cron_parser:
     git: https://github.com/kostya/cron_parser.git
     version: 0.3.0
@@ -44,22 +32,14 @@ shards:
     git: https://github.com/crystal-lang/crystal-db.git
     version: 0.9.0
 
-  debug:
-    git: https://github.com/Sija/debug.cr.git
-    version: 1.1.0
-
-  etcd:
-    git: https://github.com/place-labs/crystal-etcd.git
-    version: 1.0.1
-
-  exec_from:
-    git: https://github.com/place-labs/exec_from.git
-    version: 1.2.0
-
   faker:
     git: https://github.com/askn/faker.git
     version: 0.4.0
 
+  flux:
+    git: https://github.com/place-labs/flux.git
+    version: 1.1.2
+
   future:
     git: https://github.com/crystal-community/future.cr.git
     version: 0.1.0
@@ -68,17 +48,9 @@ shards:
     git: https://github.com/luckyframework/habitat.git
     version: 0.4.4
 
-  hardware:
-    git: https://github.com/caspiano/hardware.git
-    version: 0.6.0
-
-  hound-dog:
-    git: https://github.com/place-labs/hound-dog.git
-    version: 2.5.0
-
-  ipaddress:
-    git: https://github.com/Sija/ipaddress.cr.git
-    version: 0.2.1
+  http-params-serializable:
+    git: https://github.com/vladfaust/http-params-serializable.git
+    version: 0.4.0
 
   json_mapping:
     git: https://github.com/crystal-lang/json_mapping.cr.git
@@ -100,69 +72,41 @@ shards:
     git: https://github.com/spider-gazelle/crystal-mqtt.git
     version: 1.0.1
 
-  murmur3:
-    git: https://github.com/aca-labs/murmur3.git
-    version: 0.1.1+git.commit.7cbe25c0ca8d052c9d98c377c824dcb0e038c790
-
   neuroplastic:
     git: https://github.com/spider-gazelle/neuroplastic.git
-    version: 1.6.0
+    version: 1.6.4
 
   openssl_ext:
     git: https://github.com/stakach/openssl_ext.git
     version: 1.2.2
 
-  placeos-compiler:
-    git: https://github.com/placeos/compiler.git
-    version: 2.0.4
-
-  placeos-core:
-    git: https://github.com/placeos/core.git
-    version: 2.4.0
-
-  placeos-driver:
-    git: https://github.com/placeos/driver.git
-    version: 3.4.3
-
   placeos-models:
     git: https://github.com/placeos/models.git
-    version: 3.0.5
+    version: 4.2.4
+
+  placeos-resource:
+    git: https://github.com/place-labs/resource.git
+    version: 1.0.5
 
   pool:
     git: https://github.com/ysbaddaden/pool.git
     version: 0.2.3
 
-  priority-queue:
-    git: https://github.com/spider-gazelle/priority-queue.git
-    version: 1.0.0+git.commit.bde47d11cb4b4d1ed7680a7f417ff707c434b1e4
-
   promise:
     git: https://github.com/spider-gazelle/promise.git
     version: 2.2.0
 
-  protobuf:
-    git: https://github.com/jeromegn/protobuf.cr.git
-    version: 2.2.1
-
   redis:
     git: https://github.com/stefanwille/crystal-redis.git
     version: 2.6.0
 
-  redis-cluster:
-    git: https://github.com/caspiano/redis-cluster.cr.git
-    version: 0.8.4
-
-  rendezvous-hash:
-    git: https://github.com/caspiano/rendezvous-hash.git
-    version: 0.3.1
-
   rethinkdb:
     git: https://github.com/kingsleyh/crystal-rethinkdb.git
-    version: 0.2.0
+    version: 0.2.1
 
   rethinkdb-orm:
     git: https://github.com/spider-gazelle/rethinkdb-orm.git
-    version: 2.10.1
+    version: 3.0.2
 
   retriable:
     git: https://github.com/sija/retriable.cr.git
@@ -180,10 +124,6 @@ shards:
     git: https://github.com/spider-gazelle/simple_retry.git
     version: 1.1.0
 
-  ssh2:
-    git: https://github.com/spider-gazelle/ssh2.cr.git
-    version: 1.4.2
-
   tasker:
     git: https://github.com/spider-gazelle/tasker.git
     version: 1.3.0
@@ -192,10 +132,6 @@ shards:
     git: https://github.com/spider-gazelle/tokenizer.git
     version: 1.1.0
 
-  ulid:
-    git: https://github.com/place-labs/ulid.git
-    version: 0.1.2
-
   yaml_mapping:
     git: https://github.com/crystal-lang/yaml_mapping.cr.git
     version: 0.1.0

shard.yml

@@ -1,37 +1,44 @@
-name: placeos-mqtt
-version: 0.2.0
+name: placeos-source
+version: 1.0.0
 crystal: 0.35.1
 
 targets:
-  mqtt:
+  source:
     main: src/app.cr
 
 dependencies:
   action-controller:
     github: spider-gazelle/action-controller
 
+  flux:
+    github: place-labs/flux
+
   mqtt:
     github: spider-gazelle/crystal-mqtt
 
-  placeos-core:
-    github: placeos/core
-    version: ~> 2.3
-
   placeos-models:
     github: placeos/models
-    version: ~> 3
+    version: ~> 4
+
+  placeos-resource:
+    github: place-labs/resource
+    version: ~> 1
 
   redis:
     github: stefanwille/crystal-redis
     version: ~> 2.6
 
   rethinkdb-orm:
     github: spider-gazelle/rethinkdb-orm
-    version: ~> 2.10
+    version: ~> 3
 
   rwlock:
     github: spider-gazelle/readers-writer
 
+  secrets-env:
+    github: place-labs/secrets-env
+    version: ~> 1
+
   simple_retry:
     github: spider-gazelle/simple_retry
 

spec/mappings_spec.cr

@@ -1,9 +1,9 @@
 require "./spec_helper"
 
-module PlaceOS::MQTT
+module PlaceOS::Source
   describe Mappings do
-    describe "state_event_keys?" do
-      it "creates a state event topic" do
+    describe "status_events?" do
+      it "generates data for status events" do
         state = mock_state(
           module_id: "mod-1234",
           index: 1,
@@ -17,41 +17,19 @@ module PlaceOS::MQTT
         )
 
         mappings = Mappings.new(state)
-        keys = mappings.state_event_keys?("mod-1234", "power")
-        keys.should_not be_nil
-        keys.not_nil!.first?.should eq "placeos/org-donor/state/cards/nek/2042/cs-9445/12345/M'Odule/1/power"
-      end
-
-      it "doesn't create topics for Modules without a top-level scope Zone" do
-        state = mock_state(module_id: "mod-1234", control_system_id: "cs-id")
-
-        # Remove the top level scope mapping
-        state.system_zones["cs-id"].delete(Mappings.scope)
-
-        mappings = Mappings.new(state)
-        keys = mappings.state_event_keys?("mod-1234", "power")
-        keys.should_not be_nil
-        keys.not_nil!.should be_empty
-      end
-
-      it "replaces missing hierarchy Zone ids with a placeholder" do
-        state = mock_state(
-          module_id: "mod-1234",
-          index: 1,
-          module_name: "M'Odule",
-          driver_id: "12345",
-          control_system_id: "cs-9445",
-          area_id: "2042",
-          level_id: "nek",
-          org_id: "org-donor",
-        )
-
-        state.system_zones["cs-9445"].delete("building")
-
-        mappings = Mappings.new(state)
-        keys = mappings.state_event_keys?("mod-1234", "power")
-        keys.should_not be_nil
-        keys.not_nil!.first?.should eq "placeos/org-donor/state/_/nek/2042/cs-9445/12345/M'Odule/1/power"
+        events = mappings.status_events?("mod-1234", "power")
+        events.should_not be_nil
+        event = events.not_nil!.first
+        event.module_id.should eq "mod-1234"
+        event.index.should eq 1
+        event.module_name.should eq "M'Odule"
+        event.driver_id.should eq "12345"
+        event.control_system_id.should eq "cs-9445"
+
+        event.zone_mapping["area"].should eq "2042"
+        event.zone_mapping["level"].should eq "nek"
+        event.zone_mapping["building"].should eq "cards"
+        event.zone_mapping["org"].should eq "org-donor"
       end
     end
 

spec/publishing/influx_publisher_spec.cr

@@ -0,0 +1,58 @@
+require "../spec_helper"
+require "time"
+
+module PlaceOS::Source
+  describe InfluxPublisher do
+    describe "#transform" do
+      it "ignores PlaceOS Metadata events" do
+        message = Publisher::Message.new(
+          data: Mappings::Metadata.new("some-model-id"),
+          payload: nil,
+        )
+        point = InfluxPublisher.transform(message)
+        point.should be_nil
+      end
+
+      it "transforms a PlaceOS Status event to an InfluxDB point" do
+        state = mock_state(
+          module_id: "mod-1234",
+          index: 1,
+          module_name: "M'Odule",
+          driver_id: "12345",
+          control_system_id: "cs-9445",
+          area_id: "2042",
+          level_id: "nek",
+          building_id: "cards",
+          org_id: "org-donor",
+        )
+
+        status_event = Mappings.new(state).status_events?("mod-1234", "power").not_nil!.first
+
+        message = Publisher::Message.new(status_event, "false")
+
+        point = InfluxPublisher.transform(message)
+        point.should_not be_nil
+        point = point.not_nil!
+
+        point.measurement.should eq "M'Odule"
+
+        point.timestamp.should eq Time::UNIX_EPOCH
+
+        point.tags.should eq({
+          :org      => "org-donor",
+          :building => "cards",
+        })
+
+        point.fields.should eq({
+          :level  => "nek",
+          :area   => "2042",
+          :system => "cs-9445",
+          :driver => "12345",
+          :index  => 1,
+          :state  => "power",
+          :value  => false,
+        })
+      end
+    end
+  end
+end