feat(sysk): integrated Nari Syskeeper 2000 as a new bridge backend

apps/emqx_bridge/src/schema/emqx_bridge_enterprise.erl

@@ -50,7 +50,9 @@ api_schemas(Method) ->
         api_ref(emqx_bridge_rabbitmq, <<"rabbitmq">>, Method),
         api_ref(emqx_bridge_kinesis, <<"kinesis_producer">>, Method ++ "_producer"),
         api_ref(emqx_bridge_greptimedb, <<"greptimedb">>, Method ++ "_grpc_v1"),
-        api_ref(emqx_bridge_azure_event_hub, <<"azure_event_hub_producer">>, Method ++ "_producer")
+        api_ref(emqx_bridge_azure_event_hub, <<"azure_event_hub_producer">>, Method ++ "_producer"),
+        api_ref(emqx_bridge_syskeeper, <<"syskeeper">>, Method),
+        api_ref(emqx_bridge_syskeeper_proxy, <<"syskeeper_proxy">>, Method)
     ].
 
 schema_modules() ->
@@ -78,7 +80,9 @@ schema_modules() ->
         emqx_bridge_rabbitmq,
         emqx_bridge_kinesis,
         emqx_bridge_greptimedb,
-        emqx_bridge_azure_event_hub
+        emqx_bridge_azure_event_hub,
+        emqx_bridge_syskeeper,
+        emqx_bridge_syskeeper_proxy
     ].
 
 examples(Method) ->
@@ -126,7 +130,9 @@ resource_type(rabbitmq) -> emqx_bridge_rabbitmq_connector;
 resource_type(kinesis_producer) -> emqx_bridge_kinesis_impl_producer;
 resource_type(greptimedb) -> emqx_bridge_greptimedb_connector;
 %% We use AEH's Kafka interface.
-resource_type(azure_event_hub_producer) -> emqx_bridge_kafka_impl_producer.
+resource_type(azure_event_hub_producer) -> emqx_bridge_kafka_impl_producer;
+resource_type(syskeeper) -> emqx_bridge_syskeeper_connector;
+resource_type(syskeeper_proxy) -> emqx_bridge_syskeeper_proxy_server.
 
 %% For bridges that need to override connector configurations.
 bridge_impl_module(BridgeType) when is_binary(BridgeType) ->
@@ -215,7 +221,8 @@ fields(bridges) ->
         influxdb_structs() ++
         redis_structs() ++
         pgsql_structs() ++ clickhouse_structs() ++ sqlserver_structs() ++ rabbitmq_structs() ++
-        kinesis_structs() ++ greptimedb_structs() ++ azure_event_hub_structs().
+        kinesis_structs() ++ greptimedb_structs() ++ azure_event_hub_structs() ++
+        syskeeper_structs().
 
 mongodb_structs() ->
     [
@@ -428,6 +435,26 @@ azure_event_hub_structs() ->
             )}
     ].
 
+syskeeper_structs() ->
+    [
+        {syskeeper,
+            mk(
+                hoconsc:map(name, ref(emqx_bridge_syskeeper, "config")),
+                #{
+                    desc => <<"Syskeeper bridge config ">>,
+                    required => false
+                }
+            )},
+        {syskeeper_proxy,
+            mk(
+                hoconsc:map(name, ref(emqx_bridge_syskeeper_proxy, "config")),
+                #{
+                    desc => <<"Syskeeper proxy server config">>,
+                    required => false
+                }
+            )}
+    ].
+
 api_ref(Module, Type, Method) ->
     {Type, ref(Module, Method)}.
 

apps/emqx_bridge_syskeeper/BSL.txt

@@ -0,0 +1,94 @@
+Business Source License 1.1
+
+Licensor:             Hangzhou EMQ Technologies Co., Ltd.
+Licensed Work:        EMQX Enterprise Edition
+                      The Licensed Work is (c) 2023
+                      Hangzhou EMQ Technologies Co., Ltd.
+Additional Use Grant: Students and educators are granted right to copy,
+                      modify, and create derivative work for research
+                      or education.
+Change Date:          2027-02-01
+Change License:       Apache License, Version 2.0
+
+For information about alternative licensing arrangements for the Software,
+please contact Licensor: https://www.emqx.com/en/contact
+
+Notice
+
+The Business Source License (this document, or the “License”) is not an Open
+Source license. However, the Licensed Work will eventually be made available
+under an Open Source License, as stated in this License.
+
+License text copyright (c) 2017 MariaDB Corporation Ab, All Rights Reserved.
+“Business Source License” is a trademark of MariaDB Corporation Ab.
+
+-----------------------------------------------------------------------------
+
+Business Source License 1.1
+
+Terms
+
+The Licensor hereby grants you the right to copy, modify, create derivative
+works, redistribute, and make non-production use of the Licensed Work. The
+Licensor may make an Additional Use Grant, above, permitting limited
+production use.
+
+Effective on the Change Date, or the fourth anniversary of the first publicly
+available distribution of a specific version of the Licensed Work under this
+License, whichever comes first, the Licensor hereby grants you rights under
+the terms of the Change License, and the rights granted in the paragraph
+above terminate.
+
+If your use of the Licensed Work does not comply with the requirements
+currently in effect as described in this License, you must purchase a
+commercial license from the Licensor, its affiliated entities, or authorized
+resellers, or you must refrain from using the Licensed Work.
+
+All copies of the original and modified Licensed Work, and derivative works
+of the Licensed Work, are subject to this License. This License applies
+separately for each version of the Licensed Work and the Change Date may vary
+for each version of the Licensed Work released by Licensor.
+
+You must conspicuously display this License on each original or modified copy
+of the Licensed Work. If you receive the Licensed Work in original or
+modified form from a third party, the terms and conditions set forth in this
+License apply to your use of that work.
+
+Any use of the Licensed Work in violation of this License will automatically
+terminate your rights under this License for the current and all other
+versions of the Licensed Work.
+
+This License does not grant you any right in any trademark or logo of
+Licensor or its affiliates (provided that you may use a trademark or logo of
+Licensor as expressly required by this License).
+
+TO THE EXTENT PERMITTED BY APPLICABLE LAW, THE LICENSED WORK IS PROVIDED ON
+AN “AS IS” BASIS. LICENSOR HEREBY DISCLAIMS ALL WARRANTIES AND CONDITIONS,
+EXPRESS OR IMPLIED, INCLUDING (WITHOUT LIMITATION) WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT, AND
+TITLE.
+
+MariaDB hereby grants you permission to use this License’s text to license
+your works, and to refer to it using the trademark “Business Source License”,
+as long as you comply with the Covenants of Licensor below.
+
+Covenants of Licensor
+
+In consideration of the right to use this License’s text and the “Business
+Source License” name and trademark, Licensor covenants to MariaDB, and to all
+other recipients of the licensed work to be provided by Licensor:
+
+1. To specify as the Change License the GPL Version 2.0 or any later version,
+   or a license that is compatible with GPL Version 2.0 or a later version,
+   where “compatible” means that software provided under the Change License can
+   be included in a program with software provided under GPL Version 2.0 or a
+   later version. Licensor may specify additional Change Licenses without
+   limitation.
+
+2. To either: (a) specify an additional grant of rights to use that does not
+   impose any additional restriction on the right granted in this License, as
+   the Additional Use Grant; or (b) insert the text “None”.
+
+3. To specify a Change Date.
+
+4. Not to modify this License in any other way.

