[feat] Add a channel filter on the WS integration. Progress for #322

docs/modules/user-guide/pages/integration-ws.adoc

@@ -1,7 +1,15 @@
 = WebSocket integration
 
-The WebSocket integration allows consuming device events using a simple WebSockets based protocol.Events are
-encoded as CloudEvents.
+The WebSocket integration allows consuming device events using a simple WebSockets based protocol.
+Events are encoded as CloudEvents.
+
+The application you want to consume must be passed as a path parameter.
+As an example, here is the url to connect to if you want to stream events for the application `example-app` :
+
+[source]
+----
+wss://ws-integration.sandbox.drogue.cloud/example-app
+----
 
 == Authentication
 
@@ -27,18 +35,29 @@ The bearer token must be passed with a key named `RefreshAccessToken`.
 
 
 Here is an example:
-```json
+[json]
+----
 {
    "RefreshAccessToken":"eyJhbGciO[...]sRqg7xt_jzb_tg"
 }
-```
+----
+
+== Channel filter
+
+You can set an additional path segment after the application ID to select a specific channel you want to consume the events from.
+
+For example, if you want to consume events only from the `sensor` channel you would connect to :
+[source]
+----
+wss://ws-integration.sandbox.drogue.cloud/example-app/sensor
+----
 
 == Consumer Group ID
 
 You can set the consumer group ID by providing a query parameter named `group_id`. This will set the Kafka consumer
-group ID to the provided value, which allows to share the load between different consumers.
+group ID to the provided value, allowing to share the load between different consumers.
 
 If no consumer group id is provided, a temporary one will be provided.
 
 NOTE: Using a temporary consumer group might lead to missed events during re-connects. If that is a problem for your
-use case, you need to provide a stable group id.
+use case, you need to provide a stable group id.

websocket-integration/Cargo.toml

@@ -11,6 +11,7 @@ actix = "0.13.0"
 actix-web-actors = "4"
 anyhow = "1"
 chrono = "0.4"
+cloudevents-sdk = { version = "0.5"}
 dotenv = "0.15"
 drogue-client = "0.11.0-alpha.1"
 env_logger = "0.9"

websocket-integration/src/lib.rs

@@ -97,6 +97,10 @@ pub async fn run(config: Config, startup: &mut dyn Startup) -> anyhow::Result<()
                     authenticator.clone(),
                     user_auth.clone().map(pat::Authenticator::new),
                 )))
+                .service(
+                    web::resource("/{channel}")
+                        .route(web::get().to(route::start_connection_with_channel_filter)),
+                )
                 .service(web::resource("").route(web::get().to(route::start_connection))),
         );
     })

websocket-integration/src/messages.rs

@@ -1,4 +1,5 @@
 use actix::prelude::{Message, Recipient};
+use cloudevents::Event;
 use drogue_client::integration::ws::v1::client;
 use drogue_cloud_integration_common::stream::EventStream;
 use drogue_cloud_service_common::error::ServiceError;
@@ -7,7 +8,7 @@ use uuid::Uuid;
 // Service sends the kafka events in this message to WSHandler
 #[derive(Message)]
 #[rtype(result = "()")]
-pub struct WsEvent(pub String);
+pub struct WsEvent(pub Event);
 
 // WsHandler sends this to service to subscribe to the stream
 #[derive(Message)]

websocket-integration/src/route.rs

