gg_sdk/

ipc.rs

// aws-greengrass-component-sdk - Lightweight AWS IoT Greengrass SDK
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0

use core::{
    ffi,
    marker::PhantomData,
    mem::MaybeUninit,
    ptr, result, slice, str,
    sync::atomic::{AtomicBool, Ordering},
};

use crate::{
    c,
    error::{Error, Result},
    object::{Kv, Map, Object},
};

static INIT: AtomicBool = AtomicBool::new(false);
static CONNECTED: AtomicBool = AtomicBool::new(false);

/// AWS IoT Greengrass IPC SDK client.
#[non_exhaustive]
#[derive(Debug, Clone, Copy)]
pub struct Sdk {}

#[derive(Debug, Clone, Copy)]
pub struct IpcError<'a> {
    pub error_code: &'a str,
    pub message: &'a str,
}

/// MQTT Quality of Service level.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum Qos {
    /// At most once delivery (QoS 0)
    AtMostOnce = 0,
    /// At least once delivery (QoS 1)
    AtLeastOnce = 1,
}

/// Component lifecycle state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ComponentState {
    /// Component is running
    Running,
    /// Component encountered an error
    Errored,
}

/// Payload received from a topic subscription.
#[derive(Debug, Clone, Copy)]
pub enum SubscribeToTopicPayload<'a> {
    /// JSON payload
    Json(Map<'a>),
    /// Binary payload
    Binary(&'a [u8]),
}

impl Sdk {
    /// Initialize the SDK.
    ///
    /// Must be called before using any IPC operations.
    ///
    /// # Panics
    /// Panics if called more than once.
    pub fn init() -> Self {
        let already_init = INIT.swap(true, Ordering::AcqRel);
        assert!(!already_init, "Sdk::init() called more than once");
        unsafe { c::gg_sdk_init() };
        Self {}
    }

    /// Connect to the AWS IoT Greengrass Core IPC service.
    ///
    /// Uses `SVCUID` and `AWS_GG_NUCLEUS_DOMAIN_SOCKET_FILEPATH_FOR_COMPONENT`
    /// environment variables set by the Greengrass nucleus.
    ///
    /// # Errors
    /// Returns error if environment variables are missing, connected or connecting, or connection fails.
    pub fn connect(&self) -> Result<()> {
        let already_connected = CONNECTED.swap(true, Ordering::AcqRel);
        if already_connected {
            return Err(Error::Failure);
        }
        Result::from(unsafe { c::ggipc_connect() })
    }

    /// Connect to the AWS IoT Greengrass Core IPC service with explicit credentials.
    ///
    /// # Errors
    /// Returns error if connected or connecting, or if connection fails.
    pub fn connect_with_token(
        &self,
        socket_path: &str,
        auth_token: &str,
    ) -> Result<()> {
        let already_connected = CONNECTED.swap(true, Ordering::AcqRel);
        if already_connected {
            return Err(Error::Failure);
        }

        Result::from(unsafe {
            c::ggipc_connect_with_token(socket_path.into(), auth_token.into())
        })?;

        Ok(())
    }

    /// Publish a JSON message to a local pub/sub topic.
    ///
    /// Sends messages to other Greengrass components subscribed to the topic.
    /// Requires `aws.greengrass#PublishToTopic` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-publish-subscribe.html#ipc-operation-publishtotopic>
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use gg_sdk::{Sdk, Kv, Object};
    ///
    /// let sdk = Sdk::init();
    /// sdk.connect()?;
    ///
    /// let payload = [
    ///     Kv::new("temperature", Object::f64(72.5)),
    ///     Kv::new("humidity", Object::i64(45)),
    /// ];
    /// sdk.publish_to_topic_json("sensor/data", &payload[..])?;
    /// # Ok::<(), gg_sdk::Error>(())
    /// ```
    ///
    /// # Errors
    /// Returns error if publish fails.
    pub fn publish_to_topic_json(
        &self,
        topic: &str,
        payload: &[Kv<'_>],
    ) -> Result<()> {
        Result::from(unsafe {
            c::ggipc_publish_to_topic_json(topic.into(), payload.into())
        })
    }

    /// Publish a binary message to a local pub/sub topic.
    ///
    /// Sends messages to other Greengrass components subscribed to the topic.
    /// Requires `aws.greengrass#PublishToTopic` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-publish-subscribe.html#ipc-operation-publishtotopic>
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use gg_sdk::Sdk;
    ///
    /// let sdk = Sdk::init();
    /// sdk.connect()?;
    ///
    /// let data = b"binary payload data";
    /// sdk.publish_to_topic_binary("sensor/raw", data)?;
    /// # Ok::<(), gg_sdk::Error>(())
    /// ```
    ///
    /// # Errors
    /// Returns error if publish fails.
    pub fn publish_to_topic_binary(
        &self,
        topic: &str,
        payload: &[u8],
    ) -> Result<()> {
        Result::from(unsafe {
            c::ggipc_publish_to_topic_binary(topic.into(), payload.into())
        })
    }

    /// Subscribe to messages on a local pub/sub topic.
    ///
    /// Receives messages from other Greengrass components publishing to the topic.
    /// Requires `aws.greengrass#SubscribeToTopic` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-publish-subscribe.html#ipc-operation-subscribetotopic>
    ///
    /// # Errors
    /// Returns error if subscription fails.
    pub fn subscribe_to_topic<'a, F: Fn(&str, SubscribeToTopicPayload)>(
        &self,
        topic: &str,
        callback: &'a F,
    ) -> Result<Subscription<'a, F>> {
        extern "C" fn trampoline<F: Fn(&str, SubscribeToTopicPayload)>(
            ctx: *mut ffi::c_void,
            topic: c::GgBuffer,
            payload: c::GgObject,
            _handle: c::GgIpcSubscriptionHandle,
        ) {
            let cb = unsafe { &*ctx.cast::<F>() };
            let topic_str = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    topic.data, topic.len,
                ))
            };

            let unpacked = match unsafe { c::gg_obj_type(payload) } {
                c::GgObjectType::GG_TYPE_MAP => {
                    let map = unsafe { c::gg_obj_into_map(payload) };
                    SubscribeToTopicPayload::Json(Map(unsafe {
                        slice::from_raw_parts(map.pairs as *const Kv, map.len)
                    }))
                }
                c::GgObjectType::GG_TYPE_BUF => {
                    let buf = unsafe { c::gg_obj_into_buf(payload) };
                    SubscribeToTopicPayload::Binary(unsafe {
                        slice::from_raw_parts(buf.data, buf.len)
                    })
                }
                _ => return,
            };

            cb(topic_str, unpacked);
        }

        let ctx = callback as *const F;
        let mut handle = c::GgIpcSubscriptionHandle { val: 0 };

        Result::from(unsafe {
            c::ggipc_subscribe_to_topic(
                topic.into(),
                Some(trampoline::<F>),
                ctx.cast::<ffi::c_void>().cast_mut(),
                &raw mut handle,
            )
        })?;

        debug_assert!(handle.val != 0);
        Ok(Subscription {
            handle,
            phantom: PhantomData,
        })
    }

    /// Publish an MQTT message to AWS IoT Core.
    ///
    /// Sends messages to AWS IoT Core MQTT broker with specified QoS.
    /// Requires `aws.greengrass#PublishToIoTCore` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-iot-core-mqtt.html#ipc-operation-publishtoiotcore>
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use gg_sdk::{Sdk, Qos};
    ///
    /// let sdk = Sdk::init();
    /// sdk.connect()?;
    ///
    /// let payload = b"telemetry data";
    /// sdk.publish_to_iot_core("device/telemetry", payload, Qos::AtMostOnce)?;
    /// # Ok::<(), gg_sdk::Error>(())
    /// ```
    ///
    /// # Errors
    /// Returns error if publish fails.
    pub fn publish_to_iot_core(
        &self,
        topic: &str,
        payload: &[u8],
        qos: Qos,
    ) -> Result<()> {
        Result::from(unsafe {
            c::ggipc_publish_to_iot_core(topic.into(), payload.into(), qos as u8)
        })
    }

    /// Subscribe to MQTT messages from AWS IoT Core.
    ///
    /// Receives messages from AWS IoT Core MQTT broker on matching topics.
    /// Requires `aws.greengrass#SubscribeToIoTCore` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-iot-core-mqtt.html#ipc-operation-subscribetoiotcore>
    ///
    /// # Errors
    /// Returns error if subscription fails.
    pub fn subscribe_to_iot_core<'a, F: Fn(&str, &[u8])>(
        &self,
        topic_filter: &str,
        qos: Qos,
        callback: &'a F,
    ) -> Result<Subscription<'a, F>> {
        extern "C" fn trampoline<F: Fn(&str, &[u8])>(
            ctx: *mut ffi::c_void,
            topic: c::GgBuffer,
            payload: c::GgBuffer,
            _handle: c::GgIpcSubscriptionHandle,
        ) {
            let cb = unsafe { &*ctx.cast::<F>() };
            let topic_str = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    topic.data, topic.len,
                ))
            };
            let payload_bytes =
                unsafe { slice::from_raw_parts(payload.data, payload.len) };
            cb(topic_str, payload_bytes);
        }

        let ctx = callback as *const F;
        let mut handle = c::GgIpcSubscriptionHandle { val: 0 };

        Result::from(unsafe {
            c::ggipc_subscribe_to_iot_core(
                topic_filter.into(),
                qos as u8,
                Some(trampoline::<F>),
                ctx.cast::<ffi::c_void>().cast_mut(),
                &raw mut handle,
            )
        })?;

        debug_assert!(handle.val != 0);
        Ok(Subscription {
            handle,
            phantom: PhantomData,
        })
    }

    /// Update component state.
    ///
    /// Reports component state to the Greengrass nucleus.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-component-lifecycle.html#ipc-operation-updatestate>
    ///
    /// # Errors
    /// Returns error if state update fails.
    pub fn update_state(&self, state: ComponentState) -> Result<()> {
        let c_state = match state {
            ComponentState::Running => {
                c::GgComponentState::GG_COMPONENT_STATE_RUNNING
            }
            ComponentState::Errored => {
                c::GgComponentState::GG_COMPONENT_STATE_ERRORED
            }
        };
        Result::from(unsafe { c::ggipc_update_state(c_state) })
    }

    /// Restart a Greengrass component.
    ///
    /// Requests the nucleus to restart the specified component.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-local-deployments-components.html#ipc-operation-restartcomponent>
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use gg_sdk::Sdk;
    ///
    /// let sdk = Sdk::init();
    /// sdk.connect()?;
    ///
    /// sdk.restart_component("com.example.MyComponent")?;
    /// # Ok::<(), gg_sdk::Error>(())
    /// ```
    ///
    /// # Errors
    /// Returns error if restart fails.
    pub fn restart_component(&self, component_name: &str) -> Result<()> {
        Result::from(unsafe { c::ggipc_restart_component(component_name.into()) })
    }

    /// Get component configuration value.
    ///
    /// Retrieves configuration for the specified key path. Pass empty slice for complete config.
    /// Requires `aws.greengrass#GetConfiguration` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-component-configuration.html#ipc-operation-getconfiguration>
    ///
    /// # Errors
    /// Returns error if config retrieval fails.
    pub fn get_config<'a>(
        &self,
        key_path: &[&str],
        component_name: Option<&str>,
        result_mem: &'a mut [MaybeUninit<u8>],
    ) -> Result<Object<'a>> {
        let mut c_key_path_mem = [MaybeUninit::uninit(); MAX_KEY_PATH_LEN];
        let c_key_path = key_path_to_buf_list(key_path, &mut c_key_path_mem)?;

        let component_buf = component_name.map(c::GgBuffer::from);

        let mem = c::GgBuffer {
            data: result_mem.as_mut_ptr().cast::<u8>(),
            len: result_mem.len(),
        };

        let mut obj = c::GgObject::default();

        Result::from(unsafe {
            c::ggipc_get_config(
                c_key_path,
                component_buf.as_ref().map_or(ptr::null(), ptr::from_ref),
                mem,
                &raw mut obj,
            )
        })?;

        Ok(unsafe { ptr::read((&raw const obj).cast()) })
    }

    /// Get component configuration value as a string.
    ///
    /// Alternative API to [`Sdk::get_config`] for string type values.
    /// Requires `aws.greengrass#GetConfiguration` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-component-configuration.html#ipc-operation-getconfiguration>
    ///
    /// # Errors
    /// Returns error if config retrieval fails.
    pub fn get_config_str<'a>(
        &self,
        key_path: &[&str],
        component_name: Option<&str>,
        result_mem: &'a mut [MaybeUninit<u8>],
    ) -> Result<&'a str> {
        let mut c_key_path_mem = [MaybeUninit::uninit(); MAX_KEY_PATH_LEN];
        let c_key_path = key_path_to_buf_list(key_path, &mut c_key_path_mem)?;

        let component_buf = component_name.map(c::GgBuffer::from);

        let mut value = c::GgBuffer {
            data: result_mem.as_mut_ptr().cast::<u8>(),
            len: result_mem.len(),
        };

        Result::from(unsafe {
            c::ggipc_get_config_str(
                c_key_path,
                component_buf.as_ref().map_or(ptr::null(), ptr::from_ref),
                &raw mut value,
            )
        })?;

        Ok(unsafe {
            str::from_utf8_unchecked(slice::from_raw_parts(
                value.data, value.len,
            ))
        })
    }

    /// Update component configuration.
    ///
    /// Merges the provided value into the component's configuration at the key path.
    /// Requires `aws.greengrass#UpdateConfiguration` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-component-configuration.html#ipc-operation-updateconfiguration>
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use gg_sdk::Sdk;
    ///
    /// let sdk = Sdk::init();
    /// sdk.connect()?;
    ///
    /// sdk.update_config(&["maxRetries"], None, 100_i64)?;
    /// # Ok::<(), gg_sdk::Error>(())
    /// ```
    ///
    /// # Errors
    /// Returns error if config update fails.
    pub fn update_config<'a>(
        &self,
        key_path: &[&str],
        timestamp: Option<core::time::Duration>,
        value_to_merge: impl Into<Object<'a>>,
    ) -> Result<()> {
        fn inner(
            key_path: &[&str],
            timestamp: Option<core::time::Duration>,
            value_to_merge: Object,
        ) -> Result<()> {
            let mut c_key_path_mem = [MaybeUninit::uninit(); MAX_KEY_PATH_LEN];
            let c_key_path =
                key_path_to_buf_list(key_path, &mut c_key_path_mem)?;

            #[expect(clippy::cast_possible_wrap)]
            #[allow(clippy::cast_lossless, clippy::needless_update)]
            let timespec = timestamp.map(|d| c::timespec {
                tv_sec: d.as_secs() as _,
                tv_nsec: d.subsec_nanos() as _,
                ..c::timespec::default()
            });

            Result::from(unsafe {
                c::ggipc_update_config(
                    c_key_path,
                    timespec.as_ref().map_or(ptr::null(), ptr::from_ref),
                    *ptr::from_ref(&value_to_merge).cast::<c::GgObject>(),
                )
            })
        }
        inner(key_path, timestamp, value_to_merge.into())
    }

    /// Subscribe to component configuration updates.
    ///
    /// Receives notifications when configuration changes for the specified key path.
    /// Requires `aws.greengrass#SubscribeToConfigurationUpdate` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-component-configuration.html#ipc-operation-subscribetoconfigurationupdate>
    ///
    /// # Errors
    /// Returns error if subscription fails.
    pub fn subscribe_to_configuration_update<'a, F: Fn(&str, &[&str])>(
        &self,
        component_name: Option<&str>,
        key_path: &[&str],
        callback: &'a F,
    ) -> Result<Subscription<'a, F>> {
        extern "C" fn trampoline<F: Fn(&str, &[&str])>(
            ctx: *mut ffi::c_void,
            component_name: c::GgBuffer,
            key_path: c::GgList,
            _handle: c::GgIpcSubscriptionHandle,
        ) {
            let cb = unsafe { &*ctx.cast::<F>() };
            let component_str = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    component_name.data,
                    component_name.len,
                ))
            };
            let path_objs =
                unsafe { slice::from_raw_parts(key_path.items, key_path.len) };

            let mut path_strs_mem = [MaybeUninit::uninit(); MAX_KEY_PATH_LEN];
            for (i, obj) in path_objs.iter().enumerate() {
                let buf = unsafe { c::gg_obj_into_buf(*obj) };
                let s = unsafe {
                    str::from_utf8_unchecked(slice::from_raw_parts(
                        buf.data, buf.len,
                    ))
                };
                path_strs_mem[i].write(s);
            }
            let path_strs = unsafe {
                slice::from_raw_parts(
                    path_strs_mem.as_ptr().cast::<&str>(),
                    path_objs.len(),
                )
            };

            cb(component_str, path_strs);
        }

        let mut c_key_path_mem = [MaybeUninit::uninit(); MAX_KEY_PATH_LEN];
        let c_key_path = key_path_to_buf_list(key_path, &mut c_key_path_mem)?;

        let component_buf = component_name.map(c::GgBuffer::from);

        let ctx = callback as *const F;
        let mut handle = c::GgIpcSubscriptionHandle { val: 0 };

        Result::from(unsafe {
            c::ggipc_subscribe_to_configuration_update(
                component_buf.as_ref().map_or(ptr::null(), ptr::from_ref),
                c_key_path,
                Some(trampoline::<F>),
                ctx.cast::<ffi::c_void>().cast_mut(),
                &raw mut handle,
            )
        })?;

        debug_assert!(handle.val != 0);
        Ok(Subscription {
            handle,
            phantom: PhantomData,
        })
    }

    /// Get the shadow for a thing.
    ///
    /// Retrieves the shadow document for the specified thing and shadow name.
    /// Pass `None` for `shadow_name` to use the classic shadow.
    /// `result_mem` must be large enough to hold the decoded shadow document.
    /// Requires `aws.greengrass#GetThingShadow` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-local-shadows.html#ipc-operation-getthingshadow>
    ///
    /// # Errors
    /// Returns error if the shadow retrieval fails.
    pub fn get_thing_shadow<'a>(
        &self,
        thing_name: &str,
        shadow_name: Option<&str>,
        result_mem: &'a mut [MaybeUninit<u8>],
    ) -> Result<&'a [u8]> {
        let shadow_buf = shadow_name.map(c::GgBuffer::from);

        let mut payload = c::GgBuffer {
            data: result_mem.as_mut_ptr().cast::<u8>(),
            len: result_mem.len(),
        };

        Result::from(unsafe {
            c::ggipc_get_thing_shadow(
                thing_name.into(),
                shadow_buf.as_ref().map_or(ptr::null(), ptr::from_ref),
                &raw mut payload,
            )
        })?;

        Ok(unsafe { slice::from_raw_parts(payload.data, payload.len) })
    }

    /// Update the shadow for a thing.
    ///
    /// Updates the shadow document for the specified thing and shadow name.
    /// Pass `None` for `shadow_name` to use the classic shadow.
    /// Pass `Some` buffer for `response_mem` to receive the response payload,
    /// or `None` to ignore it.
    /// Requires `aws.greengrass#UpdateThingShadow` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-local-shadows.html#ipc-operation-updatethingshadow>
    ///
    /// # Errors
    /// Returns error if the shadow update fails.
    #[expect(clippy::needless_pass_by_value)]
    pub fn update_thing_shadow<'a>(
        &self,
        thing_name: &str,
        shadow_name: Option<&str>,
        payload: &[u8],
        response_mem: Option<&'a mut [MaybeUninit<u8>]>,
    ) -> Result<Option<&'a [u8]>> {
        let shadow_buf = shadow_name.map(c::GgBuffer::from);

        let mut response = response_mem.as_ref().map(|mem| c::GgBuffer {
            data: mem.as_ptr() as *mut u8,
            len: mem.len(),
        });

        Result::from(unsafe {
            c::ggipc_update_thing_shadow(
                thing_name.into(),
                shadow_buf.as_ref().map_or(ptr::null(), ptr::from_ref),
                payload.into(),
                response
                    .as_mut()
                    .map_or(ptr::null_mut(), ptr::from_mut),
            )
        })?;

        Ok(response
            .map(|r| unsafe { slice::from_raw_parts(r.data, r.len) }))
    }

    /// Delete the shadow for a thing.
    ///
    /// Deletes the shadow document for the specified thing and shadow name.
    /// Pass `None` for `shadow_name` to use the classic shadow.
    /// Requires `aws.greengrass#DeleteThingShadow` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-local-shadows.html#ipc-operation-deletethingshadow>
    ///
    /// # Errors
    /// Returns error if the shadow deletion fails.
    pub fn delete_thing_shadow(
        &self,
        thing_name: &str,
        shadow_name: Option<&str>,
    ) -> Result<()> {
        let shadow_buf = shadow_name.map(c::GgBuffer::from);

        Result::from(unsafe {
            c::ggipc_delete_thing_shadow(
                thing_name.into(),
                shadow_buf.as_ref().map_or(ptr::null(), ptr::from_ref),
            )
        })
    }

    /// List named shadows for a thing.
    ///
    /// Lists all named shadows for the specified thing, handling pagination
    /// internally. The callback is invoked once per shadow name.
    /// Requires `aws.greengrass#ListNamedShadowsForThing` authorization.
    ///
    /// See: <https://docs.aws.amazon.com/greengrass/v2/developerguide/ipc-local-shadows.html#ipc-operation-listnamedshadowsforthing>
    ///
    /// # Errors
    /// Returns error if listing fails.
    pub fn list_named_shadows_for_thing<F: FnMut(&str)>(
        &self,
        thing_name: &str,
        callback: &mut F,
    ) -> Result<()> {
        extern "C" fn trampoline<F: FnMut(&str)>(
            ctx: *mut ffi::c_void,
            shadow_name: c::GgBuffer,
        ) {
            let cb = unsafe { &mut *ctx.cast::<F>() };
            let name = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    shadow_name.data,
                    shadow_name.len,
                ))
            };
            cb(name);
        }

        let ctx = callback as *mut F;

        Result::from(unsafe {
            c::ggipc_list_named_shadows_for_thing(
                thing_name.into(),
                Some(trampoline::<F>),
                ctx.cast::<ffi::c_void>(),
            )
        })
    }

    /// Make a generic IPC call.
    ///
    /// Low-level interface for invoking IPC operations not covered by specific methods.
    ///
    /// # Errors
    /// Returns error if IPC call fails.
    pub fn call<
        'a,
        'b,
        F: FnOnce(result::Result<&'b [Kv<'b>], IpcError<'b>>) -> Result<()>,
    >(
        &self,
        operation: &str,
        service_model_type: &str,
        params: &[Kv<'a>],
        mut callback: F,
    ) -> Result<()> {
        extern "C" fn result_trampoline<
            'b,
            F: FnOnce(result::Result<&'b [Kv<'b>], IpcError<'b>>) -> Result<()>,
        >(
            ctx: *mut ffi::c_void,
            result: c::GgMap,
        ) -> c::GgError {
            let cb = unsafe { ctx.cast::<F>().read() };
            let result_slice = unsafe {
                slice::from_raw_parts(result.pairs as *const Kv, result.len)
            };
            cb(Ok(result_slice)).into()
        }

        extern "C" fn error_trampoline<
            'b,
            F: FnOnce(result::Result<&'b [Kv<'b>], IpcError<'b>>) -> Result<()>,
        >(
            ctx: *mut ffi::c_void,
            error_code: c::GgBuffer,
            message: c::GgBuffer,
        ) -> c::GgError {
            let cb = unsafe { ctx.cast::<F>().read() };
            let code = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    error_code.data,
                    error_code.len,
                ))
            };
            let msg = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    message.data,
                    message.len,
                ))
            };
            cb(Err(IpcError {
                error_code: code,
                message: msg,
            }))
            .into()
        }

        Result::from(unsafe {
            c::ggipc_call(
                operation.into(),
                service_model_type.into(),
                params.into(),
                Some(result_trampoline::<F>),
                Some(error_trampoline::<F>),
                (&raw mut callback).cast::<ffi::c_void>(),
            )
        })
    }

    /// Subscribe to a generic IPC stream.
    ///
    /// Low-level interface for subscribing to IPC operations not covered by specific methods.
    ///
    /// # Errors
    /// Returns error if subscription fails.
    pub fn subscribe<
        'a,
        'b,
        'c,
        F: FnOnce(result::Result<&'b [Kv<'b>], IpcError<'b>>) -> Result<()>,
        G: Fn(usize, &str, &'b [Kv<'b>]) -> Result<()>,
    >(
        &self,
        operation: &str,
        service_model_type: &str,
        params: &[Kv<'a>],
        mut response_callback: F,
        sub_callback: &'c G,
        aux_ctx: usize,
    ) -> Result<Subscription<'c, G>> {
        extern "C" fn result_trampoline<
            'b,
            F: FnOnce(result::Result<&'b [Kv<'b>], IpcError<'b>>) -> Result<()>,
        >(
            ctx: *mut ffi::c_void,
            result: c::GgMap,
        ) -> c::GgError {
            let cb = unsafe { ctx.cast::<F>().read() };
            let result_slice = unsafe {
                slice::from_raw_parts(result.pairs as *const Kv, result.len)
            };
            cb(Ok(result_slice)).into()
        }

        extern "C" fn error_trampoline<
            'b,
            F: FnOnce(result::Result<&'b [Kv<'b>], IpcError<'b>>) -> Result<()>,
        >(
            ctx: *mut ffi::c_void,
            error_code: c::GgBuffer,
            message: c::GgBuffer,
        ) -> c::GgError {
            let cb = unsafe { ctx.cast::<F>().read() };
            let code = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    error_code.data,
                    error_code.len,
                ))
            };
            let msg = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    message.data,
                    message.len,
                ))
            };
            cb(Err(IpcError {
                error_code: code,
                message: msg,
            }))
            .into()
        }

        extern "C" fn sub_trampoline<
            'b,
            G: Fn(usize, &str, &'b [Kv<'b>]) -> Result<()>,
        >(
            ctx: *mut ffi::c_void,
            aux_ctx: *mut ffi::c_void,
            _handle: c::GgIpcSubscriptionHandle,
            service_model_type: c::GgBuffer,
            data: c::GgMap,
        ) -> c::GgError {
            let cb = unsafe { &*ctx.cast::<G>() };
            let aux = aux_ctx as usize;
            let smt = unsafe {
                str::from_utf8_unchecked(slice::from_raw_parts(
                    service_model_type.data,
                    service_model_type.len,
                ))
            };
            let map = unsafe {
                slice::from_raw_parts(data.pairs.cast::<Kv>(), data.len)
            };
            cb(aux, smt, map).into()
        }

        let mut handle = c::GgIpcSubscriptionHandle { val: 0 };
        let ctx = sub_callback as *const G;

        Result::from(unsafe {
            c::ggipc_subscribe(
                operation.into(),
                service_model_type.into(),
                params.into(),
                Some(result_trampoline::<F>),
                Some(error_trampoline::<F>),
                (&raw mut response_callback).cast::<ffi::c_void>(),
                Some(sub_trampoline::<'b, G>),
                ctx.cast::<ffi::c_void>().cast_mut(),
                aux_ctx as *mut ffi::c_void,
                &raw mut handle,
            )
        })?;

        Ok(Subscription {
            handle,
            phantom: PhantomData,
        })
    }
}

