Initial support of listen()

crates/console-host/src/main.rs

@@ -23,8 +23,8 @@ use color_eyre::owo_colors::OwoColorize;
 use moor_compiler::to_literal;
 use moor_values::Objid;
 use rpc_common::{
-    AuthToken, BroadcastEvent, ClientToken, ConnectionEvent, RpcRequest, RpcResponse, RpcResult,
-    BROADCAST_TOPIC,
+    AuthToken, ClientEvent, ClientToken, ClientsBroadcastEvent, DaemonToClientReply,
+    HostClientToDaemonMessage, ReplyResult, CLIENT_BROADCAST_TOPIC,
 };
 use rpc_sync_client::RpcSendClient;
 use rpc_sync_client::{broadcast_recv, events_recv};
@@ -75,14 +75,16 @@ fn establish_connection(
 ) -> Result<(ClientToken, Objid), Error> {
     match rpc_client.make_rpc_call(
         client_id,
-        RpcRequest::ConnectionEstablish("console".to_string()),
+        HostClientToDaemonMessage::ConnectionEstablish("console".to_string()),
     ) {
-        Ok(RpcResult::Success(RpcResponse::NewConnection(token, conn_id))) => Ok((token, conn_id)),
-        Ok(RpcResult::Success(response)) => {
+        Ok(ReplyResult::ClientSuccess(DaemonToClientReply::NewConnection(token, conn_id))) => {
+            Ok((token, conn_id))
+        }
+        Ok(ReplyResult::ClientSuccess(response)) => {
             error!(?response, "Unexpected response");
             Err(Error::msg("Unexpected response"))
         }
-        Ok(RpcResult::Failure(error)) => {
+        Ok(ReplyResult::Failure(error)) => {
             error!(?error, "Failure connecting");
             Err(Error::msg("Failure connecting"))
         }
@@ -103,7 +105,7 @@ fn perform_auth(
     // Need to first authenticate with the server.
     match rpc_client.make_rpc_call(
         client_id,
-        RpcRequest::LoginCommand(
+        HostClientToDaemonMessage::LoginCommand(
             token,
             vec![
                 "connect".to_string(),
@@ -113,23 +115,23 @@ fn perform_auth(
             true,
         ),
     ) {
-        Ok(RpcResult::Success(RpcResponse::LoginResult(Some((
+        Ok(ReplyResult::ClientSuccess(DaemonToClientReply::LoginResult(Some((
             auth_token,
             connect_type,
             player,
         ))))) => {
             info!(?connect_type, ?player, "Authenticated");
             Ok((auth_token, player))
         }
-        Ok(RpcResult::Success(RpcResponse::LoginResult(None))) => {
+        Ok(ReplyResult::ClientSuccess(DaemonToClientReply::LoginResult(None))) => {
             error!("Authentication failed");
             Err(Error::msg("Authentication failed"))
         }
-        Ok(RpcResult::Success(response)) => {
+        Ok(ReplyResult::ClientSuccess(response)) => {
             error!(?response, "Unexpected response");
             Err(Error::msg("Unexpected response"))
         }
-        Ok(RpcResult::Failure(failure)) => {
+        Ok(ReplyResult::Failure(failure)) => {
             error!(?failure, "Failure authenticating");
             Err(Error::msg("Failure authenticating"))
         }
@@ -152,20 +154,20 @@ fn handle_console_line(
     if let Some(input_request_id) = input_request_id {
         match rpc_client.make_rpc_call(
             client_id,
-            RpcRequest::RequestedInput(
+            HostClientToDaemonMessage::RequestedInput(
                 client_token.clone(),
                 auth_token.clone(),
                 input_request_id.as_u128(),
                 line.to_string(),
             ),
         ) {
-            Ok(RpcResult::Success(RpcResponse::InputThanks)) => {
+            Ok(ReplyResult::ClientSuccess(DaemonToClientReply::InputThanks)) => {
                 trace!("Input complete");
             }
-            Ok(RpcResult::Success(response)) => {
+            Ok(ReplyResult::ClientSuccess(response)) => {
                 warn!(?response, "Unexpected input response");
             }
-            Ok(RpcResult::Failure(error)) => {
+            Ok(ReplyResult::Failure(error)) => {
                 error!(?error, "Failure executing input");
             }
             Err(error) => {
@@ -177,15 +179,19 @@ fn handle_console_line(
 
     match rpc_client.make_rpc_call(
         client_id,
-        RpcRequest::Command(client_token.clone(), auth_token.clone(), line.to_string()),
+        HostClientToDaemonMessage::Command(
+            client_token.clone(),
+            auth_token.clone(),
+            line.to_string(),
+        ),
     ) {
-        Ok(RpcResult::Success(RpcResponse::CommandSubmitted(_))) => {
+        Ok(ReplyResult::ClientSuccess(DaemonToClientReply::CommandSubmitted(_))) => {
             trace!("Command complete");
         }
-        Ok(RpcResult::Success(response)) => {
+        Ok(ReplyResult::ClientSuccess(response)) => {
             warn!(?response, "Unexpected command response");
         }
-        Ok(RpcResult::Failure(error)) => {
+        Ok(ReplyResult::Failure(error)) => {
             error!(?error, "Failure executing command");
         }
         Err(error) => {
@@ -248,7 +254,7 @@ fn console_loop(
                 return;
             }
             match events_recv(client_id, &narr_sub_socket) {
-                Ok(ConnectionEvent::Narrative(_, msg)) => {
+                Ok(ClientEvent::Narrative(_, msg)) => {
                     let var = match msg.event() {
                         moor_values::tasks::Event::Notify(s, _content_type) => s,
                     };
@@ -262,21 +268,21 @@ fn console_loop(
                         }
                     }
                 }
-                Ok(ConnectionEvent::SystemMessage(o, msg)) => {
+                Ok(ClientEvent::SystemMessage(o, msg)) => {
                     printer
                         .print(format!("System message from {}: {}", o.yellow(), msg.red()))
                         .unwrap();
                 }
-                Ok(ConnectionEvent::Disconnect()) => {
+                Ok(ClientEvent::Disconnect()) => {
                     printer
                         .print("Received disconnect event; Session ending.".to_string())
                         .unwrap();
                     return;
                 }
-                Ok(ConnectionEvent::TaskError(e)) => {
+                Ok(ClientEvent::TaskError(e)) => {
                     printer.print(format!("Error: {:?}", e)).unwrap();
                 }
-                Ok(ConnectionEvent::TaskSuccess(result)) => {
+                Ok(ClientEvent::TaskSuccess(result)) => {
                     printer.print(format!("=> {:?}", result)).unwrap();
                 }
                 Err(error) => {
@@ -288,7 +294,7 @@ fn console_loop(
                         .unwrap();
                     return;
                 }
-                Ok(ConnectionEvent::RequestInput(requested_input_id)) => {
+                Ok(ClientEvent::RequestInput(requested_input_id)) => {
                     *output_input_request_id.lock().unwrap() =
                         Some(Uuid::from_u128(requested_input_id));
                 }
@@ -297,7 +303,7 @@ fn console_loop(
 
     let mut broadcast_subscriber = zmq_ctx.socket(zmq::SUB)?;
     broadcast_subscriber.connect(narrative_server)?;
-    broadcast_subscriber.set_subscribe(BROADCAST_TOPIC)?;
+    broadcast_subscriber.set_subscribe(CLIENT_BROADCAST_TOPIC)?;
 
     let broadcast_rpc_socket = zmq_ctx.socket(zmq::REQ)?;
     broadcast_rpc_socket.connect(rpc_server)?;
@@ -310,10 +316,13 @@ fn console_loop(
             return;
         }
         match broadcast_recv(&mut broadcast_subscriber) {
-            Ok(BroadcastEvent::PingPong(_)) => {
+            Ok(ClientsBroadcastEvent::PingPong(_)) => {
                 if let Err(e) = broadcast_rpc_client.make_rpc_call(
                     client_id,
-                    RpcRequest::Pong(broadcast_client_token.clone(), SystemTime::now()),
+                    HostClientToDaemonMessage::ClientPong(
+                        broadcast_client_token.clone(),
+                        SystemTime::now(),
+                    ),
                 ) {
                     error!("Error sending pong: {:?}", e);
                     return;

crates/daemon/src/connections.rs

@@ -18,7 +18,7 @@ use uuid::Uuid;
 
 use moor_kernel::tasks::sessions::SessionError;
 use moor_values::Objid;
-use rpc_common::RpcRequestError;
+use rpc_common::RpcMessageError;
 
 pub const CONNECTION_TIMEOUT_DURATION: Duration = Duration::from_secs(30);
 
@@ -37,7 +37,7 @@ pub trait ConnectionsDB {
         client_id: Uuid,
         hostname: String,
         player: Option<Objid>,
-    ) -> Result<Objid, RpcRequestError>;
+    ) -> Result<Objid, RpcMessageError>;
 
     /// Record activity for the given client.
     fn record_client_activity(&self, client_id: Uuid, connobj: Objid) -> Result<(), eyre::Error>;

crates/daemon/src/connections_rb.rs

@@ -30,7 +30,7 @@ use moor_kernel::tasks::sessions::SessionError;
 use moor_values::AsByteBuffer;
 use moor_values::Objid;
 use relbox::{relation_info_for, RelBox, RelationId, RelationInfo, Transaction};
-use rpc_common::RpcRequestError;
+use rpc_common::RpcMessageError;
 
 use crate::connections::{ConnectionsDB, CONNECTION_TIMEOUT_DURATION};
 
@@ -191,7 +191,7 @@ impl ConnectionsDB for ConnectionsRb {
         client_id: Uuid,
         hostname: String,
         player: Option<Objid>,
-    ) -> Result<Objid, RpcRequestError> {
+    ) -> Result<Objid, RpcMessageError> {
         let connection_oid = match player {
             None => {
                 // The connection object is pulled from the sequence, then we invert it and subtract from

crates/daemon/src/connections_wt.rs

@@ -33,7 +33,7 @@ use moor_kernel::tasks::sessions::SessionError;
 use moor_values::model::{CommitResult, ValSet};
 use moor_values::Objid;
 use moor_values::{AsByteBuffer, DecodingError, EncodingError};
-use rpc_common::RpcRequestError;
+use rpc_common::RpcMessageError;
 
 use crate::connections::{ConnectionsDB, CONNECTION_TIMEOUT_DURATION};
 use crate::connections_wt::ConnectionRelation::{
@@ -280,7 +280,7 @@ impl ConnectionsDB for ConnectionsWT {
         client_id: Uuid,
         hostname: String,
         player: Option<Objid>,
-    ) -> Result<Objid, RpcRequestError> {
+    ) -> Result<Objid, RpcMessageError> {
         retry_tx_action(&self.db, |tx| {
             let connection_oid = match player {
                 None => {
@@ -304,7 +304,7 @@ impl ConnectionsDB for ConnectionsWT {
 
             Ok(connection_oid)
         })
-        .map_err(|e| RpcRequestError::InternalError(e.to_string()))
+        .map_err(|e| RpcMessageError::InternalError(e.to_string()))
     }
 
     fn record_client_activity(&self, client_id: Uuid, _connobj: Objid) -> Result<(), Error> {

crates/daemon/src/main.rs

@@ -128,15 +128,15 @@ struct Args {
     #[arg(
         long,
         value_name = "public_key",
-        help = "file containing a pkcs8 ed25519 public key, used for authenticating client connections",
+        help = "file containing a pkcs8 ed25519 public key, used for authenticating client & host connections",
         default_value = "public_key.pem"
     )]
     public_key: PathBuf,
 
     #[arg(
         long,
         value_name = "private_key",
-        help = "file containing a pkcs8 ed25519 private key, used for authenticating client connections",
+        help = "file containing a pkcs8 ed25519 private key, used for authenticating client & host connections",
         default_value = "private_key.pem"
     )]
     private_key: PathBuf,
@@ -355,12 +355,6 @@ fn main() -> Result<(), Report> {
         }
     };
 
-    // The pieces from core we're going to use:
-    //   Our DB.
-    //   Our scheduler.
-    let scheduler = Scheduler::new(database, tasks_db, config.clone());
-    let scheduler_client = scheduler.client().expect("Failed to get scheduler client");
-
     // We have to create the RpcServer before starting the scheduler because we need to pass it in
     // as a parameter to the scheduler for background session construction.
 
@@ -374,17 +368,22 @@ fn main() -> Result<(), Report> {
         zmq_ctx.clone(),
         args.events_listen.as_str(),
         args.db_flavour,
-        config,
+        config.clone(),
     ));
+    let kill_switch = rpc_server.kill_switch();
+
+    // The pieces from core we're going to use:
+    //   Our DB.
+    //   Our scheduler.
+    let scheduler = Scheduler::new(database, tasks_db, config, rpc_server.clone());
+    let scheduler_client = scheduler.client().expect("Failed to get scheduler client");
 
     // The scheduler thread:
     let scheduler_rpc_server = rpc_server.clone();
     let scheduler_loop_jh = std::thread::Builder::new()
         .name("moor-scheduler".to_string())
         .spawn(move || scheduler.run(scheduler_rpc_server))?;
 
-    let kill_switch = Arc::new(std::sync::atomic::AtomicBool::new(false));
-
     // Background DB checkpoint thread.
     let checkpoint_kill_switch = kill_switch.clone();
     let checkpoint_scheduler_client = scheduler_client.clone();
@@ -402,15 +401,13 @@ fn main() -> Result<(), Report> {
                 .expect("Failed to submit checkpoint");
         })?;
 
-    let rpc_kill_switch = kill_switch.clone();
-
     let rpc_loop_scheduler_client = scheduler_client.clone();
     let rpc_listen = args.rpc_listen.clone();
     let rpc_loop_thread = std::thread::Builder::new()
         .name("moor-rpc".to_string())
         .spawn(move || {
             rpc_server
-                .zmq_loop(rpc_listen, rpc_loop_scheduler_client, rpc_kill_switch)
+                .request_loop(rpc_listen, rpc_loop_scheduler_client)
                 .expect("RPC thread failed");
         })?;