@@ -7,6 +7,8 @@ use actix_web::{
 use actix_web_actors::ws;
 use drogue_cloud_service_api::webapp as actix_web;
 use drogue_cloud_service_common::actix_auth::authentication::AuthenticatedUntil;
+use drogue_cloud_service_common::client::UserAuthClient;
+use drogue_cloud_service_common::openid::Authenticator;
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
@@ -23,7 +25,6 @@ pub async fn start_connection(
     auth_expiration: Option<web::ReqData<AuthenticatedUntil>>,
 ) -> Result<HttpResponse, Error> {
     let application = application.into_inner();
-    let auth_expiration = auth_expiration.map(|e| e.into_inner().0);
 
     let authenticator = req.app_data().cloned();
     let user_auth = req.app_data().cloned();
@@ -34,10 +35,69 @@ pub async fn start_connection(
         user_auth.is_some()
     );
 
+    start_websocket(
+        req,
+        stream,
+        application,
+        None,
+        service_addr,
+        group_id.group_id,
+        auth_expiration,
+        authenticator,
+        user_auth,
+    )
+}
+
+pub async fn start_connection_with_channel_filter(
+    req: HttpRequest,
+    stream: Payload,
+    params: web::Path<(String, String)>,
+    service_addr: web::Data<Addr<Service>>,
+    web::Query(group_id): web::Query<GroupId>,
+    auth_expiration: Option<web::ReqData<AuthenticatedUntil>>,
+) -> Result<HttpResponse, Error> {
+    let (application, channel) = params.into_inner();
+
+    let authenticator = req.app_data().cloned();
+    let user_auth = req.app_data().cloned();
+
+    log::debug!(
+        "Auth state - authenticator: {}, userAuth: {}",
+        authenticator.is_some(),
+        user_auth.is_some()
+    );
+
+    start_websocket(
+        req,
+        stream,
+        application,
+        Some(channel),
+        service_addr,
+        group_id.group_id,
+        auth_expiration,
+        authenticator,
+        user_auth,
+    )
+}
+
+fn start_websocket(
+    req: HttpRequest,
+    stream: Payload,
+    application: String,
+    channel: Option<String>,
+    service_addr: web::Data<Addr<Service>>,
+    group_id: Option<String>,
+    auth_expiration: Option<web::ReqData<AuthenticatedUntil>>,
+    authenticator: Option<Authenticator>,
+    user_auth: Option<UserAuthClient>,
+) -> Result<HttpResponse, Error> {
+    let auth_expiration = auth_expiration.map(|e| e.into_inner().0);
+
     // launch web socket actor
     let ws = WsHandler::new(
         application,
-        group_id.group_id,
+        group_id,
+        channel,
         service_addr.get_ref().clone(),
         auth_expiration,
         authenticator,

websocket-integration/src/service.rs

@@ -170,10 +170,8 @@ impl Service {
         while let Some(event) = stream.next().await {
             log::debug!("Topic: {} - Event: {:?}", application, event);
 
-            // Convert the event to a JSON string
-            let event = serde_json::to_string(&event?)?;
             // Send the event as an Actor message
-            recipient.send(WsEvent(event.to_string())).await?;
+            recipient.send(WsEvent(event?)).await?;
 
             log::debug!("Sent message - go back to sleep");
         }

websocket-integration/src/wshandler/mod.rs

@@ -9,6 +9,7 @@ use actix::{
 };
 use actix_web_actors::ws::{self, CloseReason};
 use chrono::{DateTime, TimeZone, Utc};
+use cloudevents::AttributesReader;
 use drogue_client::{
     integration::ws::v1::client,
     user::{self, v1::authz},
@@ -83,6 +84,8 @@ pub struct WsHandler {
     application: String,
     /// the optional consumer group
     group_id: Option<String>,
+    /// the optional channel filter
+    channel: Option<String>,
     /// to exit the actor if the client was disconnected
     heartbeat: Instant,
     service_addr: Addr<Service>,
@@ -97,6 +100,7 @@ impl WsHandler {
     pub fn new(
         application: String,
         group_id: Option<String>,
+        channel: Option<String>,
         service_addr: Addr<Service>,
         auth_expiration: Option<DateTime<Utc>>,
         authenticator: Option<openid::Authenticator>,
@@ -116,6 +120,7 @@ impl WsHandler {
         WsHandler {
             application,
             group_id,
+            channel,
             heartbeat: Instant::now(),
             service_addr,
             id: Uuid::new_v4(),
@@ -274,7 +279,17 @@ impl Handler<WsEvent> for WsHandler {
     type Result = ();
 
     fn handle(&mut self, msg: WsEvent, ctx: &mut Self::Context) {
-        ctx.text(msg.0);
+        if let Some(channel) = &self.channel {
+            if msg.0.subject() != Some(channel.as_str()) {
+                return;
+            }
+        }
+
+        // Convert the event to a JSON string
+        match serde_json::to_string(&msg.0) {
+            Ok(evt) => ctx.text(evt),
+            Err(e) => log::warn!("Could not deserialize event : {e}"),
+        }
     }
 }