/// Handle for an active IPC subscription.
#[derive(Debug)]
pub struct Subscription<'a, T> {
    handle: c::GgIpcSubscriptionHandle,
    phantom: PhantomData<&'a T>,
}

impl<T> Drop for Subscription<'_, T> {
    fn drop(&mut self) {
        if self.handle.val != 0 {
            unsafe { c::ggipc_close_subscription(self.handle) };
        }
    }
}

impl<T> Default for Subscription<'_, T> {
    fn default() -> Self {
        Self {
            handle: c::GgIpcSubscriptionHandle { val: 0 },
            phantom: PhantomData,
        }
    }
}

const MAX_KEY_PATH_LEN: usize = (c::GG_MAX_OBJECT_DEPTH - 1) as usize;

fn key_path_to_buf_list(
    key_path: &[&str],
    bufs: &mut [MaybeUninit<c::GgBuffer>; MAX_KEY_PATH_LEN],
) -> Result<c::GgBufList> {
    if key_path.len() > MAX_KEY_PATH_LEN {
        return Err(Error::Range);
    }
    for (i, k) in key_path.iter().enumerate() {
      bufs[i].write((*k).into());
    }
    Ok(c::GgBufList {
        bufs: bufs.as_mut_ptr().cast(),
        len: key_path.len(),
    })
}

