iceoryx2 0.10.0

// Copyright (c) 2025 Contributors to the Eclipse Foundation
//
// See the NOTICE file(s) distributed with this work for additional
// information regarding copyright ownership.
//
// This program and the accompanying materials are made available under the
// terms of the Apache Software License 2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0, or the MIT license
// which is available at https://opensource.org/licenses/MIT.
//
// SPDX-License-Identifier: Apache-2.0 OR MIT

//! # Example
//!
//! ```
//! use iceoryx2::prelude::*;
//!
//! # fn main() -> Result<(), Box<dyn core::error::Error>> {
//! # let node = NodeBuilder::new().create::<ipc::Service>()?;
//! #
//! # let service = node
//! #    .service_builder(&"My/Funk/ServiceName".try_into()?)
//! #    .request_response::<u64, u64>()
//! #    .open_or_create()?;
//! #
//! # let client = service.client_builder().create()?;
//!
//! # let request = client.loan_uninit()?;
//! # let request = request.write_payload(0);
//!
//! let pending_response = request.send()?;
//!
//! println!("send request to {} server",
//!           pending_response.number_of_server_connections());
//!
//! // we receive a stream of responses from the server and are interested in 5 of them
//! for i in 0..5 {
//!     if !pending_response.is_connected() {
//!         println!("server terminated connection - abort");
//!         break;
//!     }
//!
//!     if let Some(response) = pending_response.receive()? {
//!         println!("received response: {}", *response);
//!     }
//! }
//!
//! // We are no longer interested in the responses from the server and
//! // drop the object. This informs the corresponding servers, that hold
//! // an ActiveRequest that the connection was terminated from the client
//! // side so that they can stop sending responses.
//! drop(pending_response);
//!
//! # Ok(())
//! # }
//! ```

use core::ops::Deref;
use core::{fmt::Debug, marker::PhantomData};

use flatbuffers::InvalidFlatbuffer;
use iceoryx2_bb_concurrency::atomic::Ordering;
use iceoryx2_bb_elementary_traits::iceoryx_send::IceoryxSend;
use iceoryx2_bb_elementary_traits::testing::abandonable::Abandonable;
use iceoryx2_bb_elementary_traits::zero_copy_send::ZeroCopySend;
use iceoryx2_bb_flatbuffers::FlatbufferError;
use iceoryx2_log::fail;

use crate::port::client::ClientSharedState;
use crate::port::details::chunk::Chunk;
use crate::port::details::chunk_details::ChunkDetails;
use crate::service::marker::{CustomPayloadMarker, Flatbuffer};
use crate::{port::ReceiveError, request_mut::RequestMut, response::Response, service};

/// Represents an active connection to all [`Server`](crate::port::server::Server)
/// that received the [`RequestMut`]. The
/// [`Client`](crate::port::client::Client) can use it to receive the corresponding
/// [`Response`]s.
///
/// As soon as it goes out of scope, the connections are closed and the
/// [`Server`](crate::port::server::Server)s are informed.
pub struct PendingResponse<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> {
    pub(crate) request:
        RequestMut<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>,
    pub(crate) number_of_server_connections: usize,
    pub(crate) _service: PhantomData<Service>,
    pub(crate) _response_payload: PhantomData<ResponsePayload>,
    pub(crate) _response_header: PhantomData<ResponseHeader>,
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> Abandonable
    for PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    unsafe fn abandon_in_place(mut this: core::ptr::NonNull<Self>) {
        let this = unsafe { this.as_mut() };
        unsafe { RequestMut::abandon_in_place(core::ptr::NonNull::from_mut(&mut this.request)) };
    }
}

