feat: publish / subscribe for Linkable contracts (#391, stage 1)

aimdb-core/src/typed_api.rs

@@ -862,6 +862,7 @@ pub struct OutboundLinkBuilder<T: Send + Sync + 'static + Debug + Clone, R> {
     at: R,
     url: String,
     connector: Option<crate::ConnectorHandle>,
+    schema: Option<(String, u32)>,
     config: Vec<(String, String)>,
     context_serializer: Option<TypedContextSerializerFn<T>>,
     context_serializer_into: Option<(usize, TypedContextSerializerIntoFn<T>)>,
@@ -893,6 +894,7 @@ where
             at: Detached,
             url: url.to_string(),
             connector: None,
+            schema: None,
             config: Vec::new(),
             context_serializer: None,
             context_serializer_into: None,
@@ -905,6 +907,7 @@ where
             at,
             url: self.url,
             connector: self.connector,
+            schema: self.schema,
             config: self.config,
             context_serializer: self.context_serializer,
             context_serializer_into: self.context_serializer_into,
@@ -922,6 +925,7 @@ where
             at: Detached,
             url: self.url,
             connector: self.connector,
+            schema: self.schema,
             config: self.config,
             context_serializer: self.context_serializer,
             context_serializer_into: self.context_serializer_into,
@@ -930,6 +934,18 @@ where
         (self.at, link)
     }
 
+    /// Reports `name` / `version` as the record's data-contract schema when
+    /// the link is registered, as [`RecordRegistrar::schema`] would.
+    ///
+    /// Reported even if the link itself is rejected, like a `schema()` call
+    /// before it. A different schema on the same record fails `build()`.
+    /// `aimdb-data-contracts`' `publish` / `subscribe` set it from the
+    /// contract type.
+    pub fn with_schema(mut self, name: &str, version: u32) -> Self {
+        self.schema = Some((name.to_string(), version));
+        self
+    }
+
     /// Adds a configuration option to the connector
     pub fn with_config(mut self, key: &str, value: &str) -> Self {
         self.config.push((key.to_string(), value.to_string()));
@@ -1170,6 +1186,7 @@ pub struct InboundLinkBuilder<T: Send + Sync + 'static + Debug + Clone, R> {
     at: R,
     url: String,
     connector: Option<crate::ConnectorHandle>,
+    schema: Option<(String, u32)>,
     config: Vec<(String, String)>,
     context_deserializer: Option<TypedContextDeserializerFn<T>>,
     match_deserializer: Option<TypedMatchDeserializerFn<T>>,
@@ -1202,6 +1219,7 @@ where
             at: Detached,
             url: url.to_string(),
             connector: None,
+            schema: None,
             config: Vec::new(),
             context_deserializer: None,
             match_deserializer: None,
@@ -1215,6 +1233,7 @@ where
             at,
             url: self.url,
             connector: self.connector,
+            schema: self.schema,
             config: self.config,
             context_deserializer: self.context_deserializer,
             match_deserializer: self.match_deserializer,
@@ -1233,6 +1252,7 @@ where
             at: Detached,
             url: self.url,
             connector: self.connector,
+            schema: self.schema,
             config: self.config,
             context_deserializer: self.context_deserializer,
             match_deserializer: self.match_deserializer,
@@ -1242,6 +1262,18 @@ where
         (self.at, link)
     }
 
