diff --git a/kafka/kafka-script-proxy/src/main.rs b/kafka/kafka-script-proxy/src/main.rs index 6a39cbd..9f899dd 100644 --- a/kafka/kafka-script-proxy/src/main.rs +++ b/kafka/kafka-script-proxy/src/main.rs @@ -1,7 +1,8 @@ extern crate core; +use std::collections::HashMap; use hyper::service::{make_service_fn, service_fn}; -use hyper::{Body, Method, Request, Response, Server, StatusCode}; +use hyper::{Body, Method, Request, Response, Server, StatusCode, Uri}; use std::convert::Infallible; use std::process::Output; use tokio::fs::OpenOptions; @@ -38,11 +39,61 @@ async fn handle_request(req: Request) -> Result, Infallible )) .await; match (req.method(), req.uri().path()) { + (&Method::POST, "/create_topic") => handle_create_topic(&req.uri()).await, (&Method::POST, "/add_partition") => handle_add_partition().await, _ => build_error_res("not found", StatusCode::NOT_FOUND), } } +async fn handle_create_topic(uri: &Uri) -> Result, Infallible> { + let query_params = get_query_params(uri); + let topic = query_params.get("topic"); + if let None = topic { + return build_error_res("Missing mandatory query param >topic<", StatusCode::BAD_REQUEST); + } + let topic = topic.unwrap(); + + write_to_log(&format!("Called to create topic {}.", topic)).await; + if verify_topic_exists(topic) { + return build_success_res("topic already exists"); + } + + run_command(&format!("/opt/bitnami/kafka/bin/kafka-topics.sh --create --topic {} --localhost:9093", topic)).await + .map(|| build_success_res(&format!("successfully created topic {}", topic))) + .map_err(|e| { + write_to_log(&format!("Failed to create topic {}: {:?}", topic, e)).await; + build_error_res(&format!("failed to create topic {}", topic), StatusCode::INTERNAL_SERVER_ERROR) + }) + .unwrap() +} + +pub fn get_query_params(uri: &Uri) -> HashMap<&str, &str> { + let query = uri.query(); + println!("uri={:?}, query={:?}", uri, query); + match query { + None => HashMap::new(), + Some(query) => query + .split("&") + .map(|s| s.split_at(s.find("=").unwrap())) + .map(|(key, value)| (key, &value[1..])) + .collect(), + } +} + + +async fn verify_topic_exists(topic: &str) -> bool { + return match run_command(&format!("/opt/bitnami/kafka/bin/kafka-topics.sh --describe --topic {} --localhost:9093", topic)).await { + Ok(_) => { + write_to_log(&format!("topic {} exists", topic)); + true + }, + Err(e) => { + write_to_log(&format!("topic {} does not exist or caused other issues: {:?}", topic, e)); + false + } + } +} + async fn handle_add_partition() -> Result, Infallible> { write_to_log("Called to add partition.").await; let current_count = get_highest_partition_count().await; diff --git a/server/src/main.rs b/server/src/main.rs index bc80204..75b2e1c 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -1,5 +1,6 @@ extern crate core; +use hyper::{Body, Client, Method, Request}; use log::{debug, error, info, Level, LevelFilter}; use log4rs::append::console::{ConsoleAppender, ConsoleAppenderConfig}; use log4rs::append::file::FileAppender; @@ -72,6 +73,22 @@ pub async fn main() { }; info!("KAFKA_TOPIC_MANAGER_HOST={}", kafka_topic_manager_host); + info!("BOOTSTRAP: Create topics if needed"); + let topics = ["session", "host_tick", "peer_tick", "heart_beat"]; + let http_client = Client::new(); + for topic in topics { + let req = Request::builder().method(Method::POST).uri(format!("{}/create_topic?topic={}", kafka_topic_manager_host, topic)).body(Body::empty()).expect("request builder for topic creation"); + let res = http_client.request(req).await; + match res { + Ok(_) => { + info!("Successfully created topic {}", topic); + }, + Err(e) => { + error!("Failed to create topic {}: {:?}", topic, e); + } + } + } + info!("booting up server"); HttpServer::new([0, 0, 0, 0], 4000, &kafka_host, &kafka_topic_manager_host) .run()