Merge pull request #36 from harlanc/dev

Dev

Cargo.toml

@@ -5,6 +5,8 @@ members = [
   "protocol/hls",
   "library/bytesio",
   "application/xiu",
+  "application/http-server",
+  "application/pprtmp",
   "library/container/flv",
   "library/container/mpegts",
   "library/codec/h264",

README.md

@@ -38,8 +38,8 @@ stand-alone server or cluster(RTMP relay).
 - [x] Support configuring the service using command line or a configuration file.
 - [x] Support HTTP API/Notifications.
   - [x] Support querying stream information.
-  - [ ] Support notify stream status.
-- [ ] Support token authentications.
+  - [x] Support notify stream status.
+- [x] Support token authentications.
 - [ ] Support RTSP.
 
 ## Preparation

application/http-server/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "http-server"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+serde_derive = "1.0"
+serde = { version = "1.0", features = ["derive", "rc"] }
+serde_json = { version = "1", default-features = false, features = [
+    "alloc",
+    "raw_value",
+    "std",
+] }
+axum = "0.6.10"
+log = "0.4.0"
+env_logger = "0.9.3"
+
+[dependencies.tokio]
+version = "1.4.0"
+default-features = false
+#features = ["rt-core", "rt-threaded", "macros", "time","sync"]
+features = ["full"]

application/http-server/src/main.rs

@@ -0,0 +1,47 @@
+use axum::{
+    routing::{get, post},
+    Router,
+};
+
+use std::net::SocketAddr;
+use std::env;
+
+#[tokio::main]
+async fn main() {
+    env::set_var("RUST_LOG", "info");
+    env_logger::init();
+    let app = Router::new()
+        .route("/", get(root))
+        .route("/on_publish", post(on_publish))
+        .route("/on_unpublish", post(on_unpublish))
+        .route("/on_play", post(on_play))
+        .route("/on_stop", post(on_stop));
+
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3001));
+    log::info!("http server listen on: {}", 3001);
+    axum::Server::bind(&addr)
+        .serve(app.into_make_service())
+        .await
+        .unwrap();
+}
+
+// basic handler that responds with a static string
+async fn root() -> &'static str {
+    "Hello, World!"
+}
+
+async fn on_publish(body: String) {
+    log::info!("on_publish body: {}", body);
+}
+
+async fn on_unpublish(body: String) {
+    log::info!("on_unpublish body: {}", body);
+}
+
+async fn on_play(body: String) {
+    log::info!("on_play body: {}", body);
+}
+
+async fn on_stop(body: String) {
+    log::info!("on_stop body: {}", body);
+}

application/pprtmp/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "pprtmp"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = "^1.0"
+rtmp = { path = "../../protocol/rtmp/" }
+log = "0.4.0"
+env_logger = "0.9.3"
+clap = "4.1.4"
+
+[dependencies.tokio]
+version = "1.26.0"
+default-features = false
+#features = ["rt-core", "rt-threaded", "macros", "time","sync"]
+features = ["full"]

application/pprtmp/src/main.rs

