Enable type names for Protocol Buffer message types to support Any decoding.

Cargo.toml

@@ -33,6 +33,7 @@ oci-spec = "0.6"
 os_pipe = "1.1"
 prctl = "1.0.0"
 prost = "0.12"
+prost-build = "0.12"
 prost-types = "0.12"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

crates/client/Cargo.toml

@@ -31,6 +31,7 @@ tower = { workspace = true, optional = true }
 
 [build-dependencies]
 tonic-build.workspace = true
+prost-build.workspace = true
 
 [features]
 connect = ["tokio", "tower"]

crates/client/build.rs

@@ -65,9 +65,12 @@ const FIXUP_MODULES: &[&str] = &[
 ];
 
 fn main() {
+    let mut config = prost_build::Config::new();
+    config.enable_type_names();
+
     tonic_build::configure()
         .build_server(false)
-        .compile(PROTO_FILES, &["vendor/"])
+        .compile_with_config(config, PROTO_FILES, &["vendor/"])
         .expect("Failed to generate GRPC bindings");
 
     for module in FIXUP_MODULES {

crates/client/examples/container_events.rs

@@ -0,0 +1,92 @@
+/*
+   Copyright The containerd Authors.
+
+   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 client::{
+    events::{ContainerCreate, ContainerDelete},
+    services::v1::{events_client::EventsClient, SubscribeRequest},
+};
+use containerd_client as client;
+
+/// Make sure you run containerd before running this example.
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    let channel = client::connect("/run/containerd/containerd.sock")
+        .await
+        .expect("Connect Failed");
+
+    let mut client = EventsClient::new(channel.clone());
+
+    let request = SubscribeRequest::default();
+    let mut response = client
+        .subscribe(request)
+        .await
+        .expect("failed to subscribe to events")
+        .into_inner();
+
+    loop {
+        match response.message().await {
+            Ok(event) => {
+                if let Some(event) = event {
+                    match event.topic.as_str() {
+                        "/containers/create" => {
+                            if let Some(mut payload) = event.event {
+                                // Containerd doesn't send event payloads with a leading slash on the type URL, which is
+                                // required by the `Any` type specification. We add it manually here so that `prost` can
+                                // properly decode the payload.
+                                if !payload.type_url.starts_with('/') {
+                                    payload.type_url.insert(0, '/');
+                                }
+
+                                let payload: ContainerCreate = payload
+                                    .to_msg()
+                                    .expect("failed to parse ContainerCreate payload");
+
+                                println!(
+                                    "container created: id={} payload={:?}",
+                                    payload.id, payload
+                                );
+                            }
+                        }
+                        "/containers/delete" => {
+                            if let Some(mut payload) = event.event {
+                                // Containerd doesn't send event payloads with a leading slash on the type URL, which is
+                                // required by the `Any` type specification. We add it manually here so that `prost` can
+                                // properly decode the payload.
+                                if !payload.type_url.starts_with('/') {
+                                    payload.type_url.insert(0, '/');
+                                }
+
+                                let payload: ContainerDelete = payload
+                                    .to_msg()
+                                    .expect("failed to parse ContainerDelete payload");
+
+                                println!(
+                                    "container deleted: id={} payload={:?}",
+                                    payload.id, payload
+                                );
+                            }
+                        }
+                        _ => {}
+                    }
+                }
+            }
+            Err(e) => {
+                eprintln!("error while streaming events: {:?}", e);
+                break;
+            }
+        }
+    }
+}

crates/client/src/lib.rs

@@ -245,3 +245,24 @@ impl Client {
         ContainersClient::new(self.channel())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use prost_types::Any;
+
+    use crate::events::ContainerCreate;
+
+    #[test]
+    fn any_roundtrip() {
+        let original = ContainerCreate {
+            id: "test".to_string(),
+            image: "test".to_string(),
+            runtime: None,
+        };
+
+        let any = Any::from_msg(&original).expect("should not fail to encode");
+        let decoded: ContainerCreate = any.to_msg().expect("should not fail to decode");
+
+        assert_eq!(original, decoded)
+    }
+}