diff --git a/Cargo.lock b/Cargo.lock index 7d69790..d631eaa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3354,9 +3354,9 @@ checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" [[package]] name = "wadm" -version = "0.12.1" +version = "0.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53c29efcb1a7d257c49cfc34db0d81c4e18f75f3317ac10d0c0ba4cc64545fba" +checksum = "2d20c671aabb8fee1d7a688a81be4ebe14cbc4ae3a01db5dd62202ad39538fa5" dependencies = [ "anyhow", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index 73a9340..00e4a0b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -92,7 +92,7 @@ tracing-opentelemetry = "0.20" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } utoipa = { version = "4.1", features = ["axum_extras"] } uuid = { version = "1", features = ["v5"] } -wadm = "0.12.1" +wadm = "0.12.2" wadm-client = "0.1.2" wadm-types = "0.1.0" wasmcloud-operator-types = { version = "*", path = "./crates/types" } diff --git a/src/services.rs b/src/services.rs index 5247377..6a89d20 100644 --- a/src/services.rs +++ b/src/services.rs @@ -37,8 +37,9 @@ use crate::controller::{ const CONSUMER_PREFIX: &str = "wasmcloud_operator_service"; // This should probably be exposed by wadm somewhere -const WADM_EVT_SUBJECT: &str = "wadm.evt"; +const WADM_EVENT_STREAM_NAME: &str = "wadm_events"; const OPERATOR_STREAM_NAME: &str = "wasmcloud_operator_events"; +const OPERATOR_STREAM_SUBJECT: &str = "wasmcloud_operator_events.*.>"; /// Commands that can be sent to the watcher to trigger an update or removal of a service. #[derive(Clone, Debug)] @@ -305,8 +306,6 @@ impl ServiceWatcher { } let js = jetstream::new(client.clone()); - let source_subject = format!("{WADM_EVT_SUBJECT}.{}", lattice_id.clone()); - let destination_subject = format!("wasmcloud_operator_events.{}", lattice_id.clone()); // Should we also be doing this when we first create the ServiceWatcher? let stream = js @@ -321,10 +320,10 @@ impl ServiceWatcher { allow_rollup: false, num_replicas: self.stream_replicas as usize, mirror: Some(Source { - name: "wadm_events".to_string(), + name: WADM_EVENT_STREAM_NAME.to_string(), subject_transforms: vec![SubjectTransform { - source: source_subject, - destination: format!("wasmcloud_operator_events.{}", lattice_id.clone()), + source: wadm::DEFAULT_WADM_EVENTS_TOPIC.to_string(), + destination: OPERATOR_STREAM_SUBJECT.replacen('*', "{{wildcard(1)}}", 1), }], ..Default::default() }), @@ -343,7 +342,7 @@ impl ServiceWatcher { ack_wait: std::time::Duration::from_secs(2), max_deliver: 3, deliver_policy: async_nats::jetstream::consumer::DeliverPolicy::All, - filter_subject: destination_subject.clone(), + filter_subject: OPERATOR_STREAM_SUBJECT.replacen('*', &lattice_id, 1), ..Default::default() }, ) @@ -623,7 +622,7 @@ fn http_server_component(manifest: &Manifest) -> Option { for p in props.source_config.iter() { if let Some(config_props) = &p.properties { if let Some(addr) = config_props.get("address") { - details.address = addr.clone(); + details.address.clone_from(addr); should_create_service = true; }; }