Integrate with EMQX

This page introduces how to use the Neuron southbound driver to collect data, then report the data to EMQX through the northbound Sparkplug B plug-in, and the correct and complete data results are obtained after decoding through the codec function. The process is shown in the figure:

Sparkplug B

Configure Neuron

Southbound Device

Collect the Modbus TCP simulator point value through the southbound drive to simulate the actual device point value, the configuration is as follows：

Add device

image-20230421141232191

Device Configuration

image-20230421141356669

image-20230421141441471

Create group

image-20230421141538315

Create Tag

image-20230421141639201

Northbound application

Add application

image-20230421141821812

Application configuration

image-20230421141912355

image-20230421142020855

Add subscription

image-20230421142109283

Configure EMQX

You can use MQTTX to view the forwarded messages to EMQX from Neuron. However, if you directly subscribe to the EMQX data reported by Neuron northbound to SparkPlugB, there will be garbled strings, as shown in the figure:

image-20230421151918685

Therefore, through the encoding and decoding capabilities of the EMQX rule engine, a corresponding proto file can be written and combined with the rule engine to decode the reported data and obtain correct and complete data results.

Create Schema Registry

image-20230421142801332

image-20230421142920299

  1. // Complete proto file
  2. syntax = "proto2";
  4. //
  5. // To compile:
  6. // cd client_libraries/java
  7. // protoc --proto_path=../../ --java_out=src/main/java ../../sparkplug_b.proto 
  8. //
  10. message Payload {
  11.     /*
  12.         // Indexes of Data Types
  13.         // Unknown placeholder for future expansion.
  14.         Unknown         = 0;
  15.         // Basic Types
  16.         Int8            = 1;
  17.         Int16           = 2;
  18.         Int32           = 3;
  19.         Int64           = 4;
  20.         UInt8           = 5;
  21.         UInt16          = 6;
  22.         UInt32          = 7;
  23.         UInt64          = 8;
  24.         Float           = 9;
  25.         Double          = 10;
  26.         Boolean         = 11;
  27.         String          = 12;
  28.         DateTime        = 13;
  29.         Text            = 14;
  30.         // Additional Metric Types
  31.         UUID            = 15;
  32.         DataSet         = 16;
  33.         Bytes           = 17;
  34.         File            = 18;
  35.         Template        = 19;
  37.         // Additional PropertyValue Types
  38.         PropertySet     = 20;
  39.         PropertySetList = 21;
  40.     */
  42.     message Template {
  44.         message Parameter {
  45.             optional string name        = 1;
  46.             optional uint32 type        = 2;
  48.             oneof value {
  49.                 uint32 int_value        = 3;
  50.                 uint64 long_value       = 4;
  51.                 float  float_value      = 5;
  52.                 double double_value     = 6;
  53.                 bool   boolean_value    = 7;
  54.                 string string_value     = 8;
  55.                 ParameterValueExtension extension_value = 9;
  56.             }
  58.             message ParameterValueExtension {
  59.                 extensions              1 to max;
  60.             }
  61.         }
  63.         optional string version         = 1;          // The version of the Template to prevent mismatches
  64.         repeated Metric metrics         = 2;          // Each metric is the name of the metric and the datatype of the member but does not contain a value
  65.         repeated Parameter parameters   = 3;
  66.         optional string template_ref    = 4;          // Reference to a template if this is extending a Template or an instance - must exist if an instance
  67.         optional bool is_definition     = 5;
  68.         extensions                      6 to max;
  69.     }
  71.     message DataSet {
  73.         message DataSetValue {
  75.             oneof value {
  76.                 uint32 int_value                        = 1;
  77.                 uint64 long_value                       = 2;
  78.                 float  float_value                      = 3;
  79.                 double double_value                     = 4;
  80.                 bool   boolean_value                    = 5;
  81.                 string string_value                     = 6;
  82.                 DataSetValueExtension extension_value   = 7;
  83.             }
  85.             message DataSetValueExtension {
  86.                 extensions  1 to max;
  87.             }
  88.         }
  90.         message Row {
  91.             repeated DataSetValue elements  = 1;
  92.             extensions                      2 to max;   // For third party extensions
  93.         }
  95.         optional uint64   num_of_columns    = 1;
  96.         repeated string   columns           = 2;
  97.         repeated uint32   types             = 3;
  98.         repeated Row      rows              = 4;
  99.         extensions                          5 to max;   // For third party extensions
  100.     }
  102.     message PropertyValue {
  104.         optional uint32     type                    = 1;
  105.         optional bool       is_null                 = 2; 
  107.         oneof value {
  108.             uint32          int_value               = 3;
  109.             uint64          long_value              = 4;
  110.             float           float_value             = 5;
  111.             double          double_value            = 6;
  112.             bool            boolean_value           = 7;
  113.             string          string_value            = 8;
  114.             PropertySet     propertyset_value       = 9;
  115.             PropertySetList propertysets_value      = 10;      // List of Property Values
  116.             PropertyValueExtension extension_value  = 11;
  117.         }
  119.         message PropertyValueExtension {
  120.             extensions                             1 to max;
  121.         }
  122.     }
  124.     message PropertySet {
  125.         repeated string        keys     = 1;         // Names of the properties
  126.         repeated PropertyValue values   = 2;
  127.         extensions                      3 to max;
  128.     }
  130.     message PropertySetList {
  131.         repeated PropertySet propertyset = 1;
  132.         extensions                       2 to max;
  133.     }
  135.     message MetaData {
  136.         // Bytes specific metadata
  137.         optional bool   is_multi_part   = 1;
  139.         // General metadata
  140.         optional string content_type    = 2;        // Content/Media type
  141.         optional uint64 size            = 3;        // File size, String size, Multi-part size, etc
  142.         optional uint64 seq             = 4;        // Sequence number for multi-part messages
  144.         // File metadata
  145.         optional string file_name       = 5;        // File name
  146.         optional string file_type       = 6;        // File type (i.e. xml, json, txt, cpp, etc)
  147.         optional string md5             = 7;        // md5 of data
  149.         // Catchalls and future expansion
  150.         optional string description     = 8;        // Could be anything such as json or xml of custom properties
  151.         extensions                      9 to max;
  152.     }
  154.     message Metric {
  156.         optional string   name          = 1;        // Metric name - should only be included on birth
  157.         optional uint64   alias         = 2;        // Metric alias - tied to name on birth and included in all later DATA messages
  158.         optional uint64   timestamp     = 3;        // Timestamp associated with data acquisition time
  159.         optional uint32   datatype      = 4;        // DataType of the metric/tag value
  160.         optional bool     is_historical = 5;        // If this is historical data and should not update real time tag
  161.         optional bool     is_transient  = 6;        // Tells consuming clients such as MQTT Engine to not store this as a tag
  162.         optional bool     is_null       = 7;        // If this is null - explicitly say so rather than using -1, false, etc for some datatypes.
  163.         optional MetaData metadata      = 8;        // Metadata for the payload
  164.         optional PropertySet properties = 9;
  166.         oneof value {
  167.             uint32   int_value                      = 10;
  168.             uint64   long_value                     = 11;
  169.             float    float_value                    = 12;
  170.             double   double_value                   = 13;
  171.             bool     boolean_value                  = 14;
  172.             string   string_value                   = 15;
  173.             bytes    bytes_value                    = 16;       // Bytes, File
  174.             DataSet  dataset_value                  = 17;
  175.             Template template_value                 = 18;
  176.             MetricValueExtension extension_value    = 19;
  177.         }
  179.         message MetricValueExtension {
  180.             extensions  1 to max;
  181.         }
  182.     }
  184.     optional uint64   timestamp     = 1;        // Timestamp at message sending time
  185.     repeated Metric   metrics       = 2;        // Repeated forever - no limit in Google Protobufs
  186.     optional uint64   seq           = 3;        // Sequence number
  187.     optional string   uuid          = 4;        // UUID to track message type in terms of schema definitions
  188.     optional bytes    body          = 5;        // To optionally bypass the whole definition above
  189.     extensions                      6 to max;   // For third party extensions
  190. }

