kafka consumer example

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "host",
     "infra",
     "integrtion-tests",
+    "acl"
 ]
 
 [workspace.dependencies]

README.md

@@ -6,6 +6,7 @@ Web api application prototype written in rust
 
 - Clean architecture design with CQRS pattern
 - Crud api endpoints with actix-web and mongodb driver
+- Kafka Consumer implemented with `rdkafka` 
 - Integration tests
 - Workspaces usage
 - OpenApi and Swagger-UI
@@ -15,6 +16,7 @@ Web api application prototype written in rust
 
 ## Structure
 
+- `acl` - anti corruption layer implemented as kafka consumer
 - `app` - application layer with endpoints
 - `domain` - business logic contracts (queries/commands interfaces)
 - `domain_impl` - implementation of buisiness logic contracts

acl/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name="acl"
+version="0.0.1"
+edition="2021"
+
+[dependencies]
+log = "0.4.20"
+env_logger = "0.10.0"
+tokio = { version = "1.33.0", features = ["full"] }
+messaging = { path = "../messaging" }
+domain =  { path = "../domain" }

acl/Dockerfile

@@ -0,0 +1,15 @@
+# build image
+FROM rust:1.73.0-slim-buster as build
+WORKDIR /web-api
+COPY . .
+RUN apt-get update && apt-get install -y build-essential \
+     openssl libssl-dev \
+     zlib1g \
+     cmake
+
+RUN --mount=type=cache,target=/web-api/target cargo build --release && cp target/release/acl /acl
+
+# acl image
+FROM debian:buster-slim as acl
+COPY --from=build /acl /acl
+CMD ["/acl"]

acl/src/lib.rs

@@ -0,0 +1 @@
+

acl/src/main.rs

@@ -0,0 +1,19 @@
+use domain::events::ActivityEvent;
+use log::info;
+use messaging::kafka::{IKafkaFactory, KafkaConfig, KafkaFactory};
+use std::sync::Arc;
+
+#[tokio::main]
+async fn main() {
+    std::env::set_var("RUST_LOG", "info");
+    env_logger::init();
+
+    let config = KafkaConfig::new();
+    let kafka_factory: Arc<dyn IKafkaFactory<ActivityEvent>> = Arc::new(KafkaFactory::new(config));
+    kafka_factory
+        .create_consumer()
+        .consume(&|message| {
+            info!("message: '{:?}'", message);
+        })
+        .await;
+}

app/Cargo.toml

@@ -8,4 +8,5 @@ actix-web = "4.4.0"
 serde = "1.0.188"
 serde_derive = "1.0.188"
 domain = { path = "../domain" }
+messaging = { path = "../messaging" }
 utoipa = { version = "4", features = ["actix_extras"] }

app/src/endpoints.rs

@@ -1,7 +1,11 @@
-use crate::models::{CreateUserRequest, CreatedUserIdResponse, ErrorResponse, UserResponse};
+use crate::models::TrackActivityRequest::{Click, Open};
+use crate::models::{
+    CreateUserRequest, CreatedUserIdResponse, ErrorResponse, TrackActivityRequest, UserResponse,
+};
 use actix_web::web::{Data, Json, Path};
 use actix_web::{get, post, HttpResponse};
 use domain::commands::{CreateUserCommand, ICommandHandler};
+use domain::events::{ActivityEvent, IActivityTracker};
 use domain::queries::{GetUserQuery, IQueryHandler, User};
 
 /// Get user
@@ -65,3 +69,27 @@ pub async fn create_user(
         None => HttpResponse::BadRequest().json(ErrorResponse { code: 102 }),
     }
 }
+
+/// Track user activity
+///
+/// Track user activity
+#[utoipa::path(
+    request_body = TrackActivityRequest,
+    responses(
+        (status = 200, description = "Activity successfully tracked"),
+    )
+)]
+#[post("/track_activity")]
+pub async fn track_activity(
+    tracker: Data<dyn IActivityTracker>,
+    json: Json<TrackActivityRequest>,
+) -> HttpResponse {
+    let request = json.into_inner();
+
+    match request {
+        Click { x, y } => tracker.track(&ActivityEvent::Click { x, y }).await,
+        Open { path } => tracker.track(&ActivityEvent::Open { p: path }).await,
+    }
+
+    HttpResponse::Ok().json(())
+}

app/src/models.rs

@@ -23,3 +23,11 @@ pub struct UserResponse {
 pub struct ErrorResponse {
     pub code: i32,
 }
+
+#[derive(Serialize, Deserialize, ToSchema)]
+pub enum TrackActivityRequest {
+    #[serde(rename = "click")]
+    Click { x: i32, y: i32 },
+    #[serde(rename = "open")]
+    Open { path: String },
+}

build_images.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+docker-compose build