Merge pull request #693 from swimos/stocks

Simulated Stocks Demo

Author: SirCipher

Cargo.toml

@@ -40,7 +40,9 @@ members = [
     "example_apps/join_value",
     "example_apps/aggregations",
     "example_apps/time_series",
-    "example_apps/devguide", "example_apps/ripple",
+    "example_apps/devguide",
+    "example_apps/ripple",
+    "example_apps/stocks_simulated",
 ]
 
 [workspace.package]
@@ -172,4 +174,4 @@ waker-fn = "1.1.0"
 num = "0.4"
 smol_str = "0.2.0"
 http-body-util = "0.1.2"
-hyper-util = "0.1.5"
+hyper-util = "0.1.5"

example_apps/README.md

@@ -21,6 +21,7 @@ binaries for their corresponding client applications. Each directory contains de
 - [map_store_persistence](map_store_persistence): Application demonstrating Map Stores with persistence.
 - [ripple](ripple): Real-time synchronous shared multiplayer experience.
 - [supply_lane](supply_lane): Application demonstrating Supply Lanes.
+- [stocks_simulated](stocks_simulated): Application demonstrating simulated stock market data.
 - [time_series](time_series): Reference code for the [Time Series](https://www.swimos.org/server/rust/time-series/)
   guide.
 - [transit](transit): Rust port of the [Transit](https://github.com/swimos/transit) application.

example_apps/stocks_simulated/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "stocks_simulated"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+homepage.workspace = true
+
+[dependencies]
+swimos = { workspace = true, features = ["server", "agent"] }
+swimos_form = { workspace = true }
+example-util = { path = "../example_util" }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
+futures = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+rand = { workspace = true }

example_apps/stocks_simulated/src/agent.rs

@@ -0,0 +1,160 @@
+// Copyright 2015-2024 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use futures::stream::unfold;
+use futures::StreamExt;
+use rand::rngs::ThreadRng;
+use rand::Rng;
+use std::time::Duration;
+use swimos::agent::agent_lifecycle::HandlerContext;
+use swimos::agent::event_handler::{EventHandler, HandlerActionExt};
+use swimos::agent::lanes::{CommandLane, JoinValueLane, ValueLane};
+use swimos::agent::{lifecycle, projections, AgentLaneModel};
+use swimos::model::Timestamp;
+use swimos_form::Form;
+use tokio::time::sleep;
+use tracing::info;
+
+/// Agent for tracking all the stocks.
+#[projections]
+#[derive(AgentLaneModel)]
+pub struct SymbolsAgent {
+    /// Lane for receiving stocks to track.
+    add: CommandLane<String>,
+    /// Stock downlinks.
+    stocks: JoinValueLane<String, Stock>,
+}
+
+/// Symbols Agent Lifecycle.
+#[derive(Clone)]
+pub struct SymbolsLifecycle;
+
+#[lifecycle(SymbolsAgent)]
+impl SymbolsLifecycle {
+    /// Lifecycle event handler which is invoked exactly once when `add` lane receives a command to
+    /// add a downlink to `id`.
+    #[on_command(add)]
+    pub fn add(
+        &self,
+        context: HandlerContext<SymbolsAgent>,
+        id: &str,
+    ) -> impl EventHandler<SymbolsAgent> {
+        let node = format!("/stock/{id}");
+        context.add_downlink(
+            SymbolsAgent::STOCKS,
+            id.to_string(),
+            None,
+            node.as_str(),
+            "status",
+        )
+    }
+}
+
+/// Stock Agent implementation.
+#[projections]
+#[derive(AgentLaneModel)]
+pub struct StockAgent {
+    /// The state of the current stock.
+    status: ValueLane<Stock>,
+}
+
+/// Stock model.
+#[derive(Form, Clone)]
+pub struct Stock {
+    /// The timestamp that the stock was last updated.
+    timestamp: Timestamp,
+    /// The stock's current price.
+    price: f64,
+    /// The stock's current volume.
+    volume: f64,
+    /// The stock's current bid.
+    bid: f64,
+    /// The stock's current add.
+    ask: f64,
+    /// The stock's movement since it was last updated.
+    movement: f64,
+}
+
+impl Default for Stock {
+    fn default() -> Self {
+        Stock::select_random(0.0)
+    }
+}
+
+impl Stock {
+    /// Generates a stock. Tracking its movement since `previous_price`.
+    fn select_random(previous_price: f64) -> Stock {
+        let mut rng = ThreadRng::default();
+        let current_price = truncate(rng.gen());
+
+        Stock {
+            timestamp: Timestamp::now(),
+            price: current_price,
+            volume: truncate(rng.gen::<f64>() * 400.0).powi(3),
+            bid: truncate(rng.gen::<f64>() * 100.0),
+            ask: truncate(rng.gen::<f64>() * 100.0),
+            movement: current_price - previous_price,
+        }
+    }
+}
+
+/// Truncates `f` to two decimal places.
+fn truncate(f: f64) -> f64 {
+    (f * 100.0).round() / 100.0
+}
+
+/// Stock Agent's lifecycle implementation.
+#[derive(Clone)]
+pub struct StockLifecycle;
+
+#[lifecycle(StockAgent)]
+impl StockLifecycle {
+    /// Lifecycle event handler which is invoked exactly once when the agent starts.
+    ///
+    /// The handler will spawn a stream which will generate random stocks and send a command to the
+    /// `SymbolsAgent` to add a downlink to this agent.
+    #[on_start]
+    pub fn on_start(&self, context: HandlerContext<StockAgent>) -> impl EventHandler<StockAgent> {
+        let stream = unfold(Duration::default(), move |delay| async move {
+            sleep(delay).await;
+
+            let mut rng = ThreadRng::default();
+            let handler = generate_stock(context);
+            let next_delay = Duration::from_secs(rng.gen_range(5..=10));
+
+            Some((handler, next_delay))
+        });
+        context
+            .get_parameter("symbol")
+            .and_then(move |symbol_opt: Option<String>| {
+                let symbol = symbol_opt.expect("Missing symbol for stock");
+                context.send_command(None, "/symbols", "add", symbol)
+            })
+            .followed_by(context.suspend_schedule(stream.boxed()))
+            .followed_by(context.get_agent_uri())
+            .and_then(move |uri| context.effect(move || info!(%uri, "Started agent")))
+    }
+}
+
+/// Returns an event handler which will generate a new stock and set it to the `status` lane.
+fn generate_stock(context: HandlerContext<StockAgent>) -> impl EventHandler<StockAgent> {
+    context
+        .get_value(StockAgent::STATUS)
+        .and_then(move |previous_stock: Stock| {
+            context.set_value(
+                StockAgent::STATUS,
+                Stock::select_random(previous_stock.price),
+            )
+        })
+}

example_apps/stocks_simulated/src/main.rs

@@ -0,0 +1,113 @@
+// Copyright 2015-2024 Swim Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Rust port of the [Simulated Stock Demo](https://github.com/nstreamio/sim-stock-demo) application.
+//!
+//! Run the application using the following:
+//! ```shell
+//! $ cargo run --bin stocks_simulated
+//! ```
+//!
+//! Run the UI with the following:
+//! ```shell
+//! cd ui
+//! npm install
+//! npm run dev
+//! ```
+//! Then head to `localhost:5173` to see it in action.
+//! ```
+
+use std::str::FromStr;
+use std::{error::Error, time::Duration};
+
+use rand::{thread_rng, Rng};
+use tracing_subscriber::filter::LevelFilter;
+
+use swimos::route::RouteUri;
+use swimos::server::ServerHandle;
+use swimos::{
+    agent::agent_model::AgentModel,
+    route::RoutePattern,
+    server::{Server, ServerBuilder},
+};
+
+use crate::agent::{StockAgent, StockLifecycle, SymbolsAgent, SymbolsLifecycle};
+
+mod agent;
+
+const CHARSET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    example_logging()?;
+    run_swim_server().await?;
+
+    Ok(())
+}
+
+/// Spawns the Swim server on 0.0.0.0:9001.
+async fn run_swim_server() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let symbols_agent = AgentModel::new(SymbolsAgent::default, SymbolsLifecycle.into_lifecycle());
+    let stock_agent = AgentModel::new(StockAgent::default, StockLifecycle.into_lifecycle());
+
+    let swim_server = ServerBuilder::with_plane_name("Ripple Plane")
+        .set_bind_addr("0.0.0.0:9001".parse().unwrap())
+        .add_route(RoutePattern::parse_str("/symbols")?, symbols_agent)
+        .add_route(RoutePattern::parse_str("/stock/:symbol")?, stock_agent)
+        .update_config(|config| {
+            config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
+        })
+        .build()
+        .await?;
+
+    let (task, handle) = swim_server.run();
+    let task = tokio::spawn(task);
+
+    start_agents(&handle).await?;
+    task.await??;
+
+    println!("Server stopped");
+
+    Ok(())
+}
+
+async fn start_agents(handle: &ServerHandle) -> Result<(), Box<dyn Error + Send + Sync>> {
+    for _ in 0..20000 {
+        let stock_id = {
+            let mut rng = thread_rng();
+            (0..4)
+                .map(|_| {
+                    let idx = rng.gen_range(0..CHARSET.len());
+                    CHARSET[idx] as char
+                })
+                .collect::<String>()
+        };
+        handle
+            .start_agent(RouteUri::from_str(format!("/stock/{stock_id}").as_str())?)
+            .await?;
+    }
+    Ok(())
+}
+
+/// Enables logging if `--enable-logging` was provided.
+fn example_logging() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let args = std::env::args().collect::<Vec<_>>();
+    if args.get(1).map(String::as_str) == Some("--enable-logging") {
+        let filter = example_util::example_filter()?
+            .add_directive("stocks_simulated=trace".parse()?)
+            .add_directive(LevelFilter::WARN.into());
+        tracing_subscriber::fmt().with_env_filter(filter).init();
+    }
+    Ok(())
+}

