Merge pull request #585 from swimos/join-map-lane

Adds support for join map lanes.

Author: SirCipher

api/swim_api/src/downlink/mod.rs

@@ -32,6 +32,8 @@ pub enum DownlinkKind {
     Event,
     /// Accepts key-value pairs and maintains a state as a map.
     Map,
+    /// Accepts map updates but does not maintain any state.
+    MapEvent,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]

api/swim_api/src/protocol/map/mod.rs

@@ -456,9 +456,9 @@ impl<K: StructuralWritable, V: StructuralWritable> Encoder<MapOperation<K, V>>
     }
 }
 
-/// Repreesentation of map lane messages (used to form the body of Recon messages when operating)
+/// Representation of map lane messages (used to form the body of Recon messages when operating)
 /// on downlinks. This extends [`MapOperation`] with `Take` (retain the first `n` items) and `Drop`
-/// (remove teh first `n` items). We never use these internally but must support them for communicating
+/// (remove the first `n` items). We never use these internally but must support them for communicating
 /// with other implementations.
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Form, Hash)]
 #[form_root(::swim_form)]

client/runtime/src/models.rs

@@ -163,7 +163,7 @@ impl DownlinkRuntime {
                     );
                     runtime.run().await;
                 }
-                DownlinkKind::Map => {
+                DownlinkKind::Map | DownlinkKind::MapEvent => {
                     let strategy = if config.abort_on_bad_frames {
                         ReportStrategy::new(AlwaysAbortStrategy).boxed()
                     } else {

docs/agent.md

@@ -35,6 +35,7 @@ Contents
     * `ValueLane` events.
     * `MapLane` events.
     * `JoinValueLane` events.
+    * `JoinMapLane` events.
     * `HttpLane` events.
     * Store events.
     * Borrowing from lifecycles.

docs/define.md

@@ -27,6 +27,7 @@ The derive macro for `AgentLaneModel` can be applied to any struct type where al
 * Demand Map lanes: `swim::agent::lanes::DemandMapLane`.
 * Map Lanes: `swim::agent::lanes::MapLane`.
 * Join Value Lanes: `swim::agent::lanes::JoinValueLane`.
+* Join Map Lanes: `swim::agent::lanes::JoinMapLane`.
 * HTTP Lanes: `swim::agent::lanes::HttpLane` (or the shorthand `swim::agent::lanes::SimpleHttpLane`).
 
 The supported store types are:
@@ -50,12 +51,18 @@ struct ExampleAgent {
 
 As mentioned above, all of the type parameters used in the lane types must be `Send` (which is clearly true for `i32`, `String`, and `u64`). However, to use the derive macro there is a further restriction.
 
-For the macro to be able to generate the implementation, it needs to know how to serialize and deserialize the types use in the lane. This is encoded by the `swim::form::Form` trait which is covered in the following section. Additionally, for a map-like item (`MapLane<K, V>`, `MapStore<K, V>` or `JoinValueLane<K, V>`) the key type `K` must additionally satisfy:
+For the macro to be able to generate the implementation, it needs to know how to serialize and deserialize the types use in the lane. This is encoded by the `swim::form::Form` trait which is covered in the following section. Additionally, for a map-like item (`MapLane<K, V>`, `MapStore<K, V>`, `JoinValueLane<K, V>`, `JoinMapLane<L, K, V>`) the key type `K` must additionally satisfy:
 
 ```rust
 K: Eq + Hash + Ord + Clone
 ```
 
+For `JoinMapLane`s, the link key type `L` must satisfy:
+
+```rust
+L: Eq + Hash + Clone
+```
+
 Stores are effectively private alternatives to lanes. The maintain state in exactly the same was as the corresponding lane types but are not exposed externally.
 
 The `Form` trait

docs/lifecycle.md

@@ -24,6 +24,7 @@ struct ExampleAgent {
     example_value: ValueLane<i32>,
     example_map: MapLane<String, i32>,
     example_join_value: JoinValueLane<String, i32>,
+    example_join_map: JoinMapLane<String, String, i32>,
     example_http: SimpleHttpLane<String>,
 }
 ```
@@ -212,7 +213,7 @@ Additionally, it is possible to attach events to the downlinks used for each key
 fn register_lifecycle(
     &self,
     context: JoinValueContext<ExampleAgent, String, i32>,
-) -> impl JoinValueLaneLifecycle<i32, Text, ExampleAgent> + 'static {
+) -> impl JoinValueLaneLifecycle<String, i32, ExampleAgent> + 'static {
     context.builder()
         .on_linked(|context, key, address| context.effect(move || {
             println!("Linked downlink for key: {} from lane at: {}", key, address);
@@ -246,6 +247,52 @@ The `on_unlinked` and `on_failed` handlers return a `LinkClosedResponse`. This i
 
 It is possible to add a shared state to a join value lifecycle in a similar way to other downlinks types (see [Downlinks](downlink.md)). Note that this state is shared between the event handlers of a single instance of the lifecycle and not between all instances. If you have state that needs to be shared between all instances it must be stored inside an `Arc`.
 
+Join map lane events
+--------------------
+
+A join map lane supports all of the events supported by a map lane (`on_update`, `on_remove` and `on_clear`).
+
+Additionally, it is possible to attach events to the downlinks used for each key of the join value lane. To do this, add the following to the agent lifecycle implementation:
+
+```rust
+#[join_map_lifecycle(example_join_map)]
+fn register_lifecycle(
+    &self,
+    context: JoinMapContext<ExampleAgent, String, String, i32>,
+) -> impl JoinMapLaneLifecycle<String, String, ExampleAgent> + 'static {
+    context.builder()
+        .on_linked(|context, link_key, address| context.effect(move || {
+            println!("Linked downlink for link: {} from lane at: {}", key, address);
+        }))
+        .done()
+}
+```
+
+An instance of the lifecycle will be created for each link that is attached to the lane so it is necessary that the lifecycle is `Clone` and has a `static` lifetime.
+
+Supported handlers are `on_linked`, `on_synced`, `on_unlinked` and `on_failed`. These take closures with the following signatures:
+
+1. `on_linked`: 
+```rust
+(context: HandlerContext<ExampleAgent>, link_key: L, address: Address<&str>) -> impl EventHandler<ExampleAgent>
+```
+2. `on_synced`:
+```rust
+(context: HandlerContext<ExampleAgent>, link_key: L, address: Address<&str>, keys: &HashSet<K>) -> impl EventHandler<ExampleAgent>
+```
+3. `on_unlinked` and `on_failed`:
+```rust
+(context: HandlerContext<ExampleAgent>, link_key: L, address: Address<&str>, keys: HashSet<K>) -> impl HandlerAction<ExampleAgent, Completion = LinkClosedResponse>
+```
+
+The `on_unlinked` and `on_failed` handlers return a `LinkClosedResponse`. This is an enumeration with the values `Retry`, `Abandon` and `Delete`. These have the following effects:
+
+1. `Retry`: An attempt will be made to reconnect the link (TODO: this is not implemented yet.)
+2. `Abandon`: The link will be abandoned by the entry will still remain in the map. (This is the default.)
+3. `Delete`: The link will be abandoned and the entry deleted from the map.
+
+It is possible to add a shared state to a join value lifecycle in a similar way to other downlinks types (see [Downlinks](downlink.md)). Note that this state is shared between the event handlers of a single instance of the lifecycle and not between all instances. If you have state that needs to be shared between all instances it must be stored inside an `Arc`.
+
 HTTP lane events
 ----------------
 

example_apps/map_store/src/agent.rs

@@ -84,13 +84,13 @@ impl ExampleLifecycle {
                 let key = name.clone();
                 context
                     .get_value(ExampleAgent::LANE)
-                    .and_then(move |v| context.update_store(ExampleAgent::SAVED, key, v))
+                    .and_then(move |v| context.update(ExampleAgent::SAVED, key, v))
                     .boxed()
             }
             Instruction::Restore(name) => {
                 let key = name.clone();
                 context
-                    .get_entry_store(ExampleAgent::SAVED, key)
+                    .get_entry(ExampleAgent::SAVED, key)
                     .map(|maybe: Option<i32>| maybe.unwrap_or_default())
                     .and_then(move |v| context.set_value(ExampleAgent::LANE, v))
                     .boxed()

example_apps/map_store_persistence/src/agent.rs

@@ -45,8 +45,8 @@ impl ExampleLifecycle {
     ) -> impl EventHandler<ExampleAgent> {
         join3(
             context.get_agent_uri(),
-            context.get_map_store(ExampleAgent::VALUE),
-            context.get_map_store(ExampleAgent::TEMPORARY),
+            context.get_map(ExampleAgent::VALUE),
+            context.get_map(ExampleAgent::TEMPORARY),
         )
         .and_then(move |(uri, m1, m2)| {
             context.effect(move || {
@@ -67,8 +67,8 @@ impl ExampleLifecycle {
     ) -> impl EventHandler<ExampleAgent> {
         join3(
             context.get_agent_uri(),
-            context.get_map_store(ExampleAgent::VALUE),
-            context.get_map_store(ExampleAgent::TEMPORARY),
+            context.get_map(ExampleAgent::VALUE),
+            context.get_map(ExampleAgent::TEMPORARY),
         )
         .and_then(move |(uri, m1, m2)| {
             context.effect(move || {
@@ -91,10 +91,10 @@ impl ExampleLifecycle {
         match cmd {
             Instruction::Wake => UnitHandler::default().boxed(),
             Instruction::SetValue { key, value } => context
-                .update_store(ExampleAgent::VALUE, key.clone(), *value)
+                .update(ExampleAgent::VALUE, key.clone(), *value)
                 .boxed(),
             Instruction::SetTemp { key, value } => context
-                .update_store(ExampleAgent::TEMPORARY, key.clone(), *value)
+                .update(ExampleAgent::TEMPORARY, key.clone(), *value)
                 .boxed(),
             Instruction::Stop => context.stop().boxed(),
         }

example_apps/transit/src/agents/mod.rs

@@ -18,7 +18,7 @@ use std::hash::Hash;
 use swim::{
     agent::{
         agent_lifecycle::utility::JoinValueContext,
-        lanes::join_value::{lifecycle::JoinValueLaneLifecycle, LinkClosedResponse},
+        lanes::{join_value::lifecycle::JoinValueLaneLifecycle, LinkClosedResponse},
     },
     form::Form,
 };

example_apps/value_store/src/agent.rs

@@ -82,10 +82,10 @@ impl ExampleLifecycle {
         match *cmd {
             Instruction::Save => context
                 .get_value(ExampleAgent::LANE)
-                .and_then(move |v| context.set_store_value(ExampleAgent::SAVED, v))
+                .and_then(move |v| context.set_value(ExampleAgent::SAVED, v))
                 .boxed(),
             Instruction::Restore => context
-                .get_store_value(ExampleAgent::SAVED)
+                .get_value(ExampleAgent::SAVED)
                 .and_then(move |v| context.set_value(ExampleAgent::LANE, v))
                 .boxed(),
         }