#[cfg(test)]
mod test {
    use std::sync::{Condvar, Mutex};

    use super::{Qos, Sdk};
    use crate::c;
    use crate::error::*;

    static TEST_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());

    pub(crate) fn run_ipc_handshake_test<F: FnOnce() -> Result<()>>(
        test_body: F,
    ) -> Result<()> {
        unsafe {
            Result::from(c::gg_test_setup_ipc(
                c"/tmp/gg-test".as_ptr(),
                0o666,
                c"1234567890ABCDEF".as_ptr(),
            ))?;

            let pid = libc::fork();
            assert!(pid >= 0, "fork failed");

            if pid == 0 {
                test_body().unwrap();
                libc::exit(0);
            }

            Result::from(c::gg_test_accept_client_handshake(5))?;

            Result::from(c::gg_test_wait_for_client_disconnect(30))?;

            c::gg_test_close();

            let mut status = 0;
            libc::waitpid(pid, &mut status, 0);

            assert_eq!(libc::WIFEXITED(status), true);
            assert_eq!(libc::WEXITSTATUS(status), 0);
        };

        Ok(())
    }

    pub(crate) fn run_ipc_sequence_test<F: FnOnce() -> Result<()>>(
        packet_sequence: c::GgipcPacketSequence,
        test_body: F,
    ) -> Result<()> {
        unsafe {
            Result::from(c::gg_test_setup_ipc(
                c"/tmp/gg-test".as_ptr(),
                0o666,
                c"1234567890ABCDEF".as_ptr(),
            ))?;

            let pid = libc::fork();
            assert!(pid >= 0, "fork failed");

            if pid == 0 {
                test_body().unwrap();
                libc::exit(0);
            }

            Result::from(c::gg_test_connect_request_disconnect_sequence(
                packet_sequence,
                10,
            ))?;

            c::gg_test_close();

            let mut status = 0;
            libc::waitpid(pid, &mut status, 0);

            assert_eq!(libc::WIFEXITED(status), true);
            assert_eq!(libc::WEXITSTATUS(status), 0);
        };

        Ok(())
    }

    fn get_test_socket_path() -> &'static str {
        unsafe {
            let buf = c::gg_test_get_socket_path();
            core::str::from_utf8_unchecked(core::slice::from_raw_parts(
                buf.data, buf.len,
            ))
        }
    }

    fn get_test_auth_token() -> &'static str {
        unsafe {
            let buf = c::gg_test_get_auth_token();
            core::str::from_utf8_unchecked(core::slice::from_raw_parts(
                buf.data, buf.len,
            ))
        }
    }

    #[test]
    fn test_connect_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        run_ipc_handshake_test(|| {
            let sdk = Sdk::init();
            sdk.connect()
        })
    }

    #[test]
    fn test_connect_with_token_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        run_ipc_handshake_test(|| unsafe {
            // Unset env vars to force explicit token usage
            libc::unsetenv(c"SVCUID".as_ptr());
            libc::unsetenv(
                c"AWS_GG_NUCLEUS_DOMAIN_SOCKET_FILEPATH_FOR_COMPONENT".as_ptr(),
            );

            let sdk = Sdk::init();
            sdk.connect_with_token(
                get_test_socket_path(),
                get_test_auth_token(),
            )
        })
    }

    #[test]
    fn test_publish_to_iot_core_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let topic = "my/topic";
        let payload_base64 = "SGVsbG8gd29ybGQh";
        let qos = "0";
        let seq = unsafe {
            c::gg_test_mqtt_publish_accepted_sequence(
                1,
                topic.into(),
                payload_base64.into(),
                qos.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            sdk.publish_to_iot_core(
                "my/topic",
                b"Hello world!",
                Qos::AtMostOnce,
            )
        })
    }

    #[test]
    fn test_publish_to_iot_core_bad_alloc() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        run_ipc_handshake_test(|| {
            let sdk = Sdk::init();
            sdk.connect()?;
            let payload = [0u8; 0x20000];
            assert_eq!(
                sdk.publish_to_iot_core("my/topic", &payload, Qos::AtMostOnce),
                Err(Error::Nomem),
            );
            Ok(())
        })
    }

    #[test]
    fn test_publish_to_iot_core_rejected() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let topic = "my/topic";
        let payload_base64 = "SGVsbG8gd29ybGQh";
        let qos = "0";
        let seq = unsafe {
            c::gg_test_mqtt_publish_error_sequence(
                1,
                topic.into(),
                payload_base64.into(),
                qos.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            assert!(
                sdk.publish_to_iot_core(
                    "my/topic",
                    b"Hello world!",
                    Qos::AtMostOnce
                )
                .is_err()
            );
            Ok(())
        })
    }

    // NOTE: publish_to_iot_core_invalid_qos cannot be ported; the Rust Qos
    // enum prevents invalid values at compile time.

    #[test]
    fn test_get_thing_shadow_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let payload = "hello";
        let payload_b64 = "aGVsbG8=";
        let seq = unsafe {
            c::gg_test_shadow_get_accepted_sequence(
                1,
                thing_name.into(),
                shadow_name.into(),
                payload_b64.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            let mut buf = [std::mem::MaybeUninit::uninit(); 64];
            let result =
                sdk.get_thing_shadow(thing_name, Some(shadow_name), &mut buf)?;
            assert_eq!(result, payload.as_bytes());
            Ok(())
        })
    }

    #[test]
    fn test_get_thing_shadow_rejected() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let seq = unsafe {
            c::gg_test_shadow_get_error_sequence(
                1,
                thing_name.into(),
                shadow_name.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            let mut buf = [std::mem::MaybeUninit::uninit(); 64];
            assert!(sdk
                .get_thing_shadow(thing_name, Some(shadow_name), &mut buf)
                .is_err());
            Ok(())
        })
    }

    #[test]
    fn test_update_thing_shadow_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let payload = b"hello";
        let payload_b64 = "aGVsbG8=";
        let seq = unsafe {
            c::gg_test_shadow_update_accepted_sequence(
                1,
                thing_name.into(),
                shadow_name.into(),
                payload_b64.into(),
                payload_b64.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            sdk.update_thing_shadow(thing_name, Some(shadow_name), payload, None)?;
            Ok(())
        })
    }

    #[test]
    fn test_update_thing_shadow_rejected() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let payload = b"hello";
        let payload_b64 = "aGVsbG8=";
        let seq = unsafe {
            c::gg_test_shadow_update_error_sequence(
                1,
                thing_name.into(),
                shadow_name.into(),
                payload_b64.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            assert!(sdk
                .update_thing_shadow(thing_name, Some(shadow_name), payload, None)
                .is_err());
            Ok(())
        })
    }

    #[test]
    fn test_delete_thing_shadow_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let seq = unsafe {
            c::gg_test_shadow_delete_accepted_sequence(
                1,
                thing_name.into(),
                shadow_name.into(),
                "".into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            sdk.delete_thing_shadow(thing_name, Some(shadow_name))?;
            Ok(())
        })
    }

    #[test]
    fn test_delete_thing_shadow_rejected() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let seq = unsafe {
            c::gg_test_shadow_delete_error_sequence(
                1,
                thing_name.into(),
                shadow_name.into(),
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            assert!(sdk
                .delete_thing_shadow(thing_name, Some(shadow_name))
                .is_err());
            Ok(())
        })
    }

    #[test]
    fn test_list_named_shadows_okay() -> Result<()> {
        let thing_name = "MyThing";
        let shadow_name = "myShadow";
        let timestamp: f64 = 1773436831.0;

        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        unsafe {
            Result::from(c::gg_test_setup_ipc(
                c"/tmp/gg-test".as_ptr(),
                0o666,
                c"1234567890ABCDEF".as_ptr(),
            ))?;

            let pid = libc::fork();
            assert!(pid >= 0, "fork failed");

            if pid == 0 {
                let sdk = Sdk::init();
                sdk.connect().unwrap();
                let mut count = 0usize;
                sdk.list_named_shadows_for_thing(
                    thing_name,
                    &mut |name: &str| {
                        assert_eq!(name, "myShadow");
                        count += 1;
                    },
                )
                .unwrap();
                assert_eq!(count, 1);
                libc::exit(0);
            }

            let mut result_item = c::gg_obj_buf(shadow_name.into());
            let results = c::GgList {
                items: &mut result_item,
                len: 1,
            };

            Result::from(c::gg_test_accept_client_handshake(5))?;

            Result::from(c::gg_test_expect_packet_sequence(
                c::gg_test_shadow_list_accepted_sequence(
                    1,
                    thing_name.into(),
                    core::ptr::null_mut(),
                    results,
                    timestamp,
                    core::ptr::null_mut(),
                ),
                5,
            ))?;

            Result::from(c::gg_test_wait_for_client_disconnect(30))?;

            c::gg_test_close();

            let mut status = 0;
            libc::waitpid(pid, &mut status, 0);
            assert!(libc::WIFEXITED(status));
            assert_eq!(libc::WEXITSTATUS(status), 0);
        }

        Ok(())
    }

    #[test]
    fn test_list_named_shadows_rejected() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let thing_name = "MyThing";
        let seq = unsafe {
            c::gg_test_shadow_list_error_sequence(1, thing_name.into())
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;
            assert!(sdk
                .list_named_shadows_for_thing(thing_name, &mut |_: &str| {})
                .is_err());
            Ok(())
        })
    }

    #[test]
    fn test_list_named_shadows_paginated_okay() -> Result<()> {
        let thing_name = "MyThing";
        let timestamp: f64 = 1773436831.0;

        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        unsafe {
            Result::from(c::gg_test_setup_ipc(
                c"/tmp/gg-test".as_ptr(),
                0o666,
                c"1234567890ABCDEF".as_ptr(),
            ))?;

            let pid = libc::fork();
            assert!(pid >= 0, "fork failed");

            if pid == 0 {
                let sdk = Sdk::init();
                sdk.connect().unwrap();
                let mut count = 0usize;
                let expected = ["shadow1", "shadow2"];
                sdk.list_named_shadows_for_thing(
                    thing_name,
                    &mut |name: &str| {
                        assert_eq!(name, expected[count]);
                        count += 1;
                    },
                )
                .unwrap();
                assert_eq!(count, 2);
                libc::exit(0);
            }

            let mut next_token1: c::GgBuffer = "token123".into();
            let mut next_token2: c::GgBuffer = "token456".into();

            let mut page1_item = c::gg_obj_buf("shadow1".into());
            let page1 = c::GgList {
                items: &mut page1_item,
                len: 1,
            };

            let mut page2_item = c::gg_obj_buf("shadow2".into());
            let page2 = c::GgList {
                items: &mut page2_item,
                len: 1,
            };

            let empty = c::GgList {
                items: core::ptr::null_mut(),
                len: 0,
            };

            Result::from(c::gg_test_accept_client_handshake(5))?;

            Result::from(c::gg_test_expect_packet_sequence(
                c::gg_test_shadow_list_accepted_sequence(
                    1,
                    thing_name.into(),
                    core::ptr::null_mut(),
                    page1,
                    timestamp,
                    &mut next_token1,
                ),
                5,
            ))?;

            Result::from(c::gg_test_expect_packet_sequence(
                c::gg_test_shadow_list_accepted_sequence(
                    2,
                    thing_name.into(),
                    &mut next_token1,
                    page2,
                    timestamp,
                    &mut next_token2,
                ),
                5,
            ))?;

            Result::from(c::gg_test_expect_packet_sequence(
                c::gg_test_shadow_list_accepted_sequence(
                    3,
                    thing_name.into(),
                    &mut next_token2,
                    empty,
                    timestamp,
                    core::ptr::null_mut(),
                ),
                5,
            ))?;

            Result::from(c::gg_test_wait_for_client_disconnect(30))?;

            c::gg_test_close();

            let mut status = 0;
            libc::waitpid(pid, &mut status, 0);
            assert!(libc::WIFEXITED(status));
            assert_eq!(libc::WEXITSTATUS(status), 0);
        }

        Ok(())
    }

    #[test]
    fn test_subscribe_to_iot_core_okay() -> Result<()> {
        let _guard = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
        let topic = "my/topic";
        let payload_base64 = "SGVsbG8gd29ybGQh";
        let qos = "0";
        let expected_calls: usize = 3;
        let seq = unsafe {
            c::gg_test_mqtt_subscribe_accepted_sequence(
                1,
                topic.into(),
                payload_base64.into(),
                qos.into(),
                expected_calls,
            )
        };
        run_ipc_sequence_test(seq, || {
            let sdk = Sdk::init();
            sdk.connect()?;

            let pair = (Mutex::new(0usize), Condvar::new());
            let cb = |t: &str, p: &[u8]| {
                assert_eq!(t, "my/topic");
                assert_eq!(p, b"Hello world!");
                let mut count = pair.0.lock().unwrap();
                *count += 1;
                if *count >= expected_calls {
                    pair.1.notify_one();
                }
            };
            let sub =
                sdk.subscribe_to_iot_core("my/topic", Qos::AtMostOnce, &cb)?;

            let guard = pair.0.lock().unwrap();
            let (guard, timeout) = pair
                .1
                .wait_timeout_while(
                    guard,
                    std::time::Duration::from_secs(5),
                    |count| *count < expected_calls,
                )
                .unwrap();
            assert!(
                !timeout.timed_out(),
                "Timed out waiting for subscription responses"
            );
            assert_eq!(*guard, expected_calls);
            std::mem::forget(sub);
            Ok(())
        })
    }
}