@@ -0,0 +1,105 @@
+use {
+    anyhow::Result,
+    clap::{value_parser, Arg, Command},
+    rtmp::channels::ChannelsManager,
+    rtmp::session::client_session::ClientSession,
+    rtmp::session::client_session::ClientType,
+    rtmp::utils::RtmpUrlParser,
+    std::env,
+    std::process::exit,
+    tokio::net::TcpStream,
+    tokio::signal,
+    tokio::time::Duration,
+};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env::set_var("RUST_LOG", "info");
+    env_logger::init();
+
+    let mut cmd = Command::new("pprtmp")
+        .bin_name("pprtmp")
+        .version("0.1.0")
+        .author("HarlanC <harlanc@foxmail.com>")
+        .about("pull and push rtmp!!!")
+        .arg(
+            Arg::new("pullrtmp")
+                .long("pull_rtmp_url")
+                .short('i')
+                .value_name("path")
+                .help("Specify the pull rtmp url.")
+                .value_parser(value_parser!(String))
+                .required(true),
+        )
+        .arg(
+            Arg::new("pushrtmp")
+                .long("push_rtmp_url")
+                .short('o')
+                .value_name("path")
+                .help("Specify the push rtmp url.")
+                .value_parser(value_parser!(String))
+                .required(true),
+        );
+
+    let args: Vec<String> = env::args().collect();
+    if 1 == args.len() {
+        cmd.print_help()?;
+        return Ok(());
+    }
+    let matches = cmd.clone().get_matches();
+    let pull_rtmp_url = matches.get_one::<String>("pullrtmp").unwrap().clone();
+    let push_rtmp_url = matches.get_one::<String>("pushrtmp").unwrap().clone();
+
+    let mut channel = ChannelsManager::new(None);
+    let producer = channel.get_channel_event_producer();
+    tokio::spawn(async move { channel.run().await });
+
+    let mut pull_parser = RtmpUrlParser::new(pull_rtmp_url);
+    if let Err(err) = pull_parser.parse_url() {
+        log::error!("err: {}", err);
+    }
+    pull_parser.append_port(String::from("1935"));
+    let stream1 = TcpStream::connect(pull_parser.raw_domain_name.clone()).await?;
+    let mut pull_client_session = ClientSession::new(
+        stream1,
+        ClientType::Play,
+        pull_parser.raw_domain_name,
+        pull_parser.app_name.clone(),
+        pull_parser.raw_stream_name,
+        producer.clone(),
+    );
+    tokio::spawn(async move {
+        if let Err(err) = pull_client_session.run().await {
+            log::error!("pull_client_session as pull client run error: {}", err);
+        }
+    });
+
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    let mut push_parser = RtmpUrlParser::new(push_rtmp_url);
+    if let Err(err) = push_parser.parse_url() {
+        log::error!("err: {}", err);
+    }
+    push_parser.append_port(String::from("1935"));
+    // push the rtmp stream from local to remote rtmp server
+    let stream2 = TcpStream::connect(push_parser.raw_domain_name.clone()).await?;
+    let mut push_client_session = ClientSession::new(
+        stream2,
+        ClientType::Publish,
+        push_parser.raw_domain_name,
+        push_parser.app_name,
+        push_parser.raw_stream_name,
+        producer.clone(),
+    );
+
+    push_client_session.subscribe(pull_parser.app_name, pull_parser.stream_name);
+    tokio::spawn(async move {
+        if let Err(err) = push_client_session.run().await {
+            log::error!("push_client_session as push client run error: {}", err);
+            exit(0);
+        }
+    });
+
+    signal::ctrl_c().await?;
+    Ok(())
+}

application/xiu/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "xiu"
 description = "A powerful live server by Rust ."
-version = "0.5.0"
+version = "0.6.0"
 authors = ["HarlanC <wawacry@qq.com"]
 repository = "https://github.com/harlanc/xiu"
 license = "MIT"
@@ -24,12 +24,14 @@ serde_json = { version = "1", default-features = false, features = [
     "std",
 ] }
 axum = "0.6.10"
+tokio-metrics = { version = "0.2.0", default-features = false }
+uuid = { version = "0.6.5", features = ["v4"] }
 
-rtmp = "0.2.0"   
-httpflv = "0.1.2"
-hls = "0.1.2"    
+rtmp = "0.3.0"   
+httpflv = "0.2.0"
+hls = "0.2.0"    
 env_logger_extend = "0.1.1"
-# rtmp = { path = "../../protocol/rtmp/" }             
+# rtmp = { path = "../../protocol/rtmp/" }
 # httpflv = { path = "../../protocol/httpflv/" }
 # hls = { path = "../../protocol/hls/" }
 # env_logger_extend = { path = "../../library/logger/" }
@@ -40,7 +42,7 @@ default = ["std"]
 std = ["serde"]
 
 [dependencies.tokio]
-version = "1.4.0"
+version = "1.26.0"
 default-features = false
 #features = ["rt-core", "rt-threaded", "macros", "time","sync"]
 features = ["full"]