+    /// Reports `name` / `version` as the record's data-contract schema when
+    /// the link is registered, as [`RecordRegistrar::schema`] would.
+    ///
+    /// Reported even if the link itself is rejected, like a `schema()` call
+    /// before it. A different schema on the same record fails `build()`.
+    /// `aimdb-data-contracts`' `publish` / `subscribe` set it from the
+    /// contract type.
+    pub fn with_schema(mut self, name: &str, version: u32) -> Self {
+        self.schema = Some((name.to_string(), version));
+        self
+    }
+
     /// Adds a configuration option to the connector
     pub fn with_config(mut self, key: &str, value: &str) -> Self {
         self.config.push((key.to_string(), value.to_string()));
@@ -1366,6 +1398,10 @@ where
         use crate::connector::{ConnectorLink, LinkAddress};
         use crate::error::ConfigError;
 
+        if let Some((name, version)) = link.schema.take() {
+            self.schema(&name, version);
+        }
+
         let record_key = self.record_key.clone();
 
         if let Some(handle) = link.connector {
@@ -1529,6 +1565,9 @@ where
 
     /// Validates and registers an inbound link, or records why not.
     fn push_inbound(&mut self, mut link: InboundLink<T>) {
+        if let Some((name, version)) = link.schema.take() {
+            self.schema(&name, version);
+        }
         match self.inbound_link(&mut link) {
             Ok(inbound) => self.rec.add_inbound_connector(inbound),
             Err(message) => {
@@ -2333,6 +2372,61 @@ mod tests {
         }
     }
 
+    /// `with_schema` reports the record's schema on registration, in both
+    /// directions and on both forms, even when the link is rejected; a second
+    /// schema conflicts as with `reg.schema(..)`.
+    #[test]
+    fn with_schema_reports_the_record_schema() {
+        let schema_of = |rec: &crate::typed_record::TypedRecord<TestRecord>| {
+            rec.schema().map(|s| (s.name.clone(), s.version))
+        };
+        let expected = Some(("contract".to_string(), 2));
+
+        let mut detached = registered(|reg| {
+            reg.output(
+                OutboundLink::new("mqtt://out")
+                    .with_serializer(ser)
+                    .with_schema("contract", 2),
+            )
+            .input(
+                InboundLink::new("mqtt://in")
+                    .with_deserializer(de)
+                    .with_schema("contract", 2),
+            );
+        });
+        assert_eq!(schema_of(&detached), expected);
+        assert_eq!(detached.outbound_connectors().len(), 1);
+        assert!(drain_errors(&mut detached).is_empty());
+
+        let mut attached = registered(|reg| {
+            reg.link_from("mqtt://in")
+                .with_deserializer(de)
+                .with_schema("contract", 2)
+                .finish();
+        });
+        assert_eq!(schema_of(&attached), expected);
+        assert!(drain_errors(&mut attached).is_empty());
+
+        // Rejected (no serializer), yet the schema is reported.
+        let mut rejected = registered(|reg| {
+            reg.output(OutboundLink::new("mqtt://out").with_schema("contract", 2));
+        });
+        assert_eq!(schema_of(&rejected), expected);
+        assert!(rejected.outbound_connectors().is_empty());
+        assert_eq!(drain_errors(&mut rejected).len(), 1);
+
+        let mut conflicting = registered(|reg| {
+            reg.schema("contract", 1).output(
+                OutboundLink::new("mqtt://out")
+                    .with_serializer(ser)
+                    .with_schema("contract", 2),
+            );
+        });
+        let errors = drain_errors(&mut conflicting);
+        assert_eq!(errors.len(), 1, "{errors:?}");
+        assert!(errors[0].message.contains("already carries contract v1"));
+    }
+
     /// A connector registered with `connect` serves links exactly like one
     /// registered with `with_connector`.
     #[tokio::test]

aimdb-data-contracts/src/lib.rs

@@ -89,7 +89,7 @@ mod linkable;
 pub use link_codec::{link_codecs, LinkCodec, LinkCodecBuilderExt, WireFormat};
 
 #[cfg(feature = "linkable")]
-pub use linkable::{LinkCodecRegistrarExt, LinkableRegistrarExt};
+pub use linkable::{publish, subscribe, LinkCodecRegistrarExt, LinkableRegistrarExt};
 
 /// JSON `#[derive(Linkable)]` (feature `linkable-json`) — see [`Linkable`] and
 /// `aimdb_derive::Linkable`'s docs.
@@ -249,10 +249,10 @@ pub trait Observable: SchemaType {
 
 /// Types that can be serialized/deserialized for connector links.
 ///
-/// Implement this trait, then call
-/// [`LinkableRegistrarExt::linked_from`] /
-/// [`linked_to`](linkable::LinkableRegistrarExt::linked_to) to wire `link_from`
-/// / `link_to` in AimDB connectors (MQTT, KNX, etc.) with the codec defaulted to
+/// Implement this trait, then link records with [`publish`] / [`subscribe`]
+/// on a connector handle, or with [`LinkableRegistrarExt::linked_from`] /
+/// [`linked_to`](linkable::LinkableRegistrarExt::linked_to) from a full URL,
+/// in AimDB connectors (MQTT, KNX, etc.) with the codec defaulted to
 /// `from_bytes`/`to_bytes`. This provides the wire format for transporting
 /// schema types across network boundaries.
 ///
@@ -268,16 +268,21 @@ pub trait Observable: SchemaType {
 /// Not compiled: the snippet needs `aimdb-core`'s builder types in scope.
 ///
 /// ```rust,ignore
-/// use aimdb_data_contracts::{Linkable, LinkableRegistrarExt};
+/// use aimdb_data_contracts::subscribe;
 /// use my_app::Temperature;  // user-defined type implementing Linkable
 ///
-/// // In connector configuration:
+/// let mqtt = builder.connect(MqttConnector::new(broker_url));
 /// builder.configure::<Temperature>(NODE_ID, |reg| {
-///     reg.buffer(BufferCfg::SingleLatest);
-///     reg.linked_from("mqtt://sensors/temperature");
+///     reg.buffer(BufferCfg::SingleLatest)
+///         .input(subscribe(mqtt, "sensors/temperature"));
 /// });
 /// ```
 #[cfg(feature = "linkable")]
+#[diagnostic::on_unimplemented(
+    message = "`{Self}` is not a `Linkable` contract",
+    label = "`{Self}` has no wire format",
+    note = "`publish`, `subscribe` and the `linked_*` verbs encode with the record's own codec: derive or implement `Linkable` for `{Self}`"
+)]
 pub trait Linkable: SchemaType + Sized {
     /// Preferred reusable scratch capacity for [`encode_into`](Self::encode_into).
     ///

aimdb-data-contracts/src/link_codec.rs

@@ -863,6 +863,65 @@ mod tests {
         });
     }
 
+    /// `publish` / `subscribe` register what `linked_to` / `linked_from` do:
+    /// the same links, config and schema, from a bare address or a full URL.
+    #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))]
+    #[test]
+    fn publish_and_subscribe_register_like_the_linked_verbs() {
+        use crate::{publish, subscribe, LinkableRegistrarExt};
+
+        futures::executor::block_on(async {
+            let reading = Reading {
+                value: 1.0,
+                sequence: 1,
+            };
+            let mut builder = AimDbBuilder::new().runtime(Arc::new(NoopRuntimeOps));
+            let test = builder.connect(NoopConnector);
+            let linked_reading = reading.clone();
+            builder.configure::<Reading>("linked", move |registrar| {
+                registrar.buffer_raw(Box::new(CannedBuffer {
+                    value: linked_reading,
+                }));
+                registrar.linked_to("test://out").linked_from("test://in");
+            });
+            builder.configure::<Reading>("handle", move |registrar| {
+                registrar
+                    .buffer_raw(Box::new(CannedBuffer { value: reading }))
+                    .output(publish(test, "out"))
+                    // A full URL, as a `#[link_address]` key gives.
+                    .input(subscribe(test, "test://in"));
+            });
+            let (db, _runner) = builder.build().await.expect("build");
+
+            let record = |key: &str| {
+                let id = db.inner().resolve_str(key).unwrap();
+                db.inner().storage(id).unwrap()
+            };
+            let (linked, handle) = (record("linked"), record("handle"));
+            let urls = |r: &dyn aimdb_core::AnyRecord| {
+                (
+                    r.outbound_connectors()[0].url.to_string(),
+                    r.inbound_connectors()[0].url.to_string(),
+                )
+            };
+            assert_eq!(urls(linked), urls(handle));
+            assert_eq!(
+                linked.outbound_connectors()[0].config,
+                handle.outbound_connectors()[0].config
+            );
+            assert_eq!(
+                linked.inbound_connectors()[0].config,
+                handle.inbound_connectors()[0].config
+            );
+
+            let graph = db.inner().dependency_graph();
+            let schema = |key: &str| graph.node(key).unwrap().schema.clone();
+            assert_eq!(schema("linked"), schema("handle"));
+            let schema = schema("handle").expect("publish/subscribe report the schema");
+            assert_eq!((schema.name.as_str(), schema.version), (Reading::NAME, 1));
+        });
+    }
+
     /// `with_link_codec` on detached links: the same codec and wire format
     /// as on `link_to` / `link_from`.
     #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))]