apps/emqx_bridge_syskeeper/README.md

@@ -0,0 +1,30 @@
+# EMQX Syskeeper Bridge
+
+Nari Syskeeper 2000 is a one-way Physical Isolation Net Gap.
+
+The application is used to connect EMQX and Syskeeper.
+Users can create a rule and quickly ingest IoT data to the Syskeeper by leveraging
+[EMQX Rules](https://docs.emqx.com/en/enterprise/v5.0/data-integration/rules.html).
+
+# Documentation
+
+- Refer to [Rules engine](https://docs.emqx.com/en/enterprise/v5.0/data-integration/rules.html)
+  for the EMQX rules engine introduction.
+
+# HTTP APIs
+
+- Several APIs are provided for bridge management, which includes create bridge,
+  update bridge, get bridge, stop or restart bridge and list bridges etc.
+
+  Refer to [API Docs - Bridges](https://docs.emqx.com/en/enterprise/v5.0/admin/api-docs.html#tag/Bridges)
+  for more detailed information.
+
+
+# Contributing
+
+Please see our [contributing.md](../../CONTRIBUTING.md).
+
+
+# License
+
+EMQ Business Source License 1.1, refer to [LICENSE](BSL.txt).

apps/emqx_bridge_syskeeper/doc/protocol_v1.md

@@ -0,0 +1,370 @@
+
+# Table of Contents
+
+1.  [Packet Format](#orgb2a43d1)
+2.  [Common Header](#org5ca4c69)
+    1.  [Types](#org240efb3)
+    2.  [Shared Flags](#org804fcce)
+3.  [Handshake Packet](#org6a73ea8)
+4.  [Forward Packet](#org39c753e)
+    1.  [Flags](#org5177d26)
+    2.  [Payload](#orgb29cbd7)
+        1.  [Message Content map structure](#org75acfe6)
+5.  [Heartbeat Packet](#org388b69a)
+
+
+<a id="orgb2a43d1"></a>
+
+# Packet Format
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;bytes&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;0&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;1&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;2&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;3&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;5&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;6&nbsp;..&nbsp;end&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="4" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;variable&nbsp;length&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;common&nbsp;header&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;payload&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+</table>
+
+The length of the remaining part(common header + payload) is indicated by the Length Header of each packet
+
+
+<a id="org5ca4c69"></a>
+
+# Common Header
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;bits&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;0&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;1&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;2&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;3&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;4&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;5&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;6&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;7&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="4" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;packet&nbsp;type&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="4" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;shared&nbsp;flags&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+</table>
+
+
+<a id="org240efb3"></a>
+
+## Types
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;type&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;usage&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;0&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;handshake&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;1&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;forward&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;2&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;heartbeat&nbsp;
+    </td>
+  </tr>
+</table>
+
+
+<a id="org804fcce"></a>
+
+## Shared Flags
+
+The usage of each bit is determined by the type of packet
+
+
+<a id="org6a73ea8"></a>
+
+# Handshake Packet
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;bytes&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;0&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;1&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;common&nbsp;header&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;version&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+</table>
+
+
+<a id="org39c753e"></a>
+
+# Forward Packet
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;bits&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;0&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;1&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;2&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;3&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;4&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;5&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;6&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;7&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;...&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td rowspan="2" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<br />
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<br />
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="4" rowspan="2" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<br />
+      &nbsp;&nbsp;packet&nbsp;type&nbsp;&nbsp;<br />
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="3" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;ACK&nbsp;
+    </td>
+    <td rowspan="2" align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<br />
+      &nbsp;&nbsp;payload&nbsp;&nbsp;<br />
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td colspan="4" align="left" valign="top">
+      &nbsp;&nbsp;forward&nbsp;flags&nbsp;&nbsp;
+    </td>
+  </tr>
+</table>
+
+
+<a id="org5177d26"></a>
+
+## Flags
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;flag&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;usage&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;ACK&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;This&nbsp;packet&nbsp;need&nbsp;a&nbsp;ACK&nbsp;response&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+</table>
+
+
+<a id="orgb29cbd7"></a>
+
+## Payload
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;bytes&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;0&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;..&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;n&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;n+1&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;..&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;x&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="3" align="left" valign="top">
+      &nbsp;&nbsp;Content&nbsp;Length&nbsp;&nbsp;&nbsp;
+    </td>
+    <td colspan="3" align="left" valign="top">
+      &nbsp;Message&nbsp;Content&nbsp;
+    </td>
+  </tr>
+</table>
+
+-   Content length is a variable length number.
+-   Message content is a list in an opaque binary format whose element is a map structure
+
+
+<a id="org75acfe6"></a>
+
+### Message Content map structure
+
+    {
+      id: "0006081CCFF3D48F03C10000058B0000", // unique message id
+      qos: 1,
+      flags: {dup: false, retain: false},
+      from: "clientid",
+      topic: "t/1",
+      payload: "hello, world",
+      timestamp: 1697786555281
+    }
+
+
+<a id="org388b69a"></a>
+
+# Heartbeat Packet
+
+<!-- This HTML table template is generated by emacs 29.0.92 -->
+<table border="1">
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;bytes&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;0&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+  </tr>
+  <tr>
+    <td align="left" valign="top">
+      &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
+    </td>
+    <td align="left" valign="top">
+      &nbsp;common&nbsp;header&nbsp;
+    </td>
+  </tr>
+</table>
+

apps/emqx_bridge_syskeeper/doc/protocol_v1.org

@@ -0,0 +1,80 @@
+* Packet Format
+    +-------+-----+-----+-----+-----+-----------------+----------------+
+    | bytes |  0  |  1  |  2  |  3  |        5        |    6 .. end    |
+    +-------+-----+-----+-----+-----+-----------------+----------------+
+    |       |    variable length    |  common header  |    payload     |
+    +-------+-----------------------+-----------------+----------------+
+
+    The length of the remaining part(common header + payload) is indicated by the Length Header of each packet
+
+* Common Header
+    +------+-----+-----+-----+-----+-----+-----+-----+-----+
+    | bits |  0  |  1  |  2  |  3  |  4  |  5  |  6  |  7  |
+    +------+-----+-----+-----+-----+-----+-----+-----+-----+
+    |      |      packet type      |     shared flags      |
+    +------+-----------------------+-----------------------+
+** Types
+    +----------+-----------+
+    |   type   |   usage   |
+    +----------+-----------+
+    |    0     | handshake |
+    +----------+-----------+
+    |    1     |  forward  |
+    +----------+-----------+
+    |    2     | heartbeat |
+    +----------+-----------+
+** Shared Flags
+    The usage of each bit is determined by the type of packet
+* Handshake Packet
+    +-------+---------------+---------------+
+    | bytes |       0       |       1       |
+    +-------+---------------+---------------+
+    |       | common header |    version    |
+    +-------+---------------+---------------+
+* Forward Packet
+    +------+---+---+---+---+---+---+---+-----+-----------+
+    | bits | 0 | 1 | 2 | 3 | 4 | 5 | 6 |  7  |    ...    |
+    +------+---+---+---+---+---+---+---+-----+-----------+
+    |      |               |           | ACK |           |
+    |      |  packet type  +-----------+-----+  payload  |
+    |      |               |  forward flags  |           |
+    +------+---------------+-----------------+-----------+
+
+** Flags
+    +------+-------------------------------------------+
+    | flag |                   usage                   |
+    +------+-------------------------------------------+
+    | ACK  |      This packet need a ACK response      |
+    +------+-------------------------------------------+
+
+** Payload
+    +-------+-----+-------+-----+-----+-----+-----+
+    | bytes |  0  |  ..   |  n  | n+1 | ..  |  x  |
+    +-------+-----+-------+-----+-----+-----+-----+
+    |       |  Content Length   | Message Content |
+    +-------+-------------------+-----------------+
+
+    + Content length is a variable length number.
+    + Message content is a list in an opaque binary format whose element is a map structure
+
+*** Message Content map structure
+
+#+begin_src json
+  {
+    id: "0006081CCFF3D48F03C10000058B0000", // unique message id
+    qos: 1,
+    flags: {dup: false, retain: false},
+    from: "clientid",
+    topic: "t/1",
+    payload: "hello, world",
+    timestamp: 1697786555281
+  }
+#+end_src
+
+* Heartbeat Packet
+
+    +-------+---------------+
+    | bytes |       0       |
+    +-------+---------------+
+    |       | common header |
+    +-------+---------------+

apps/emqx_bridge_syskeeper/docker-ct

@@ -0,0 +1 @@
+toxiproxy

apps/emqx_bridge_syskeeper/include/emqx_bridge_syskeeper.hrl

@@ -0,0 +1,15 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-ifndef(EMQX_BRIDGE_SYSKEEPER).
+-define(EMQX_BRIDGE_SYSKEEPER, true).
+
+-define(TYPE_HANDSHAKE, 0).
+-define(TYPE_FORWARD, 1).
+-define(TYPE_HEARTBEAT, 2).
+
+-type packet_type() :: handshake | forward | heartbeat.
+-type packet_data() :: none | binary() | [binary()].
+-type packet_type_val() :: ?TYPE_HANDSHAKE..?TYPE_HEARTBEAT.
+
+-endif.

apps/emqx_bridge_syskeeper/rebar.config

@@ -0,0 +1,6 @@
+%% -*- mode: erlang; -*-
+{erl_opts, [debug_info]}.
+{deps, [  {emqx_connector, {path, "../../apps/emqx_connector"}}
+       , {emqx_resource, {path, "../../apps/emqx_resource"}}
+       , {emqx_bridge, {path, "../../apps/emqx_bridge"}}
+       ]}.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper.app.src

@@ -0,0 +1,13 @@
+{application, emqx_bridge_syskeeper, [
+    {description, "EMQX Enterprise Bridge"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {applications, [
+        kernel,
+        stdlib,
+        emqx_resource
+    ]},
+    {env, []},
+    {modules, []},
+    {links, []}
+]}.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper.erl

@@ -0,0 +1,117 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_bridge_syskeeper).
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("emqx_bridge/include/emqx_bridge.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-export([
+    conn_bridge_examples/1,
+    values/1
+]).
+
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+%% -------------------------------------------------------------------------------------------------
+%% api
+conn_bridge_examples(Method) ->
+    [
+        #{
+            <<"syskeeper">> => #{
+                summary => <<"Syskeeper Bridge">>,
+                value => values(Method)
+            }
+        }
+    ].
+
+values(_Method) ->
+    #{
+        enable => true,
+        type => syskeeper,
+        name => <<"foo">>,
+        server => <<"127.0.0.1:9092">>,
+        ack_mode => <<"no_ack">>,
+        ack_timeout => <<"10s">>,
+        pool_size => 16,
+        target_topic => <<"${topic}">>,
+        target_qos => <<"-1">>,
+        template => <<"${payload}">>,
+        resource_opts => #{
+            worker_pool_size => 16,
+            health_check_interval => ?HEALTHCHECK_INTERVAL_RAW,
+            batch_size => ?DEFAULT_BATCH_SIZE,
+            batch_time => ?DEFAULT_BATCH_TIME,
+            query_mode => sync,
+            max_buffer_bytes => ?DEFAULT_BUFFER_BYTES
+        }
+    }.
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon Schema Definitions
+namespace() -> "bridge_syskeeper".
+
+roots() -> [].
+
+fields("config") ->
+    [
+        {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
+        {target_topic,
+            mk(
+                binary(),
+                #{desc => ?DESC("target_topic"), default => <<"${topic}">>}
+            )},
+        {target_qos,
+            mk(
+                range(-1, 2),
+                #{desc => ?DESC("target_qos"), default => -1}
+            )},
+        {template,
+            mk(
+                binary(),
+                #{desc => ?DESC("template"), default => <<"${payload}">>}
+            )},
+        {resource_opts,
+            mk(
+                ref(?MODULE, "creation_opts"),
+                #{
+                    required => false,
+                    default => #{},
+                    desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
+                }
+            )}
+    ] ++ emqx_bridge_syskeeper_connector:fields(config);
+fields("creation_opts") ->
+    emqx_resource_schema:create_opts([{request_ttl, #{default => infinity}}]);
+fields("post") ->
+    [type_field(), name_field() | fields("config")];
+fields("put") ->
+    fields("config");
+fields("get") ->
+    emqx_bridge_schema:status_fields() ++ fields("post").
+
+desc("config") ->
+    ?DESC("desc_config");
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for Syskeeper using `", string:to_upper(Method), "` method."];
+desc("creation_opts" = Name) ->
+    emqx_resource_schema:desc(Name);
+desc(_) ->
+    undefined.
+
+%% -------------------------------------------------------------------------------------------------
+
+type_field() ->
+    {type, mk(enum([syskeeper]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+    {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_client.erl

@@ -0,0 +1,180 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_bridge_syskeeper_client).
+
+-behaviour(gen_server).
+
+%% API
+-export([
+    start_link/1,
+    forward/3,
+    heartbeat/2
+]).
+
+%% gen_server callbacks
+-export([
+    init/1,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    terminate/2,
+    code_change/3,
+    format_status/2
+]).
+
+-include("emqx_bridge_syskeeper.hrl").
+
+-type state() :: #{
+    ack_mode := need_ack | no_ack,
+    ack_timeout := timer:time(),
+    socket := undefined | inet:socket(),
+    frame_state := emqx_bridge_syskeeper_frame:state(),
+    last_error := undefined | tuple()
+}.
+
+-type send_result() :: {ok, state()} | {error, term()}.
+
+%% -------------------------------------------------------------------------------------------------
+%% API
+forward(Pid, Msg, Timeout) ->
+    call(Pid, {?FUNCTION_NAME, Msg}, Timeout).
+
+heartbeat(Pid, Timeout) ->
+    ok =:= call(Pid, ?FUNCTION_NAME, Timeout).
+
+%% -------------------------------------------------------------------------------------------------
+%% Starts Bridge which transfer data to Syskeeper
+
+start_link(Options) ->
+    gen_server:start_link(?MODULE, Options, []).
+
+%% -------------------------------------------------------------------------------------------------
+%%% gen_server callbacks
+
+%% Initialize syskeeper client
+init(#{ack_timeout := AckTimeout, ack_mode := AckMode} = Options) ->
+    erlang:process_flag(trap_exit, true),
+    connect(Options, #{
+        ack_timeout => AckTimeout,
+        ack_mode => AckMode,
+        socket => undefined,
+        last_error => undefined,
+        frame_state => emqx_bridge_syskeeper_frame:make_state_with_conf(Options)
+    }).
+
+handle_call({forward, Msgs}, _From, State) ->
+    Result = send_packet(forward, Msgs, State),
+    handle_reply_result(Result, State);
+handle_call(heartbeat, _From, State) ->
+    Result = send_ack_packet(heartbeat, none, State),
+    handle_reply_result(Result, State);
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+handle_info({tcp_closed, _} = Reason, State) ->
+    {noreply, State#{socket := undefined, last_error := Reason}};
+handle_info({last_error, _, _} = Reason, State) ->
+    {noreply, State#{socket := undefined, last_error := Reason}};
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, #{socket := Socket} = _State) ->
+    close_socket(Socket),
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+-spec format_status(
+    Opt :: normal | terminate,
+    Status :: list()
+) -> Status :: term().
+format_status(_Opt, Status) ->
+    Status.
+
+%% ------------------------------------------------------------------------------------------------
+connect(
+    #{
+        hostname := Host,
+        port := Port
+    },
+    State
+) ->
+    case
+        gen_tcp:connect(Host, Port, [
+            {active, true},
+            {mode, binary},
+            {nodelay, true}
+        ])
+    of
+        {ok, Socket} ->
+            send_ack_packet(handshake, none, State#{socket := Socket});
+        {error, Reason} ->
+            {stop, Reason}
+    end.
+
+-spec send_ack_packet(packet_type(), packet_data(), state()) -> send_result().
+send_ack_packet(Type, Data, State) ->
+    send_packet(Type, Data, State, true).
+
+-spec send_packet(packet_type(), packet_data(), state()) -> send_result().
+send_packet(Type, Data, State) ->
+    send_packet(Type, Data, State, false).
+
+-spec send_packet(packet_type(), packet_data(), state(), boolean()) -> send_result().
+send_packet(_Type, _Data, #{socket := undefined, last_error := Reason}, _Force) ->
+    {error, Reason};
+send_packet(Type, Data, #{frame_state := FrameState} = State, Force) ->
+    Packet = emqx_bridge_syskeeper_frame:encode(Type, Data, FrameState),
+    case socket_send(Packet, State) of
+        ok ->
+            wait_ack(State, Force);
+        {error, _} = Error ->
+            Error
+    end.
+
+-spec socket_send(binary() | [binary()], state()) -> ok | {error, _Reason}.
+socket_send(Bin, State) when is_binary(Bin) ->
+    socket_send([Bin], State);
+socket_send(Bins, #{socket := Socket}) ->
+    Map = fun(Data) ->
+        Len = erlang:byte_size(Data),
+        VarLen = emqx_bridge_syskeeper_frame:serialize_variable_byte_integer(Len),
+        <<VarLen/binary, Data/binary>>
+    end,
+    gen_tcp:send(Socket, lists:map(Map, Bins)).
+
+-spec wait_ack(state(), boolean()) -> send_result().
+wait_ack(#{ack_timeout := AckTimeout, ack_mode := AckMode} = State, Force) when
+    AckMode =:= need_ack; Force
+->
+    receive
+        {tcp, _Socket, <<16#FF>>} ->
+            {ok, State};
+        {tcp_closed, _} = Reason ->
+            {error, Reason};
+        {tcp_error, _, _} = Reason ->
+            {error, Reason}
+    after AckTimeout ->
+        {error, wait_ack_timeout}
+    end;
+wait_ack(State, _Force) ->
+    {ok, State}.
+
+close_socket(undefined) ->
+    ok;
+close_socket(Socket) ->
+    catch gen_tcp:close(Socket),
+    ok.
+
+call(Pid, Msg, Timeout) ->
+    gen_server:call(Pid, Msg, Timeout).
+
+handle_reply_result({ok, _}, State) ->
+    {reply, ok, State};
+handle_reply_result({error, Reason}, State) ->
+    {reply, {error, {recoverable_error, Reason}}, State#{last_error := Reason}}.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_connector.erl

@@ -0,0 +1,262 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_syskeeper_connector).
+
+-behaviour(emqx_resource).
+
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+-include_lib("typerefl/include/types.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+
+-export([roots/0, fields/1]).
+
+%% `emqx_resource' API
+-export([
+    callback_mode/0,
+    query_mode/1,
+    on_start/2,
+    on_stop/2,
+    on_query/3,
+    on_batch_query/3,
+    on_get_status/2
+]).
+
+-export([
+    connect/1
+]).
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-define(SYSKEEPER_HOST_OPTIONS, #{
+    default_port => 9092
+}).
+
+-define(EXTRA_CALL_TIMEOUT, 2000).
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon schema
+roots() ->
+    [{config, #{type => hoconsc:ref(?MODULE, config)}}].
+
+fields(config) ->
+    [
+        {server, server()},
+        {ack_mode,
+            mk(
+                enum([need_ack, no_ack]),
+                #{desc => ?DESC(ack_mode), default => <<"no_ack">>}
+            )},
+        {ack_timeout,
+            mk(
+                emqx_schema:timeout_duration_ms(),
+                #{desc => ?DESC(ack_timeout), default => <<"10s">>}
+            )},
+        {pool_size, fun
+            (default) ->
+                16;
+            (Other) ->
+                emqx_connector_schema_lib:pool_size(Other)
+        end}
+    ].
+
+server() ->
+    Meta = #{desc => ?DESC("server")},
+    emqx_schema:servers_sc(Meta, ?SYSKEEPER_HOST_OPTIONS).
+
+%% -------------------------------------------------------------------------------------------------
+%% `emqx_resource' API
+
+callback_mode() -> always_sync.
+
+query_mode(_) -> sync.
+
+on_start(
+    InstanceId,
+    #{
+        server := Server,
+        pool_size := PoolSize,
+        ack_timeout := AckTimeout,
+        target_topic := TargetTopic,
+        target_qos := TargetQoS
+    } = Config
+) ->
+    ?SLOG(info, #{
+        msg => "starting_syskeeper_connector",
+        connector => InstanceId,
+        config => redact(Config)
+    }),
+
+    HostCfg = emqx_schema:parse_server(Server, ?SYSKEEPER_HOST_OPTIONS),
+
+    Options = [
+        {options,
+            maps:merge(
+                HostCfg,
+                maps:with([ack_mode, ack_timeout], Config)
+            )},
+        {pool_size, PoolSize}
+    ],
+
+    State = #{
+        pool_name => InstanceId,
+        target_qos => TargetQoS,
+        ack_timeout => AckTimeout,
+        templates => parse_template(Config),
+        target_topic_tks => emqx_placeholder:preproc_tmpl(TargetTopic)
+    },
+    case emqx_resource_pool:start(InstanceId, ?MODULE, Options) of
+        ok ->
+            {ok, State};
+        Error ->
+            Error
+    end.
+
+on_stop(InstanceId, _State) ->
+    ?SLOG(info, #{
+        msg => "stopping_syskeeper_connector",
+        connector => InstanceId
+    }),
+    emqx_resource_pool:stop(InstanceId).
+
+on_query(InstanceId, {send_message, _} = Query, State) ->
+    do_query(InstanceId, [Query], State);
+on_query(_InstanceId, Query, _State) ->
+    {error, {unrecoverable_error, {invalid_request, Query}}}.
+
+%% we only support batch insert
+on_batch_query(InstanceId, [{send_message, _} | _] = Query, State) ->
+    do_query(InstanceId, Query, State);
+on_batch_query(_InstanceId, Query, _State) ->
+    {error, {unrecoverable_error, {invalid_request, Query}}}.
+
+on_get_status(_InstanceId, #{pool_name := Pool, ack_timeout := AckTimeout}) ->
+    Health = emqx_resource_pool:health_check_workers(
+        Pool, {emqx_bridge_syskeeper_client, heartbeat, [AckTimeout + ?EXTRA_CALL_TIMEOUT]}
+    ),
+    status_result(Health).
+
+status_result(true) -> connected;
+status_result(false) -> connecting;
+status_result({error, _}) -> connecting.
+
+%% -------------------------------------------------------------------------------------------------
+%% Helper fns
+
+do_query(
+    InstanceId,
+    Query,
+    #{pool_name := PoolName, ack_timeout := AckTimeout} = State
+) ->
+    ?TRACE(
+        "QUERY",
+        "syskeeper_connector_received",
+        #{connector => InstanceId, query => Query, state => State}
+    ),
+
+    Result =
+        case try_apply_template(Query, State) of
+            {ok, Msg} ->
+                ecpool:pick_and_do(
+                    PoolName,
+                    {emqx_bridge_syskeeper_client, forward, [Msg, AckTimeout + ?EXTRA_CALL_TIMEOUT]},
+                    no_handover
+                );
+            Error ->
+                Error
+        end,
+
+    case Result of
+        {error, Reason} ->
+            ?tp(
+                syskeeper_connector_query_return,
+                #{error => Reason}
+            ),
+            %% ?SLOG(error, #{
+            %%     msg => "syskeeper_connector_do_query_failed",
+            %%     connector => InstanceId,
+            %%     query => Query,
+            %%     reason => Reason
+            %% }),
+            case Reason of
+                ecpool_empty ->
+                    {error, {recoverable_error, Reason}};
+                _ ->
+                    Result
+            end;
+        _ ->
+            %% ?tp(
+            %%     syskeeper_connector_query_return,
+            %%     #{result => Result}
+            %% ),
+            Result
+    end.
+
+connect(Opts) ->
+    Options = proplists:get_value(options, Opts),
+    emqx_bridge_syskeeper_client:start_link(Options).
+
+parse_template(Config) ->
+    Templates =
+        case maps:get(template, Config, undefined) of
+            undefined -> #{};
+            <<>> -> #{};
+            Template -> #{send_message => Template}
+        end,
+
+    parse_template(maps:to_list(Templates), #{}).
+
+parse_template([{Key, H} | T], Templates) ->
+    ParamsTks = emqx_placeholder:preproc_tmpl(H),
+    parse_template(
+        T,
+        Templates#{Key => ParamsTks}
+    );
+parse_template([], Templates) ->
+    Templates.
+
+try_apply_template([{Type, _} | _] = Datas, #{templates := Templates} = State) ->
+    case maps:find(Type, Templates) of
+        {ok, Template} ->
+            {ok, apply_template(Datas, Template, State)};
+        _ ->
+            {error, {unrecoverable_error, {invalid_request, Datas}}}
+    end.
+
+apply_template(Datas, Template, State) ->
+    lists:map(
+        fun({_, Data}) ->
+            do_apply_template(Data, Template, State)
+        end,
+        Datas
+    ).
+
+do_apply_template(#{id := Id, qos := QoS, clientid := From} = Data, Template, #{
+    target_qos := TargetQoS, target_topic_tks := TargetTopicTks
+}) ->
+    Msg = maps:with([qos, flags, topic, payload, timestamp], Data),
+    Topic = emqx_placeholder:proc_tmpl(TargetTopicTks, Msg),
+    Msg#{
+        id => emqx_guid:from_hexstr(Id),
+        qos :=
+            case TargetQoS of
+                -1 ->
+                    QoS;
+                _ ->
+                    TargetQoS
+            end,
+        from => From,
+        topic := Topic,
+        payload := format_data(Template, Msg)
+    }.
+
+format_data([], Msg) ->
+    emqx_utils_json:encode(Msg);
+format_data(Tokens, Msg) ->
+    emqx_placeholder:proc_tmpl(Tokens, Msg).
+
+redact(Data) ->
+    emqx_utils:redact(Data, fun(Any) -> Any =:= aws_secret_access_key end).

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_frame.erl

@@ -0,0 +1,163 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% @doc EMQ X Bridge Sysk Frame
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_syskeeper_frame).
+
+%% API
+-export([
+    versions/0,
+    current_version/0,
+    make_state_with_conf/1,
+    make_state/1,
+    encode/3,
+    parse/2,
+    parse_handshake/1
+]).
+
+-export([
+    bool2int/1,
+    int2bool/1,
+    marshaller/1,
+    serialize_variable_byte_integer/1,
+    parse_variable_byte_integer/1
+]).
+
+-export_type([state/0, versions/0, handshake/0, forward/0, packet/0]).
+
+-include("emqx_bridge_syskeeper.hrl").
+
+-type state() :: #{
+    handler := atom(),
+    version := versions(),
+    ack => boolean()
+}.
+
+-type versions() :: 1.
+
+-type handshake() :: #{type := handshake, version := versions()}.
+-type forward() :: #{type := forward, ack := boolean(), messages := list(map())}.
+-type heartbeat() :: #{type := heartbeat}.
+
+-type packet() ::
+    handshake()
+    | forward()
+    | heartbeat().
+
+-callback version() -> versions().
+-callback encode(packet_type_val(), packet_data(), state()) -> binary().
+-callback parse(packet_type(), binary(), state()) -> packet().
+
+-define(HIGHBIT, 2#10000000).
+-define(LOWBITS, 2#01111111).
+-define(MULTIPLIER_MAX, 16#200000).
+
+-export_type([packet_type/0]).
+
+%%-------------------------------------------------------------------
+%%% API
+%%-------------------------------------------------------------------
+-spec versions() -> list(versions()).
+versions() ->
+    [1].
+
+-spec current_version() -> versions().
+current_version() ->
+    1.
+
+-spec make_state_with_conf(map()) -> state().
+make_state_with_conf(#{ack_mode := Mode}) ->
+    State = make_state(current_version()),
+    State#{ack => Mode =:= need_ack}.
+
+-spec make_state(versions()) -> state().
+make_state(Version) ->
+    case lists:member(Version, versions()) of
+        true ->
+            Handler = erlang:list_to_existing_atom(
+                io_lib:format("emqx_bridge_syskeeper_frame_v~B", [Version])
+            ),
+            #{
+                handler => Handler,
+                version => Version
+            };
+        _ ->
+            erlang:throw({unsupport_version, Version})
+    end.
+
+-spec encode(packet_type(), term(), state()) -> binary().
+encode(Type, Data, #{handler := Handler} = State) ->
+    Handler:encode(packet_type_val(Type), Data, State).
+
+-spec parse(binary(), state()) -> _.
+parse(<<TypeVal:4, _:4, _/binary>> = Bin, #{handler := Handler} = State) ->
+    Type = to_packet_type(TypeVal),
+    Handler:parse(Type, Bin, State).
+
+parse_handshake(Data) ->
+    State = make_state(1),
+    parse_handshake(Data, State).
+
+parse_handshake(Data, #{version := Version} = State) ->
+    case parse(Data, State) of
+        {ok, #{type := handshake, version := Version} = Shake} ->
+            {ok, {State, Shake}};
+        {ok, #{type := handshake, version := NewVersion}} ->
+            State2 = make_state(NewVersion),
+            parse_handshake(Data, State2);
+        Error ->
+            Error
+    end.
+
+bool2int(true) ->
+    1;
+bool2int(_) ->
+    0.
+
+int2bool(1) ->
+    true;
+int2bool(_) ->
+    false.
+
+marshaller(Item) when is_binary(Item) ->
+    erlang:binary_to_term(Item);
+marshaller(Item) ->
+    erlang:term_to_binary(Item).
+
+serialize_variable_byte_integer(N) when N =< ?LOWBITS ->
+    <<0:1, N:7>>;
+serialize_variable_byte_integer(N) ->
+    <<1:1, (N rem ?HIGHBIT):7, (serialize_variable_byte_integer(N div ?HIGHBIT))/binary>>.
+
+parse_variable_byte_integer(Bin) ->
+    parse_variable_byte_integer(Bin, 1, 0).
+
+%%-------------------------------------------------------------------
+%%% Internal functions
+%%-------------------------------------------------------------------
+to_packet_type(?TYPE_HANDSHAKE) ->
+    handshake;
+to_packet_type(?TYPE_FORWARD) ->
+    forward;
+to_packet_type(?TYPE_HEARTBEAT) ->
+    heartbeat.
+
+packet_type_val(handshake) ->
+    ?TYPE_HANDSHAKE;
+packet_type_val(forward) ->
+    ?TYPE_FORWARD;
+packet_type_val(heartbeat) ->
+    ?TYPE_HEARTBEAT.
+
+parse_variable_byte_integer(<<1:1, _Len:7, _Rest/binary>>, Multiplier, _Value) when
+    Multiplier > ?MULTIPLIER_MAX
+->
+    {error, malformed_variable_byte_integer};
+parse_variable_byte_integer(<<1:1, Len:7, Rest/binary>>, Multiplier, Value) ->
+    parse_variable_byte_integer(Rest, Multiplier * ?HIGHBIT, Value + Len * Multiplier);
+parse_variable_byte_integer(<<0:1, Len:7, Rest/binary>>, Multiplier, Value) ->
+    {ok, Value + Len * Multiplier, Rest};
+parse_variable_byte_integer(<<>>, _Multiplier, _Value) ->
+    {error, incomplete}.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_frame_v1.erl

@@ -0,0 +1,70 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% @doc EMQ X Bridge Sysk Frame version 1
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_syskeeper_frame_v1).
+
+%% API
+-export([
+    version/0,
+    encode/3,
+    parse/3
+]).
+
+-behaviour(emqx_bridge_syskeeper_frame).
+
+-include("emqx_bridge_syskeeper.hrl").
+
+-define(B2I(X), emqx_bridge_syskeeper_frame:bool2int((X))).
+-define(I2B(X), emqx_bridge_syskeeper_frame:int2bool((X))).
+
+-import(emqx_bridge_syskeeper_frame, [
+    serialize_variable_byte_integer/1, parse_variable_byte_integer/1, marshaller/1
+]).
+
+%%-------------------------------------------------------------------
+%%% API
+%%-------------------------------------------------------------------
+version() ->
+    1.
+
+encode(?TYPE_HANDSHAKE = Type, _, _) ->
+    Version = version(),
+    <<Type:4, 0:4, Version:8>>;
+encode(?TYPE_FORWARD = Type, Messages, #{ack := Ack}) ->
+    encode_forward(Messages, Type, Ack);
+encode(?TYPE_HEARTBEAT = Type, _, _) ->
+    <<Type:4, 0:4>>.
+
+-dialyzer({nowarn_function, parse/3}).
+parse(handshake, <<_:4, _:4, Version:8>>, _) ->
+    {ok, #{type => handshake, version => Version}};
+parse(forward, Bin, _) ->
+    parse_forward(Bin);
+parse(heartbeat, <<_:4, _:4>>, _) ->
+    {ok, #{type => heartbeat}}.
+
+%%-------------------------------------------------------------------
+%%% Internal functions
+%%-------------------------------------------------------------------
+encode_forward(Messages, Type, Ack) ->
+    AckVal = ?B2I(Ack),
+    Data = marshaller(Messages),
+    Len = erlang:byte_size(Data),
+    LenVal = serialize_variable_byte_integer(Len),
+    <<Type:4, AckVal:4, LenVal/binary, Data/binary>>.
+
+parse_forward(<<_:4, AckVal:4, Bin/binary>>) ->
+    case parse_variable_byte_integer(Bin) of
+        {ok, Len, Rest} ->
+            <<MsgBin:Len/binary, _/binary>> = Rest,
+            {ok, #{
+                type => forward,
+                ack => ?I2B(AckVal),
+                messages => emqx_bridge_syskeeper_frame:marshaller(MsgBin)
+            }};
+        Error ->
+            Error
+    end.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_proxy.erl

@@ -0,0 +1,100 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_bridge_syskeeper_proxy).
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("emqx_bridge/include/emqx_bridge.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-export([
+    conn_bridge_examples/1,
+    values/1
+]).
+
+-export([
+    namespace/0,
+    roots/0,
+    fields/1,
+    desc/1
+]).
+
+-define(SYSKEEPER_HOST_OPTIONS, #{
+    default_port => 9092
+}).
+
+%% -------------------------------------------------------------------------------------------------
+%% api
+conn_bridge_examples(Method) ->
+    [
+        #{
+            <<"syskeeper_proxy">> => #{
+                summary => <<"Syskeeper Bridge Proxy">>,
+                value => values(Method)
+            }
+        }
+    ].
+
+values(_Method) ->
+    #{
+        enable => true,
+        type => syskeeper_proxy,
+        name => <<"foo">>,
+        listen => <<"127.0.0.1:9092">>,
+        acceptors => 16,
+        handshake_timeout => <<"16s">>
+    }.
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon Schema Definitions
+namespace() -> "bridge_syskeeper_proxy".
+
+roots() -> [].
+
+fields("config") ->
+    [
+        {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
+        {listen, listen()},
+        {acceptors,
+            mk(
+                non_neg_integer(),
+                #{desc => ?DESC("acceptors"), default => 16}
+            )},
+        {handshake_timeout,
+            mk(
+                emqx_schema:timeout_duration_ms(),
+                #{desc => ?DESC(handshake_timeout), default => <<"10s">>}
+            )}
+    ];
+fields("creation_opts") ->
+    emqx_resource_schema:create_opts([{worker_pool_size, #{default => 1}}]);
+fields("post") ->
+    [type_field(), name_field() | fields("config")];
+fields("put") ->
+    fields("config");
+fields("get") ->
+    emqx_bridge_schema:status_fields() ++ fields("post").
+
+desc("config") ->
+    ?DESC("desc_config");
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+    ["Configuration for Syskeeper Proxy using `", string:to_upper(Method), "` method."];
+desc("creation_opts" = Name) ->
+    emqx_resource_schema:desc(Name);
+desc(_) ->
+    undefined.
+
+listen() ->
+    Meta = #{desc => ?DESC("listen")},
+    emqx_schema:servers_sc(Meta, ?SYSKEEPER_HOST_OPTIONS).
+
+%% -------------------------------------------------------------------------------------------------
+
+type_field() ->
+    {type, mk(enum([syskeeper_proxy]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+    {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.

apps/emqx_bridge_syskeeper/src/emqx_bridge_syskeeper_proxy_server.erl

@@ -0,0 +1,251 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_bridge_syskeeper_proxy_server).
+
+-behaviour(gen_statem).
+
+-include_lib("emqx/include/logger.hrl").
+
+-elvis([{elvis_style, invalid_dynamic_call, disable}]).
+
+%% `emqx_resource' API
+-export([
+    query_mode/1,
+    on_start/2,
+    on_stop/2,
+    on_get_status/2
+]).
+
+%% API
+-export([start_link/3]).
+
+%% gen_statem callbacks
+-export([callback_mode/0, init/1, terminate/3, code_change/4]).
+-export([handle_event/4]).
+
+-type state() :: wait_ready | handshake | running.
+-type data() :: #{
+    transport := atom(),
+    socket := inet:socket(),
+    frame_state :=
+        undefined
+        | emqx_bridge_sysk_frame:state(),
+    buffer := binary(),
+    conf := map()
+}.
+
+-define(DEFAULT_PORT, 9092).
+
+%% -------------------------------------------------------------------------------------------------
+%% emqx_resource
+
+query_mode(_) ->
+    no_queries.
+
+on_start(
+    InstanceId,
+    #{
+        listen := Server,
+        acceptors := Acceptors
+    } = Config
+) ->
+    ?SLOG(info, #{
+        msg => "starting_syskeeper_connector",
+        connector => InstanceId,
+        config => Config
+    }),
+
+    #{hostname := Host, port := Port} = emqx_schema:parse_server(Server, #{
+        default_port => ?DEFAULT_PORT
+    }),
+    ListenOn = {Host, Port},
+
+    Options = [
+        {acceptors, Acceptors},
+        {tcp_options, [{mode, binary}, {reuseaddr, true}, {nodelay, true}]}
+    ],
+    MFArgs = {?MODULE, start_link, [maps:with([handshake_timeout], Config)]},
+    ok = emqx_resource:allocate_resource(InstanceId, listen_on, ListenOn),
+
+    case esockd:open(?MODULE, ListenOn, Options, MFArgs) of
+        {ok, _} ->
+            {ok, #{listen_on => ListenOn}};
+        Error ->
+            Error
+    end.
+
+on_stop(InstanceId, _State) ->
+    ?SLOG(info, #{
+        msg => "stopping_syskeeper_connector",
+        connector => InstanceId
+    }),
+    case emqx_resource:get_allocated_resources(InstanceId) of
+        #{listen_on := ListenOn} ->
+            esockd:close(?MODULE, ListenOn);
+        _ ->
+            ok
+    end.
+
+on_get_status(_InstanceId, #{listen_on := ListenOn}) ->
+    try
+        _ = esockd:listener({?MODULE, ListenOn}),
+        connected
+    catch
+        _:_ ->
+            disconnected
+    end.
+
+%% -------------------------------------------------------------------------------------------------
+-spec start_link(atom(), inet:socket(), map()) ->
+    {ok, Pid :: pid()}
+    | ignore
+    | {error, Error :: term()}.
+start_link(Transport, Socket, Conf) ->
+    gen_statem:start_link(?MODULE, [Transport, Socket, Conf], []).
+
+%% -------------------------------------------------------------------------------------------------
+%% gen_statem callbacks
+
+-spec callback_mode() -> gen_statem:callback_mode_result().
+callback_mode() -> handle_event_function.
+
+%% -------------------------------------------------------------------------------------------------
+-spec init(Args :: term()) ->
+    gen_statem:init_result(term()).
+init([Transport, Socket, Conf]) ->
+    {ok, wait_ready,
+        #{
+            transport => Transport,
+            socket => Socket,
+            conf => Conf,
+            buffer => <<>>,
+            frame_state => undefined
+        },
+        {next_event, internal, wait_ready}}.
+
+handle_event(internal, wait_ready, wait_ready, Data) ->
+    wait_ready(Data);
+handle_event(state_timeout, handshake_timeout, handshake, _Data) ->
+    %%    ?LOG(error, "Handshake tiemout~n", []),
+    {stop, normal};
+handle_event(internal, try_parse, running, Data) ->
+    try_parse(running, Data);
+handle_event(info, {tcp, _Socket, Bin}, State, Data) ->
+    try_parse(State, combine_buffer(Bin, Data));
+handle_event(info, {tcp_closed, _}, _State, _Data) ->
+    {stop, normal};
+handle_event(info, {tcp_error, _, _Reason}, _State, _Data) ->
+    %%    ?LOG(warning, "TCP error, reason:~p~n", [Reason]),
+    {stop, normal};
+handle_event(_Event, _Content, _State, _Data) ->
+    %%    ?LOG(warning, "Unexpected event:~p, Context:~p, State:~p~n", [Event, Content, State]),
+    keep_state_and_data.
+
+-spec terminate(Reason :: term(), State :: state(), Data :: data()) ->
+    any().
+terminate(_Reason, _State, _Data) ->
+    ok.
+
+code_change(_OldVsn, State, Data, _Extra) ->
+    {ok, State, Data}.
+
+%% -------------------------------------------------------------------------------------------------
+%%% Internal functions
+send(#{transport := Transport, socket := Socket}, Bin) ->
+    Transport:send(Socket, Bin).
+
+ack(Data) ->
+    ack(Data, true).
+
+ack(Data, false) ->
+    send(Data, <<0>>);
+ack(Data, true) ->
+    send(Data, <<16#FF>>).
+
+wait_ready(
+    #{
+        transport := Transport,
+        socket := RawSocket,
+        conf := #{handshake_timeout := Timeout}
+    } =
+        Data
+) ->
+    case Transport:wait(RawSocket) of
+        {ok, Socket} ->
+            Transport:setopts(Socket, [{active, true}]),
+            {next_state, handshake,
+                Data#{
+                    socket => Socket,
+                    frame_state => undefined
+                },
+                {state_timeout, Timeout, handshake_timeout}};
+        {error, Reason} ->
+            ok = Transport:fast_close(RawSocket),
+            {stop, Reason}
+    end.
+
+combine_buffer(Bin, #{buffer := Buffer} = Data) ->
+    Data#{buffer := <<Buffer/binary, Bin/binary>>}.
+
+try_parse(State, #{buffer := Bin} = Data) ->
+    case emqx_bridge_syskeeper_frame:parse_variable_byte_integer(Bin) of
+        {ok, Len, Rest} ->
+            case Rest of
+                <<Payload:Len/binary, Rest2/binary>> ->
+                    Data2 = Data#{buffer := Rest2},
+                    Result = parse(Payload, Data2),
+                    handle_parse_result(Result, State, Data2);
+                _ ->
+                    {keep_state, Data}
+            end;
+        {error, incomplete} ->
+            {keep_state, Data};
+        {error, _Reason} ->
+            %%            ?LOG(warning, "Parse error, reason:~p, buffer:~p~n", [Reason, Bin]),
+            {stop, parse_error}
+    end.
+
+%% maybe handshake
+parse(Bin, #{frame_state := undefined}) ->
+    emqx_bridge_syskeeper_frame:parse_handshake(Bin);
+parse(Bin, #{frame_state := State}) ->
+    emqx_bridge_syskeeper_frame:parse(Bin, State).
+
+do_forward(Ack, Messages, Data) ->
+    lists:foreach(
+        fun(Message) ->
+            Msg = emqx_message:from_map(Message#{headers => #{}, extra => #{}}),
+            _ = emqx_broker:safe_publish(Msg)
+        end,
+        Messages
+    ),
+    case Ack of
+        true ->
+            ack(Data);
+        _ ->
+            ok
+    end.
+
+handle_parse_result({ok, Msg}, State, Data) ->
+    handle_packet(Msg, State, Data);
+handle_parse_result({error, _Reason} = Error, State, Data) ->
+    handle_parse_error(Error, State, #{buffer := _Bin} = Data),
+    %%    ?LOG(warning, "Parse error, state:~p, reason:~p, buffer:~p~n", [State, Reason, Bin]),
+    {stop, parse_error}.
+
+handle_parse_error(_, handshake, Data) ->
+    ack(Data, false);
+handle_parse_error(_, _, _) ->
+    ok.
+
+handle_packet({FrameState, _Shake}, handshake, Data) ->
+    ack(Data),
+    {next_state, running, Data#{frame_state := FrameState}, {next_event, internal, try_parse}};
+handle_packet(#{type := forward, ack := Ack, messages := Messages}, running, Data) ->
+    do_forward(Ack, Messages, Data),
+    try_parse(running, Data);
+handle_packet(#{type := heartbeat}, running, Data) ->
+    ack(Data),
+    try_parse(running, Data).

apps/emqx_machine/priv/reboot_lists.eterm

@@ -127,7 +127,8 @@
             emqx_dashboard_sso,
             emqx_audit,
             emqx_gateway_gbt32960,
-            emqx_gateway_ocpp
+            emqx_gateway_ocpp,
+            emqx_bridge_syskeeper
         ],
     %% must always be of type `load'
     ce_business_apps =>

mix.exs

@@ -217,7 +217,8 @@ defmodule EMQXUmbrella.MixProject do
       :emqx_dashboard_sso,
       :emqx_audit,
       :emqx_gateway_gbt32960,
-      :emqx_gateway_ocpp
+      :emqx_gateway_ocpp,
+      :emqx_bridge_syskeeper
     ])
   end
 

rebar.config.erl

@@ -113,6 +113,7 @@ is_community_umbrella_app("apps/emqx_dashboard_sso") -> false;
 is_community_umbrella_app("apps/emqx_audit") -> false;
 is_community_umbrella_app("apps/emqx_gateway_gbt32960") -> false;
 is_community_umbrella_app("apps/emqx_gateway_ocpp") -> false;
+is_community_umbrella_app("apps/emqx_bridge_syskeeper") -> false;
 is_community_umbrella_app(_) -> true.
 
 is_jq_supported() ->

rel/i18n/emqx_bridge_syskeeper.hocon

@@ -0,0 +1,45 @@
+emqx_bridge_syskeeper {
+
+config_enable.desc:
+"""Enable or disable this bridge"""
+
+config_enable.label:
+"""Enable Or Disable Bridge"""
+
+desc_config.desc:
+"""Configuration for a Syskeeper bridge"""
+
+desc_config.label:
+"""Syskeeper Bridge Configuration"""
+
+desc_name.desc:
+"""Bridge name."""
+
+desc_name.label:
+"""Bridge Name"""
+
+desc_type.desc:
+"""The Bridge Type"""
+
+desc_type.label:
+"""Bridge Type"""
+
+template.desc:
+"""Template"""
+
+template.label:
+"""Template"""
+
+target_topic.desc:
+"""The topic for the forwarded message"""
+
+target_topic.label:
+"""Target Topic"""
+
+target_qos.desc:
+"""The QoS for the forwarded message, -1 is for the original topic"""
+
+target_qos.label:
+"""Target QoS"""
+
+}

rel/i18n/emqx_bridge_syskeeper_connector.hocon

@@ -0,0 +1,21 @@
+emqx_bridge_syskeeper_connector {
+
+server.desc:
+"""The address of the Syskeeper proxy server"""
+
+server.label:
+"""Server"""
+
+ack_mode.desc:
+"""Specify whether the proxy server should reply with an acknowledgement for the message forwarding, can be:<br>- need_ack <br>- no_ack <br>"""
+
+ack_mode.label:
+"""Acknowledgement Mode"""
+
+ack_timeout.desc:
+"""The maximum time to wait for an acknowledgement from the proxy server"""
+
+ack_timeout.label:
+"""Acknowledgement Timeout"""
+
+}

rel/i18n/emqx_bridge_syskeeper_proxy.hocon

@@ -0,0 +1,45 @@
+emqx_bridge_syskeeper_proxy {
+
+config_enable.desc:
+"""Enable or disable this bridge"""
+
+config_enable.label:
+"""Enable Or Disable Bridge"""
+
+desc_config.desc:
+"""Configuration for a Syskeeper proxy bridge"""
+
+desc_config.label:
+"""Syskeeper Proxy Bridge Configuration"""
+
+desc_name.desc:
+"""Bridge name"""
+
+desc_name.label:
+"""Bridge Name"""
+
+desc_type.desc:
+"""The Bridge Type"""
+
+desc_type.label:
+"""Bridge Type"""
+
+listen.desc:
+"""The listening address for this Syskeeper proxy server"""
+
+listen.label:
+"""Listen Address"""
+
+acceptors.desc:
+"""The number of the acceptors"""
+
+acceptors.label:
+"""Acceptors"""
+
+handshake_timeout.desc:
+"""The maximum to wait for the handshake when a connection is created"""
+
+handshake_timeout.label:
+"""Handshake Timeout"""
+
+}