The astarte-message-hub from astarte-platform

Separate the publisher and subscriber from the Astarte handler

After this PR is merged #202

The publisher and subscriber are passed in the AstarteHandle, this will not make it possible to restart the MessageHub if the thread that handles the MQTT event loop crashes.

Bump Astarte Message Hub Proto to v0.6.0

Depends on https://github.com/astarte-platform/astarte-message-hub-proto/milestone/1

Caching nodes Introspection

Currently all node interfaces have been received during the attach procedure without being cached or persisted, so the message-hub should internally keep all interfaces until the detach method is called by the node..
This can improve in case of astarte-message restart, it already having the introspection will be able to receive the message from an Astarte instance without waiting for the interfaces to be sent froms nodes.

Implement message hub runner in the main file

The main.rs function does not yet contain anything except the initialization of the MessageHubOptions.
The initialization of the AstarteMessageHub should be added together with a runner for the MessageHubServer.

Implement Attach Method

Implement new AstarteMessage API

astarte-platform/astarte-message-hub-proto#7

Send `all_properties` received by message-hub to a node was attached

Implement Detach Method

Missing initializer for the Astarte handler

The astarte handler is missing a new() function. It furthermore only exposes as public the device_sdk field.

Implement methods to obtain properties of the interfaces

Missing clippy linter in CI

No worflow/job is present in the CI to perform linting using clippy.

Expose mqtt `keep_alive` to config file

Add to the message-hub config file the configuration of number of seconds after which client should ping the broker if there is no other data exchange.

Add to MessageHubOptions struct:

astarte-message-hub/src/config/mod.rs

Line 52 in e673860