example_apps/stocks_simulated/ui/.eslintrc.cjs

@@ -0,0 +1,14 @@
+module.exports = {
+  root: true,
+  env: { browser: true, es2020: true },
+  extends: ["eslint:recommended", "plugin:@typescript-eslint/recommended", "plugin:react-hooks/recommended"],
+  ignorePatterns: ["dist", ".eslintrc.cjs"],
+  parser: "@typescript-eslint/parser",
+  plugins: ["react-refresh"],
+  rules: {
+    "react-refresh/only-export-components": ["warn", { allowConstantExport: true }],
+    "@typescript-eslint/no-unnecessary-type-constraint": "warn",
+    "@typescript-eslint/no-explicit-any": "off",
+    "@typescript-eslint/no-unused-vars": "warn",
+  },
+};

example_apps/stocks_simulated/ui/.gitignore

@@ -0,0 +1,24 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

example_apps/stocks_simulated/ui/README.md

@@ -0,0 +1,10 @@
+# Stocks Demo
+
+This UI displays a real-time table of a simulated collection of stocks that one might find listed on an exchange. Their
+key properties such as current price and daily movement are simulated in the Rust Swim application located in the `/src`
+directory of this repository. A single MapDownlink is opened by the UI to the `stocks` lane of the backend application's
+`SymbolsAgent`. This downlink syncs with the lane's state containing pricing for all entries and then receives follow-on
+price updates until the downlink is closed. With each update received, UI local state and table content is updated.
+
+The UI in this folder was bootstrapped with React + TypeScript + Vite and uses `ag-grid-react` for its table component.
+

example_apps/stocks_simulated/ui/index.html

@@ -0,0 +1,14 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="UTF-8" />
+    <link rel="icon" type="image/png" href="/favicon.ico" />
+    <link href='https://fonts.googleapis.com/css?family=Open Sans' rel='stylesheet'>
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>NStream Stocks Demo</title>
+  </head>
+  <body>
+    <div id="root"></div>
+    <script type="module" src="/src/main.tsx"></script>
+  </body>
+</html>