Create Rule

SQL statement

  1. SELECT
  2.   schema_decode('neuron', payload, 'Payload') as SparkPlugB
  3. FROM
  4.   "spBv1.0/group1/DDATA/node1/modbus"

The key point here is schema_decode('neuron', payload, 'Payload'):

  • schema_decode The function decodes the content of the payload field according to the Schema ‘protobuf_person’;
  • as SparkPlugB Store the decoded value in the variable “SparkPlugB”;
  • The last parameter Payload indicates that the message type in the payload is the ‘Payload’ type defined in the protobuf schema.

image-20230421143952695

Then add the action with the following parameters:

  • Action Type: Message republish
  • Purpose topic: SparkPlugB/test

This action sends the decoded “Payload” to the SparkPlugB/test topic in JSON format.

image-20230421143453011

Verify with MQTTX

Here, the MQTTX tool is used to subscribe to the data decoded by the codec function of the EMQX rule engine, as shown in the figure:

image-20230421151817047

As shown in the figure above, it can be seen that the original data before decoding is garbled, and the complete and correct data result is obtained after decoding; so far, the point value of the device is collected in the south direction of Neuron, and reported to EMQX in the north direction of SparkPlugB, and the complete data is obtained by decoding through the codec function Data results are complete.

Further Reading

The topic that Neuron reports data to EMQX is namespace/group_id/DDATA/edge_node_id/device_id defined according to the Sparkplug B protocol specification, as shown in the figure:

image-20230419143059088

As for how to define more Neuron northbound Sparkplug B plug-in related standards, you can refer to the Sparkplug B protocol specification🔗