nebulous 0.1.86

A globally distributed container orchestrator
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

use nebulous::config::GlobalConfig;
use nebulous::models::V1StreamData;
use serde_json::Value;
use std::error::Error;
use std::fs;
use std::io::{self, Read};
use tracing::debug;

pub async fn send_messages(args: &crate::cli::SendMessageCommands) -> Result<(), Box<dyn Error>> {
    let config = GlobalConfig::read()?;
    debug!("Config: {:?}", config);
    let current_server = config.get_current_server_config().unwrap(); // Handle error more gracefully
    let server = current_server.server.as_ref().unwrap(); // Handle error
    let api_key = current_server.api_key.as_ref().unwrap(); // Handle error

    // Get required namespace
    let namespace = args
        .namespace
        .as_ref()
        .ok_or("Namespace must be provided with --namespace")?;
    let stream_name = &args.name;

    let bearer_token = format!("Bearer {}", api_key);

    // Construct stream message URL
    let url = format!(
        "{}/v1/processors/{}/{}/messages",
        server, namespace, stream_name
    );

    // Read message content (from file or stdin)
    let content_str = if let Some(file_path) = &args.file {
        debug!("Reading message content from file: {}", file_path);
        fs::read_to_string(file_path)?
    } else {
        debug!("Reading message content from stdin");
        let mut stdin_content = String::new();
        io::stdin().read_to_string(&mut stdin_content)?;
        stdin_content
    };

    // Deserialize message content into a generic JSON Value
    let message_content: Value = serde_yaml::from_str(&content_str).map_err(|e| {
        format!(
            "Failed to parse YAML/JSON input into message content: {}",
            e
        )
    })?;

    // Create the payload using V1StreamData
    let payload = V1StreamData {
        content: message_content,
        wait: if args.wait { Some(true) } else { None },
        user_key: None,
    };
    debug!("Payload: {:?}", payload);

    // --- API Call ---
    let client = reqwest::Client::new();
    debug!(
        "Sending message to URL: {} with payload: {:?}",
        url, payload
    );
    let response = client
        .post(&url)
        .header("Authorization", &bearer_token)
        .header("Content-Type", "application/json")
        .json(&payload)
        .send()
        .await?;

    // --- Response Handling ---
    if response.status().is_success() {
        println!(
            "Message sent successfully to stream '{}' in namespace '{}'.",
            stream_name, namespace
        );
        let response_body: Value = response.json().await?;
        println!("{:?}", response_body);
    } else {
        let status = response.status();
        let error_body = response.text().await?;
        eprintln!("Failed to send message: {} - {}", status, error_body);
        return Err(format!("API request failed with status {}", status).into());
    }

    Ok(())
}