github/wasm-pong
1
0
Fork 0
mirror of https://github.com/thilo-behnke/wasm-pong.git synced 2026-07-13 11:39:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

create topics from within pong server

Browse Source
This commit is contained in:
Thilo Behnke
2022-08-09 14:49:03 +02:00
parent 1e99616192
commit cee77e3bdd
2 changed files with 69 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
kafka/kafka-script-proxy/src/main.rs
View File

@@ -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<Body>) -> Result<Response<Body>, 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<Response<Body>, 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<Response<Body>, Infallible> {
write_to_log("Called to add partition.").await;
let current_count = get_highest_partition_count().await;

17
server/src/main.rs
View File

@@ -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()