theater_cli/client/
theater_client.rs

1use std::net::SocketAddr;
2use std::sync::Arc;
3use tokio::sync::Mutex;
4use tracing::{debug, info};
5use uuid::Uuid;
6
7use theater::{messages::ChannelParticipant, ChainEvent};
8use theater_server::{ManagementCommand, ManagementResponse};
9
10
11use crate::error::{CliError, CliResult};
12use theater_client::TheaterConnection;
13
14/// High-level client for Theater server operations
15#[derive(Debug, Clone)]
16pub struct TheaterClient {
17    connection: Arc<Mutex<TheaterConnection>>,
18}
19
20impl TheaterClient {
21    /// Create a new TheaterClient
22    pub fn new(address: SocketAddr) -> Self {
23        let connection = TheaterConnection::new(address);
24        Self {
25            connection: Arc::new(Mutex::new(connection)),
26        }
27    }
28
29    /// Get the server address
30    pub async fn address(&self) -> SocketAddr {
31        let conn = self.connection.lock().await;
32        conn.address
33    }
34
35    /// Check if connected to the server
36    pub async fn is_connected(&self) -> bool {
37        let conn = self.connection.lock().await;
38        conn.is_connected()
39    }
40
41    /// Explicitly connect to the server (usually not needed as commands auto-connect)
42    pub async fn connect(&self) -> CliResult<()> {
43        let mut conn = self.connection.lock().await;
44        conn.connect()
45            .await
46            .map_err(|e| CliError::ConnectionFailed {
47                address: conn.address,
48                source: e,
49            })
50    }
51
52    /// Close the connection
53    pub async fn close(&self) {
54        let mut conn = self.connection.lock().await;
55        let _ = conn.close().await;
56    }
57
58    /// List all running actors
59    pub async fn list_actors(&self) -> CliResult<Vec<(String, String)>> {
60        let mut conn = self.connection.lock().await;
61        let response = conn.send_and_receive(ManagementCommand::ListActors).await?;
62
63        match response {
64            ManagementResponse::ActorList { actors } => {
65                debug!("Listed {} actors", actors.len());
66                // Convert TheaterId to String for the CLI layer
67                let string_actors: Vec<(String, String)> = actors
68                    .into_iter()
69                    .map(|(id, status)| (id.to_string(), status))
70                    .collect();
71                Ok(string_actors)
72            }
73            ManagementResponse::Error { error } => Err(CliError::ServerError {
74                message: format!("{:?}", error),
75            }),
76            _ => Err(CliError::UnexpectedResponse {
77                response: format!("{:?}", response),
78            }),
79        }
80    }
81
82    /// Start an actor from a manifest
83    pub async fn start_actor(
84        &self,
85        manifest_content: String,
86        initial_state: Option<Vec<u8>>,
87        parent: bool,
88        subscribe: bool,
89    ) -> CliResult<()> {
90        let mut conn = self.connection.lock().await;
91        conn.send(ManagementCommand::StartActor {
92            manifest: manifest_content,
93            initial_state,
94            parent,
95            subscribe,
96        })
97        .await
98        .map_err(|e| CliError::ConnectionFailed {
99            address: conn.address,
100            source: e,
101        })
102    }
103
104    /// Stop a running actor
105    pub async fn stop_actor(&self, actor_id: &str) -> CliResult<()> {
106        let mut conn = self.connection.lock().await;
107        let theater_id = actor_id
108            .parse()
109            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
110        let response = conn
111            .send_and_receive(ManagementCommand::StopActor { id: theater_id })
112            .await?;
113
114        match response {
115            ManagementResponse::ActorStopped { id: _ } => {
116                info!("Actor {} stopped successfully", actor_id);
117                Ok(())
118            }
119            ManagementResponse::Error { error } => {
120                let error_str = format!("{:?}", error);
121                if error_str.contains("not found") {
122                    Err(CliError::actor_not_found(actor_id))
123                } else {
124                    Err(CliError::ServerError { message: error_str })
125                }
126            }
127            _ => Err(CliError::UnexpectedResponse {
128                response: format!("{:?}", response),
129            }),
130        }
131    }
132
133    /// Get actor state
134    pub async fn get_actor_state(&self, actor_id: &str) -> CliResult<serde_json::Value> {
135        let mut conn = self.connection.lock().await;
136        let theater_id = actor_id
137            .parse()
138            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
139        let response = conn
140            .send_and_receive(ManagementCommand::GetActorState { id: theater_id })
141            .await?;
142
143        match response {
144            ManagementResponse::ActorState { id: _, state } => {
145                // Convert Vec<u8> state to JSON Value if present
146                match state {
147                    Some(bytes) => {
148                        serde_json::from_slice(&bytes).map_err(|e| CliError::ParseError {
149                            message: format!("Failed to parse actor state as JSON: {}", e),
150                        })
151                    }
152                    None => Ok(serde_json::Value::Null),
153                }
154            }
155            ManagementResponse::Error { error } => {
156                let error_str = format!("{:?}", error);
157                if error_str.contains("not found") {
158                    Err(CliError::actor_not_found(actor_id))
159                } else {
160                    Err(CliError::ServerError { message: error_str })
161                }
162            }
163            _ => Err(CliError::UnexpectedResponse {
164                response: format!("{:?}", response),
165            }),
166        }
167    }
168
169    /// Get actor events
170    pub async fn get_actor_events(&self, actor_id: &str) -> CliResult<Vec<ChainEvent>> {
171        let mut conn = self.connection.lock().await;
172        let theater_id = actor_id
173            .parse()
174            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
175        let response = conn
176            .send_and_receive(ManagementCommand::GetActorEvents { id: theater_id })
177            .await?;
178
179        match response {
180            ManagementResponse::ActorEvents { id: _, events } => {
181                debug!("Retrieved {} events for actor {}", events.len(), actor_id);
182                Ok(events)
183            }
184            ManagementResponse::Error { error } => {
185                let error_str = format!("{:?}", error);
186                if error_str.contains("not found") {
187                    Err(CliError::actor_not_found(actor_id))
188                } else {
189                    Err(CliError::ServerError { message: error_str })
190                }
191            }
192            _ => Err(CliError::UnexpectedResponse {
193                response: format!("{:?}", response),
194            }),
195        }
196    }
197
198    /// Send a message to an actor (fire and forget)
199    pub async fn send_message(&self, actor_id: &str, message: Vec<u8>) -> CliResult<()> {
200        let mut conn = self.connection.lock().await;
201        let theater_id = actor_id
202            .parse()
203            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
204        let response = conn
205            .send_and_receive(ManagementCommand::SendActorMessage {
206                id: theater_id,
207                data: message,
208            })
209            .await?;
210
211        match response {
212            ManagementResponse::SentMessage { id: _ } => Ok(()),
213            ManagementResponse::Error { error } => {
214                let error_str = format!("{:?}", error);
215                if error_str.contains("not found") {
216                    Err(CliError::actor_not_found(actor_id))
217                } else {
218                    Err(CliError::ServerError { message: error_str })
219                }
220            }
221            _ => Err(CliError::UnexpectedResponse {
222                response: format!("{:?}", response),
223            }),
224        }
225    }
226
227    /// Send a request to an actor and wait for response
228    pub async fn request_message(&self, actor_id: &str, message: Vec<u8>) -> CliResult<Vec<u8>> {
229        let mut conn = self.connection.lock().await;
230        let theater_id = actor_id
231            .parse()
232            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
233        let response = conn
234            .send_and_receive(ManagementCommand::RequestActorMessage {
235                id: theater_id,
236                data: message,
237            })
238            .await?;
239
240        match response {
241            ManagementResponse::RequestedMessage { id: _, message } => Ok(message),
242            ManagementResponse::Error { error } => {
243                let error_str = format!("{:?}", error);
244                if error_str.contains("not found") {
245                    Err(CliError::actor_not_found(actor_id))
246                } else {
247                    Err(CliError::ServerError { message: error_str })
248                }
249            }
250            _ => Err(CliError::UnexpectedResponse {
251                response: format!("{:?}", response),
252            }),
253        }
254    }
255
256    /// Subscribe to events from an actor (returns a stream-like interface)
257    pub async fn subscribe_to_events(&self, actor_id: &str) -> CliResult<EventStream> {
258        let mut conn = self.connection.lock().await;
259        let theater_id = actor_id
260            .parse()
261            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
262        let response = conn
263            .send_and_receive(ManagementCommand::SubscribeToActor { id: theater_id })
264            .await?;
265
266        match response {
267            ManagementResponse::Subscribed {
268                id: _,
269                subscription_id,
270            } => Ok(EventStream {
271                client: self.clone(),
272                actor_id: actor_id.to_string(),
273                subscription_id,
274            }),
275            ManagementResponse::Error { error } => {
276                let error_str = format!("{:?}", error);
277                if error_str.contains("not found") {
278                    Err(CliError::actor_not_found(actor_id))
279                } else {
280                    Err(CliError::ServerError { message: error_str })
281                }
282            }
283            _ => Err(CliError::UnexpectedResponse {
284                response: format!("{:?}", response),
285            }),
286        }
287    }
288
289    /// Get the next response from the connection (for streaming operations)
290    pub async fn next_response(&self) -> CliResult<ManagementResponse> {
291        let mut conn = self.connection.lock().await;
292        conn.receive()
293            .await
294            .map_err(|e| CliError::ConnectionFailed {
295                address: conn.address.clone(),
296                source: e,
297            })
298    }
299
300    /// Get actor status
301    pub async fn get_actor_status(&self, actor_id: &str) -> CliResult<String> {
302        let mut conn = self.connection.lock().await;
303        let theater_id = actor_id
304            .parse()
305            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
306        let response = conn
307            .send_and_receive(ManagementCommand::GetActorStatus { id: theater_id })
308            .await?;
309
310        match response {
311            ManagementResponse::ActorStatus { id: _, status } => Ok(format!("{:?}", status)),
312            ManagementResponse::Error { error } => {
313                let error_str = format!("{:?}", error);
314                if error_str.contains("not found") {
315                    Err(CliError::actor_not_found(actor_id))
316                } else {
317                    Err(CliError::ServerError { message: error_str })
318                }
319            }
320            _ => Err(CliError::UnexpectedResponse {
321                response: format!("{:?}", response),
322            }),
323        }
324    }
325
326    /// Restart an actor
327    pub async fn restart_actor(&self, actor_id: &str) -> CliResult<()> {
328        let mut conn = self.connection.lock().await;
329        let theater_id = actor_id
330            .parse()
331            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
332        let response = conn
333            .send_and_receive(ManagementCommand::RestartActor { id: theater_id })
334            .await?;
335
336        match response {
337            ManagementResponse::Restarted { id: _ } => Ok(()),
338            ManagementResponse::Error { error } => {
339                let error_str = format!("{:?}", error);
340                if error_str.contains("not found") {
341                    Err(CliError::actor_not_found(actor_id))
342                } else {
343                    Err(CliError::ServerError { message: error_str })
344                }
345            }
346            _ => Err(CliError::UnexpectedResponse {
347                response: format!("{:?}", response),
348            }),
349        }
350    }
351
352    /// Update actor component
353    pub async fn update_actor_component(&self, actor_id: &str, component: String) -> CliResult<()> {
354        let mut conn = self.connection.lock().await;
355        let theater_id = actor_id
356            .parse()
357            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
358        let response = conn
359            .send_and_receive(ManagementCommand::UpdateActorComponent {
360                id: theater_id,
361                component,
362            })
363            .await?;
364
365        match response {
366            ManagementResponse::ActorComponentUpdated { id: _ } => Ok(()),
367            ManagementResponse::Error { error } => {
368                let error_str = format!("{:?}", error);
369                if error_str.contains("not found") {
370                    Err(CliError::actor_not_found(actor_id))
371                } else {
372                    Err(CliError::ServerError { message: error_str })
373                }
374            }
375            _ => Err(CliError::UnexpectedResponse {
376                response: format!("{:?}", response),
377            }),
378        }
379    }
380
381    /// Unsubscribe from actor events
382    pub async fn unsubscribe_from_actor(
383        &self,
384        actor_id: &str,
385        subscription_id: Uuid,
386    ) -> CliResult<()> {
387        let mut conn = self.connection.lock().await;
388        let theater_id = actor_id
389            .parse()
390            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
391        let response = conn
392            .send_and_receive(ManagementCommand::UnsubscribeFromActor {
393                id: theater_id,
394                subscription_id,
395            })
396            .await?;
397
398        match response {
399            ManagementResponse::Unsubscribed { id: _ } => Ok(()),
400            ManagementResponse::Error { error } => Err(CliError::ServerError {
401                message: format!("{:?}", error),
402            }),
403            _ => Err(CliError::UnexpectedResponse {
404                response: format!("{:?}", response),
405            }),
406        }
407    }
408
409    /// Open a channel with an actor
410    pub async fn open_channel(
411        &self,
412        actor_id: &str,
413        initial_message: Vec<u8>,
414    ) -> CliResult<String> {
415        let mut conn = self.connection.lock().await;
416        let theater_id = actor_id
417            .parse()
418            .map_err(|_| CliError::invalid_actor_id(actor_id))?;
419        let response = conn
420            .send_and_receive(ManagementCommand::OpenChannel {
421                actor_id: ChannelParticipant::Actor(theater_id),
422                initial_message,
423            })
424            .await?;
425
426        match response {
427            ManagementResponse::ChannelOpened { channel_id, .. } => Ok(channel_id),
428            ManagementResponse::Error { error } => Err(CliError::ServerError {
429                message: format!("{:?}", error),
430            }),
431            _ => Err(CliError::UnexpectedResponse {
432                response: format!("{:?}", response),
433            }),
434        }
435    }
436
437    /// Send a message on a channel
438    pub async fn send_on_channel(&self, channel_id: &str, message: Vec<u8>) -> CliResult<()> {
439        let mut conn = self.connection.lock().await;
440        let response = conn
441            .send_and_receive(ManagementCommand::SendOnChannel {
442                channel_id: channel_id.to_string(),
443                message,
444            })
445            .await?;
446
447        match response {
448            ManagementResponse::ChannelMessage { .. } => Ok(()),
449            ManagementResponse::Error { error } => Err(CliError::ServerError {
450                message: format!("{:?}", error),
451            }),
452            _ => Err(CliError::UnexpectedResponse {
453                response: format!("{:?}", response),
454            }),
455        }
456    }
457
458    /// Close a channel
459    pub async fn close_channel(&self, channel_id: &str) -> CliResult<()> {
460        let mut conn = self.connection.lock().await;
461        let response = conn
462            .send_and_receive(ManagementCommand::CloseChannel {
463                channel_id: channel_id.to_string(),
464            })
465            .await?;
466
467        match response {
468            ManagementResponse::ChannelClosed { .. } => Ok(()),
469            ManagementResponse::Error { error } => Err(CliError::ServerError {
470                message: format!("{:?}", error),
471            }),
472            _ => Err(CliError::UnexpectedResponse {
473                response: format!("{:?}", response),
474            }),
475        }
476    }
477
478    /// Receive channel message (for channel communication)
479    pub async fn receive_channel_message(&self) -> CliResult<Option<(String, Vec<u8>)>> {
480        let mut conn = self.connection.lock().await;
481        match conn.receive().await? {
482            ManagementResponse::ChannelMessage {
483                channel_id,
484                message,
485                ..
486            } => Ok(Some((channel_id, message))),
487            ManagementResponse::ChannelClosed { .. } => Ok(None),
488            ManagementResponse::Error { error } => Err(CliError::ServerError {
489                message: format!("{:?}", error),
490            }),
491            _ => {
492                // Ignore other message types and try again
493                Box::pin(self.receive_channel_message()).await
494            }
495        }
496    }
497}
498
499/// A stream of events from an actor
500pub struct EventStream {
501    client: TheaterClient,
502    actor_id: String,
503    subscription_id: Uuid,
504}
505
506impl EventStream {
507    /// Get the next event from the stream
508    pub async fn next_event(&self) -> CliResult<Option<ChainEvent>> {
509        let mut conn = self.client.connection.lock().await;
510        match conn.receive().await? {
511            ManagementResponse::ActorEvent { event } => Ok(Some(event)),
512            ManagementResponse::ActorStopped { .. } => Ok(None),
513            ManagementResponse::Error { error } => Err(CliError::EventStreamError {
514                reason: format!("{:?}", error),
515            }),
516            _ => {
517                // Ignore other response types in event stream
518                Box::pin(self.next_event()).await
519            }
520        }
521    }
522
523    /// Get the actor ID this stream is associated with
524    pub fn actor_id(&self) -> &str {
525        &self.actor_id
526    }
527
528    /// Get the subscription ID
529    pub fn subscription_id(&self) -> Uuid {
530        self.subscription_id
531    }
532
533    /// Unsubscribe from this event stream
534    pub async fn unsubscribe(self) -> CliResult<()> {
535        self.client
536            .unsubscribe_from_actor(&self.actor_id, self.subscription_id)
537            .await
538    }
539}
540
541#[cfg(test)]
542mod tests {
543    use super::*;
544
545    #[tokio::test]
546    async fn test_client_creation() {
547        let addr = "127.0.0.1:9000".parse().unwrap();
548        let client = TheaterClient::new(addr);
549
550        assert_eq!(client.address().await, addr);
551        assert!(!client.is_connected().await);
552    }
553
554    #[tokio::test]
555    async fn test_client_clone() {
556        let addr = "127.0.0.1:9000".parse().unwrap();
557        let client = TheaterClient::new(addr);
558        let client2 = client.clone();
559
560        assert_eq!(client.address().await, client2.address().await);
561    }
562}
theater_cli/client/theater_client.rs

theater_cli/client/
theater_client.rs
