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

wip

Browse Source
This commit is contained in:
Thilo Behnke
2022-05-21 18:44:56 +02:00
parent 65720f3a57
commit 0283e0da76
5 changed files with 58 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
server/docker-compose.yml
View File

@@ -13,6 +13,7 @@ services:
context: kafka
ports:
- '9092:9092'
- '7243:7243'
environment:
- KAFKA_BROKER_ID=1
- KAFKA_CFG_INTER_BROKER_LISTENER_NAME=DOCKER

5
server/kafka/Dockerfile
View File

@@ -1,4 +1,4 @@
FROM rust:latest as build-stage
FROM rust:1.54 as build-stage
RUN mkdir -p /opt/pong/kafka-script-proxy
WORKDIR /opt/pong/kafka-script-proxy
@@ -10,5 +10,8 @@ FROM bitnami/kafka:latest
ADD custom-entrypoint.sh /
COPY --from=build-stage /opt/pong/kafka-script-proxy/target/release/kafka-script-proxy /bin/
USER root
RUN mkdir -p /var/lib/kafka-script-proxy
ENTRYPOINT []
CMD [ "/custom-entrypoint.sh" ]

2
server/kafka/kafka-script-proxy/.dockerignore Normal file
View File

@@ -0,0 +1,2 @@
/.idea
/target

3
server/kafka/kafka-script-proxy/Cargo.toml
View File

@@ -1,7 +1,8 @@
[package]
name = "kafka-script-proxy"
version = "0.1.0"
edition = "2021"
edition = "2018"
[workspace]

60
server/kafka/kafka-script-proxy/src/main.rs
View File

@@ -1,7 +1,11 @@
use std::convert::Infallible;
use hyper::{Body, Method, Request, Response, Server};
use hyper::server::conn::AddrStream;
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Method, Request, Response, Server};
use std::convert::Infallible;
use std::fs::write;
use std::process::Output;
use tokio::fs::OpenOptions;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
const TOPICS: [&str; 3] = ["move", "status", "input"];
@@ -18,34 +22,39 @@ pub async fn run() -> Result<(), ()> {
}))
});
let host = ([0,0,0,0], 7243).into();
let host = ([0, 0, 0, 0], 7243).into();
let server = Server::bind(&host).serve(make_svc);
println!("Listening on http://{}", host);
write_to_log(&format!("Listening on http://{}", host)).await;
let graceful = server.with_graceful_shutdown(shutdown_signal());
graceful.await;
Ok(())
}
async fn handle_request(req: Request<Body>) -> Result<Response<Body>, Infallible> {
println!("req to {} with method {}", req.uri().path(), req.method());
write_to_log(&format!(
"req to {} with method {}",
req.uri().path(),
req.method()
))
.await;
match (req.method(), req.uri().path()) {
(&Method::POST, "/add_partition") => handle_add_partition().await,
_ => Ok(Response::new("unknown".into()))
_ => Ok(Response::new("unknown".into())),
}
}
async fn handle_add_partition() -> Result<Response<Body>, Infallible> {
async fn handle_add_partition() -> Result<Response<Body>, Infallible> {
let current_count = get_highest_partition_count().await;
let next_partition = current_count + 1;
for topic in TOPICS {
let output = Command::new("/bin/bash").arg(format!("-c /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic {} --partitions {}", topic, next_partition)).output().await;
let output = run_command(&format!("/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic {} --partitions {}", topic, next_partition)).await;
if let Err(e) = output {
println!("{}", e);
write_to_log(&format!("{}", e)).await;
return Ok(Response::new(Body::empty()));
}
let output = output.unwrap();
if !output.status.success() {
println!("{:?}", std::str::from_utf8(&*output.stderr));
write_to_log(&format!("{:?}", std::str::from_utf8(&*output.stderr))).await;
return Ok(Response::new(Body::empty()));
}
}
@@ -53,14 +62,43 @@ async fn handle_add_partition() -> Result<Response<Body>, Infallible> {
}
async fn get_highest_partition_count() -> u32 {
let output = Command::new("/bin/bash").arg("-c /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe | grep -Po '(?<=Partition: )(\\d+)' | sort -r | head -1").output().await.unwrap();
let output = run_command("/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe | grep -Po '(?<=PartitionCount: )(\\d+)' | sort -r | head -1").await.unwrap();
let stdout = std::str::from_utf8(&*output.stdout).unwrap().to_owned();
stdout.parse::<u32>().unwrap()
}
async fn run_command(command: &str) -> std::io::Result<Output> {
write_to_log(&format!("Running command: {}", command)).await;
let output = Command::new("/bin/bash")
.arg(format!("-c {}", command))
.output()
.await.unwrap();
let stdout = std::str::from_utf8(&*output.stdout).unwrap();
let stderr = std::str::from_utf8(&*output.stderr).unwrap();
write_to_log(&format!("Command returned stdout: {}", stdout));
write_to_log(&format!("Command returned stdout: {}", stderr));
Ok(output)
}
pub async fn shutdown_signal() {
// Wait for the CTRL+C signal
tokio::signal::ctrl_c()
.await
.expect("failed to install CTRL+C signal handler");
}
async fn write_to_log(value: &str) -> std::io::Result<()> {
let content = format!("{}\n", value);
let mut options = OpenOptions::new();
let mut file = options
.create(true)
.write(true)
.append(true)
.open("/var/lib/kafka-script-proxy/log.txt")
.await;
if let Err(e) = file {
println!("Failed to open file: {}", e);
return Ok(());
}
file.unwrap().write_all(content.as_bytes()).await
}