pub struct MessageHubOptions {
Pass to Astarte Device SDK Options: https://github.com/astarte-platform/astarte-message-hub/blob/e673860701ca42a2e663d9c1be279669b1b872fc/src/main.rs#L93C31-L93C45

Possible names:
mqtt_keep_alive
mqtt_connection_keep_alive
...

[NodeId Metadata]: Update the architecture file

Write some lines about the NodeId Metadata that a Node have to send during the GRPC connection.
Update the sequence diagrams with the new metadata (optional).

Missing docstrings

Most of the sources are lacking docstrings.
It would be useful to add short comments for each function/struct.

Implement Type conversion for AstarteDataType (proto)

Add configuration example to Readme

Add something like this https://github.com/edgehog-device-manager/edgehog-device-runtime#configuration

Missing port for the protobuf server

The Astarte message hub will eventually run a protobuf server (once #75 is completed).
However, there is no way of setting a specific port to use when communicating with the protobuf server.
It should be evaluated if a dynamic configuration of the message hub protobuf server port is required.

Clearer files/structs names

Some files, modules, and functions could be renamed to have clearer names.
For example:

The struct Astarte could be renamed to AstarteHandler
The file astarte_provider.rs could be renamed to astarte_handler.rs
The file mock_astarte.rs could be renamed to mock_astarte_sdk.rs

Add Astarte Device Rust SDK

#27
#28

NodeId via GRPC Metadata

Metadata is a side channel that allows clients and servers to provide information to each other that is associated with an RPC.

gRPC metadata is a key-value pair of data that is sent with initial or final gRPC requests or responses. It is used to provide additional information about the call, such as authentication credentials, tracing information, or custom headers.

Try to figure out if it is possible to move the node id from the Attach message to the GRPC metadata.

https://grpc.io/docs/guides/metadata/#:~:text=gRPC%20metadata%20can%20be%20sent,when%20it%20closes%20an%20RPC.

astarte-platform/astarte-device-sdk-rust#292

Add Statistics about nodes attached to the Message Hub

It would be nice to have an Astarte interface that collects statistics from the nodes attached to the Message Hub.
An interface name would be : org.astarte-paltform.messagehub.NodeStats

Mapping:

number of messages(sent/received) on specific interface : /$node_id/$interface_name/messages

Adding a layer on AstarteMessage to enrich the API

Currently AstarteMessage can only send/receive AstarteData, so it is better to add a layer on top of AstarteMessage giving the possibility to add other fields in AstarteDataMessage, without introduce a breaking changes in the API.
Other fields can be:

an error during the send or receive data.
an event from Astarte for a specific node ( for instance , an interface with higher major was added in the introscpection)
etc.
Rename astarte_message.proto to astarte_data_message.proto .
Add AstarteDataMessge.

message AstarteDataMessage{
     AstarteMessage astarte_message = 1;
     AstarteFailure/AstarteError? = 2;
     ...
}

Implement Type conversion for Astarte Device SDK Type

Missing calls to Astarte handler `run` function

The Astarte handler run function is not called within the AstarteMessageHub.
I propose two different approaches:

Exposing a run function for the AstarteMessageHub, containing a call to the run function of the Astarte handler. This function however would need to be called in a separate thread due to its blocking nature. This would require the clone trait to be implemented for the AstarteMessageHub struct.
Starting a separate tokio task within the initialization function of the AstarteMessageHub in which to call the run function.

Save toml-config in writable path

When the configuration of Astarte-message-hub was provided via HTTP or GRPC-PROTOBUF APIs , the toml file generated by MessageHubOptions must be saved in a writable path, because the default path configured in /etc or the local path of message-hub may not be writable, so it should be saved in the store_directory provided by the MessageHubOptions.

Create a separate repository for the message hub proto definitions

This will let us share the protobuf definitions as a git submodule between the rust and python SDK

Implement API for supporting dynamic introspection

ref astarte-platform/astarte-message-hub-proto#8

Storing messages for a node that goes offline

If a node goes offline without calling the detach method, all messages directed to it should be stored and on return sent back to the node.

sequenceDiagram
    Node-A->>+MessageHub: Attach
    
    Node-A->>+Node-A: go offline
    Astarte->>+MessageHub: Send data
    alt check Node-A is online
        MessageHub->>MessageHub: Store message for Node-A
    end
    
    Node-A-->>-Node-A: back online
    Node-A->>+MessageHub: Attach
    MessageHub->>+Node-A: Send data

No message is logged upon failure of SDK `handle_events`

The result failed of calls handle_events is not logged in main run function.

astarte-message-hub/src/data/astarte_handler.rs

Lines 193 to 197 in e880fd6

    
           async fn run(&mut self) { 
        
               use crate::proto_message_hub::AstarteMessage; 
        
               if let Ok(astarte_data_event) = self.device_sdk.handle_events().await { 
        
                   println!("incoming: {:?}", astarte_data_event);

Add an API for supporting dynamic introspection

Currently a messageHub node can add the astarte interfaces only calling the Attach method, it should be better add a separated method.

Avoid send message from a node not attached to Message Hub

Currently the message hub allows to send a message from node even if it is not attached to the message hub, this should be avoided.

Solution: Check node-id before send the message.

Add `store_directory` to MessageHubOptions

Directory used by Astarte-Message-Hub to retain persistent data

Make it configurable

Make it configurable, the Astarte Message Hub should be support different configuration ways ( config.toml, API Rest, proto method etc )

Bump Astarte Device SDK Rust to 0.8

Licensing refers to edgehog

Configuration files licensing headers refer to Edgehog instead of Astarte.
The files are file.rs, http.rs, protobuf.rs, mod.rs.

astarte-message-hub/src/config/http.rs

Line 2 in b870447

* This file is part of Edgehog.

astarte-message-hub/src/config/file.rs

Line 2 in b870447

* This file is part of Edgehog.

astarte-message-hub/src/config/mod.rs

Line 2 in b870447

* This file is part of Edgehog.

astarte-message-hub/src/config/protobuf.rs

Line 2 in b870447

* This file is part of Edgehog.

Support hardware id from dbus

Retrieve the hardware id form the dbus.
Use the hardware id retrieved from the dbus to generate a valid Astarte device id.

Add end to end test

gRPC service definition

Specifying the methods that can be called with their parameters and return types.

Convert project to Cargo Workspace

Split the package into multiple library crates to separate astarte-message-hub-types from the astarte-message-hub application allowing separate distribution of the astarte-message-hub-types crate.

Cargo offers a feature called workspaces that can help manage multiple related packages that are developed in tandem.
A workspace is a set of packages that share the same Cargo.lock and output directory.
Cargo-Workspace

Add Authentication and Security Layer

Implement Send Method

Bump Astarte Device SDK to v0.7

Write library to wrap Astarte SDK

Write simple library to wrap Astarte SDK, use something like trait Publisher that we are using in Edgehog Device Runtime.

Example of Astarte traits for wrapping Astarte SDK Device

#[async_trait]
pub trait AstartePublisher: Send + Sync {
    async fn send_object<T: 'static>(
        &self,
        interface_name: &str,
        interface_path: &str,
        data: T,
    ) -> Result<(), AstarteError>
    where
        T: serde::Serialize + Send;
    async fn send(
        &self,
        interface_name: &str,
        interface_path: &str,
        data: AstarteType,
    ) -> Result<(), AstarteError>;
}

#[async_trait]
pub trait AstarteSubscriber {
    async fn subscribe(
        &self,
        astarte_node: &AstarteNode,
    ) -> Result<Receiver<Result<AstarteMessage, Status>>, AstarteError>;
    async fn unsubscribe(&self, astarte_node: AstarteNode) -> Result<(), AstarteError>;
}

https://github.com/harlem88/astarte-message-hub/blob/5d7268c94e785d44828dfeefdf8f23385b474dd4/src/data/mod.rs#L32

Add Ack signaling when the Astarte Message Hub send a message to the node

Currently the Astarte message hub doesn't support an ack mechanism to signal the exact receipt of a message by a node.
For add the ack support the message-hub needs:

Add an ID to identify each message sent from Message-hub.
An ack method ack_message(message_id) to send the ack signal from the node to Astarte message hub.
Add a configuration on the node to specify whether it supports ack signaling or whether the message-hub should consider that all messages have been successfully received (the default shoud be the second option) .

All node_id attached to the Message Hub, maybe a settable/unsettable boolean property like /${node_id}/attached.
The interfaces of each node, it would be a settable/unsettable strings array like /${node_id}/interfaces

Persist Astarte credentials secret after register

If no credential secret is present, the Astarte message hub register a new device, this operation return a credentials secret should be preserved even after restart or reboot.

astarte-message-hub/src/main.rs

Line 42 in 59d4b0a

    
           // If no credential secret is present, register a new device using the Astarte device SDK

	async fn run(&mut self) {
	use crate::proto_message_hub::AstarteMessage;

	if let Ok(astarte_data_event) = self.device_sdk.handle_events().await {
	println!("incoming: {:?}", astarte_data_event);

astarte-platform / astarte-message-hub Goto Github PK

astarte-message-hub's People

Contributors

Stargazers

Watchers

Forkers

astarte-message-hub's Issues

Recommend Projects

Recommend Topics

Recommend Org