unsafe impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> Send for PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
where
    Service::ArcThreadSafetyPolicy<ClientSharedState<Service>>: Send + Sync,
{
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> Drop
    for PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    fn drop(&mut self) {
        let _ = self.request.shared_state.call(|s| -> Result<(), ()> {
            s.active_request_counter.fetch_sub(1, Ordering::Relaxed);

            Ok(())
        });
        self.close();
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ZeroCopySend,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> Deref
    for PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    type Target = RequestPayload;
    fn deref(&self) -> &Self::Target {
        self.request.payload()
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ZeroCopySend,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> Deref
    for PendingResponse<Service, [RequestPayload], RequestHeader, ResponsePayload, ResponseHeader>
{
    type Target = [RequestPayload];
    fn deref(&self) -> &Self::Target {
        self.request.payload()
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> Debug
    for PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(
            f,
            "PendingResponse<{}, {}, {}, {}, {}> {{ number_of_server_connections: {} }}",
            core::any::type_name::<Service>(),
            core::any::type_name::<RequestPayload>(),
            core::any::type_name::<RequestHeader>(),
            core::any::type_name::<ResponsePayload>(),
            core::any::type_name::<ResponseHeader>(),
            self.number_of_server_connections
        )
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    fn close(&self) {
        let _ = self.request.shared_state.call(|s| -> Result<(), ()> {
            s.response_receiver
                .close_channel(self.request.channel_id, self.request.header().request_id);

            Ok(())
        });
    }

    /// Marks the connection state that the [`Client`](crate::port::client::Client) wants to gracefully
    /// disconnect. When the [`Server`](crate::port::server::Server) reads this, it can send the last
    /// [`Response`] and drop the corresponding [`ActiveRequest`](crate::active_request::ActiveRequest) to
    /// terminate the connection ensuring that no [`Response`] is lost on the
    /// [`Client`](crate::port::client::Client) side.
    pub fn set_disconnect_hint(&self) {
        let _ = self.request.shared_state.call(|s| -> Result<(), ()> {
            s.response_receiver
                .set_disconnect_hint(self.request.channel_id, self.request.header().request_id);

            Ok(())
        });
    }

    /// Returns [`true`] until the [`ActiveRequest`](crate::active_request::ActiveRequest)
    /// goes out of scope on the [`Server`](crate::port::server::Server)s side indicating that the
    /// [`Server`](crate::port::server::Server) will no longer send [`Response`]s.
    /// It also returns [`false`] when there are no [`Server`](crate::port::server::Server)s.
    pub fn is_connected(&self) -> bool {
        self.request
            .shared_state
            .call(|s| -> Result<bool, ()> {
                Ok(s.response_receiver.at_least_one_channel_has_state(
                    self.request.channel_id,
                    self.request.header().request_id,
                ))
            })
            .expect("Always returns ok")
    }

    /// Returns a reference to the iceoryx2 internal
    /// [`service::header::request_response::RequestHeader`] of the corresponding
    /// [`RequestMut`]
    pub fn header(&self) -> &service::header::request_response::RequestHeader {
        self.request.header()
    }

    /// Returns a reference to the user defined request header of the corresponding
    /// [`RequestMut`]
    pub fn user_header(&self) -> &RequestHeader {
        self.request.user_header()
    }

    /// Returns how many [`Server`](crate::port::server::Server)s received the corresponding
    /// [`RequestMut`] initially.
    pub fn number_of_server_connections(&self) -> usize {
        self.number_of_server_connections
    }

    /// Returns [`true`] when a [`Server`](crate::port::server::Server) has sent a [`Response`]
    /// otherwise [`false`].
    pub fn has_response(&self) -> bool {
        self.request
            .shared_state
            .call(|s| -> Result<bool, ()> {
                Ok(s.response_receiver.has_chunks(self.request.channel_id))
            })
            .expect("Always returns ok")
    }

    fn receive_impl(&self) -> Result<Option<(ChunkDetails, Chunk)>, ReceiveError> {
        self.request.shared_state.call(|s| {
            let msg = "Unable to receive response";
            fail!(from self, when s.update_connections(),
                    "{msg} since the connections could not be updated.");

            s.response_receiver.receive(self.request.channel_id)
        })
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ZeroCopySend,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    /// Returns a reference to the request payload of the corresponding
    /// [`RequestMut`]
    pub fn payload(&self) -> &RequestPayload {
        self.request.payload()
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ZeroCopySend,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
> PendingResponse<Service, [RequestPayload], RequestHeader, ResponsePayload, ResponseHeader>
{
    /// Returns a reference to the request payload of the corresponding
    /// [`RequestMut`]
    pub fn payload(&self) -> &[RequestPayload] {
        self.request.payload()
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + ?Sized,
    ResponseHeader: Debug + ZeroCopySend,
>
    PendingResponse<
        Service,
        Flatbuffer<RequestPayload>,
        RequestHeader,
        ResponsePayload,
        ResponseHeader,
    >
{
    /// Returns the serialized flatbuffer data as bytes.
    pub fn payload_bytes(&self) -> &[u8] {
        self.request.payload_bytes()
    }

    /// Returns the root of the flatbuffer.
    pub fn payload_root<'a>(
        &'a self,
    ) -> Result<RequestPayload::Inner, FlatbufferError<InvalidFlatbuffer>>
    where
        RequestPayload: flatbuffers::Follow<'a> + flatbuffers::Verifiable,
    {
        self.request.payload_root()
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + IceoryxSend + Sized,
    ResponseHeader: Debug + ZeroCopySend,
> PendingResponse<Service, RequestPayload, RequestHeader, ResponsePayload, ResponseHeader>
{
    /// Receives a [`Response`] from one of the [`Server`](crate::port::server::Server)s that
    /// received the [`RequestMut`].
    ///
    /// # Example
    ///
    /// ```
    /// use iceoryx2::prelude::*;
    ///
    /// # fn main() -> Result<(), Box<dyn core::error::Error>> {
    /// # let node = NodeBuilder::new().create::<ipc::Service>()?;
    /// #
    /// # let service = node
    /// #    .service_builder(&"My/Funk/ServiceName".try_into()?)
    /// #    .request_response::<u64, u64>()
    /// #    .open_or_create()?;
    /// #
    /// # let client = service.client_builder().create()?;
    ///
    /// # let request = client.loan_uninit()?;
    /// # let request = request.write_payload(0);
    ///
    /// let pending_response = request.send()?;
    ///
    /// if let Some(response) = pending_response.receive()? {
    ///     println!("received response: {}", *response);
    /// }
    ///
    /// # Ok(())
    /// # }
    /// ```
    pub fn receive(
        &self,
    ) -> Result<Option<Response<Service, ResponsePayload, ResponseHeader>>, ReceiveError> {
        loop {
            match self.receive_impl()? {
                None => return Ok(None),
                Some((details, chunk)) => {
                    let response = Response::<Service, ResponsePayload, ResponseHeader> {
                        details,
                        client_shared_state: self.request.shared_state.clone_shared_port_state(),
                        channel_id: self.request.channel_id,
                        chunk,
                        _response_header: PhantomData,
                        _response_payload: PhantomData,
                    };

                    if response.header().request_id != self.request.header().request_id {
                        continue;
                    }

                    return Ok(Some(response));
                }
            }
        }
    }
}

impl<
    Service: crate::service::Service,
    RequestPayload: Debug + IceoryxSend + ?Sized,
    RequestHeader: Debug + ZeroCopySend,
    ResponsePayload: Debug + ZeroCopySend,
    ResponseHeader: Debug + ZeroCopySend,
> PendingResponse<Service, RequestPayload, RequestHeader, [ResponsePayload], ResponseHeader>
{
    /// Receives a [`Response`] from one of the [`Server`](crate::port::server::Server)s that
    /// received the [`RequestMut`].
    ///
    /// # Example
    ///
    /// ```
    /// use iceoryx2::prelude::*;
    ///
    /// # fn main() -> Result<(), Box<dyn core::error::Error>> {
    /// # let node = NodeBuilder::new().create::<ipc::Service>()?;
    /// #
    /// # let service = node
    /// #    .service_builder(&"My/Funk/ServiceName".try_into()?)
    /// #    .request_response::<u64, [usize]>()
    /// #    .open_or_create()?;
    /// #
    /// # let client = service.client_builder().create()?;
    ///
    /// # let request = client.loan_uninit()?;
    /// # let request = request.write_payload(0);
    ///
    /// let pending_response = request.send()?;
    ///
    /// if let Some(response) = pending_response.receive()? {
    ///     println!("received response: {:?}", response);
    /// }
    ///
    /// # Ok(())
    /// # }
    /// ```
    pub fn receive(
        &self,
    ) -> Result<Option<Response<Service, [ResponsePayload], ResponseHeader>>, ReceiveError> {
        loop {
            match self.receive_impl()? {
                None => return Ok(None),
                Some((details, chunk)) => {
                    let response = Response::<Service, [ResponsePayload], ResponseHeader> {
                        details,
                        channel_id: self.request.channel_id,
                        client_shared_state: self.request.shared_state.clone_shared_port_state(),
                        chunk,
                        _response_header: PhantomData,
                        _response_payload: PhantomData,
                    };

                    if response.header().request_id != self.request.header().request_id {
                        continue;
                    }

                    return Ok(Some(response));
                }
            }
        }
    }
}

impl<
    Service: crate::service::Service,
    RequestHeader: Debug + ZeroCopySend,
    ResponseHeader: Debug + ZeroCopySend,
>
    PendingResponse<
        Service,
        [CustomPayloadMarker],
        RequestHeader,
        [CustomPayloadMarker],
        ResponseHeader,
    >
{
    #[doc(hidden)]
    pub unsafe fn receive_custom_payload(
        &self,
    ) -> Result<Option<Response<Service, [CustomPayloadMarker], ResponseHeader>>, ReceiveError>
    {
        loop {
            match self.receive_impl()? {
                None => return Ok(None),
                Some((details, chunk)) => {
                    let response = Response::<Service, [CustomPayloadMarker], ResponseHeader> {
                        details,
                        channel_id: self.request.channel_id,
                        client_shared_state: self.request.shared_state.clone_shared_port_state(),
                        chunk,
                        _response_header: PhantomData,
                        _response_payload: PhantomData,
                    };

                    if response.header().request_id != self.request.header().request_id {
                        continue;
                    }

                    return Ok(Some(response));
                }
            }
        }
    }
}