liminal-server 0.25.4

//! Local responder selection without an implicit server for an unserved subject.

use std::sync::mpsc;
use std::time::{Duration, Instant};

use liminal::channel::ChannelMode;
use liminal::conversation::CrashPolicy;
use liminal::protocol::MessageEnvelope;

use super::LiminalConnectionServices;
use crate::ServerError;
use crate::server::connection::conversation::{
    ConnectionConversation, ConversationResource, LiminalConversationResource,
};

impl LiminalConnectionServices {
    pub(super) fn open_routed_conversation(
        &self,
        conversation_id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        let behaviour = self.lock_responders()?.get(subject).cloned();
        let Some(behaviour) = behaviour else {
            // Keep the subject binding for worker and built-in routes, which
            // are resolved per request before the local participant route.
            return Ok(ConnectionConversation::new(
                Box::new(UnservedConversation {
                    subject: subject.to_owned(),
                }),
                subject.to_owned(),
            ));
        };
        let (actor, participant) = self
            .conversation_supervisor
            .spawn_with_participant(behaviour, None, ChannelMode::Ephemeral, CrashPolicy::Fail)
            .map_err(|error| ServerError::ListenerAccept {
                message: format!(
                    "failed to spawn supervised conversation {conversation_id} ('{subject}'): {error}"
                ),
            })?;

        // Link the participant before forwarding and arm its structural exit
        // signal before publishing the conversation to the connection.
        actor.pid().map_err(|error| ServerError::ListenerAccept {
            message: format!(
                "failed to boot supervised conversation {conversation_id} ('{subject}'): {error}"
            ),
        })?;
        let (exit_tx, exit_rx) = mpsc::sync_channel::<Instant>(1);
        actor
            .notify_on_participant_exit(participant, exit_tx)
            .map_err(|error| ServerError::ListenerAccept {
                message: format!(
                    "failed to arm crash detection for conversation {conversation_id}: {error}"
                ),
            })?;

        Ok(ConnectionConversation::new(
            Box::new(LiminalConversationResource::new(
                actor,
                participant,
                exit_rx,
            )),
            subject.to_owned(),
        ))
    }

    pub(super) fn routed_conversation_message(
        &self,
        conversation: &ConnectionConversation,
        envelope: &MessageEnvelope,
        op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        // Removal takes effect for already-open local conversations too.
        // Worker and built-in routes resolve before entering this method.
        if !self.lock_responders()?.contains_key(conversation.subject()) {
            return Err(no_server(conversation.subject()));
        }
        conversation.message(envelope, op_id)
    }
}

#[derive(Debug)]
struct UnservedConversation {
    subject: String,
}

impl UnservedConversation {
    fn refusal(&self) -> ServerError {
        no_server(&self.subject)
    }
}

fn no_server(subject: &str) -> ServerError {
    // Conversation errors carry their refusal name in the message under the
    // existing server-error reason code.
    ServerError::ListenerAccept {
        message: format!("no_server: no live server serves {subject}"),
    }
}

impl ConversationResource for UnservedConversation {
    fn message(&self, _: &MessageEnvelope, _: Option<u64>) -> Result<(), ServerError> {
        Err(self.refusal())
    }

    fn participant_pids(&self) -> Vec<u64> {
        Vec::new()
    }

    fn has_detected_crash(&self) -> bool {
        false
    }

    fn await_crash(&self, _: Duration) -> Option<Instant> {
        None
    }

    fn receive_reply(&self, _: Duration) -> Result<MessageEnvelope, ServerError> {
        Err(self.refusal())
    }

    fn close(self: Box<Self>) -> Result<(), ServerError> {
        Ok(())
    }

    fn finalize(self: Box<Self>) {}
}