use actix_web::{web, App, HttpServer, HttpResponse, Responder};
use tokio::sync::Mutex;
use std::sync::Arc;
use redis::AsyncCommands;

struct AppState {
    concurrent_connections: Arc<Mutex<u32>>,
    redis_client: redis::Client,
}

async fn stream_handler(state: web::Data<AppState>) -> impl Responder {
    let mut conn_count = state.concurrent_connections.lock().await;
    *conn_count += 1;

    let mut con = state.redis_client.get_async_connection().await.expect("Failed to connect to Redis");
    let stream_url: Option<String> = con.get("active_stream").await.ok();

    match stream_url {
        Some(url) => HttpResponse::Ok().body(format!("Streaming from: {}\nActive Connections: {}", url, conn_count)),
        None => HttpResponse::NotFound().body("No active stream available"),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let redis_client = redis::Client::open("redis://127.0.0.1/").expect("Failed to connect to Redis");

    let state = web::Data::new(AppState {
        concurrent_connections: Arc::new(Mutex::new(0)),
        redis_client,
    });

    HttpServer::new(move || {
        App::new()
            .app_data(state.clone())
            .route("/stream", web::get().to(stream_handler))
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}

use webrtc::api::APIBuilder;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::RTCPeerConnection;
use std::sync::Arc;
use tokio::sync::Mutex;

struct WebRTCServer {
    peer_connection: Arc<Mutex<RTCPeerConnection>>,
}

impl WebRTCServer {
    async fn new() -> Self {
        let api = APIBuilder::new().build();
        let config = RTCConfiguration::default();
        let peer_connection = api.new_peer_connection(config).await.expect("Failed to create peer connection");

        WebRTCServer {
            peer_connection: Arc::new(Mutex::new(peer_connection)),
        }
    }

    async fn create_offer(&self) -> String {
        let offer = self.peer_connection.lock().await.create_offer(None).await.expect("Failed to create offer");
        self.peer_connection.lock().await.set_local_description(offer.clone()).await.expect("Failed to set local description");

        offer.sdp
    }
}

#[tokio::main]
async fn main() {
    let webrtc_server = WebRTCServer::new().await;
    let offer_sdp = webrtc_server.create_offer().await;
    println!("WebRTC Offer SDP:\n{}", offer_sdp);
}

use redis::{AsyncCommands, Client};
use tokio::runtime::Runtime;

async fn cache_stream_url(stream_id: &str, url: &str) -> redis::RedisResult<()> {
    let client = Client::open("redis://127.0.0.1/")?;
    let mut con = client.get_async_connection().await?;
    con.set_ex(stream_id, url, 3600).await // 快取一小時
}

async fn get_stream_url(stream_id: &str) -> redis::RedisResult<Option<String>> {
    let client = Client::open("redis://127.0.0.1/")?;
    let mut con = client.get_async_connection().await?;
    con.get(stream_id).await
}

fn main() {
    let rt = Runtime::new().unwrap();

    rt.block_on(async {
        cache_stream_url("stream123", "http://cdn.example.com/stream123.m3u8").await.unwrap();
        match get_stream_url("stream123").await {
            Ok(Some(url)) => println!("從快取得URL: {}", url),
            Ok(None) => println!("串流URL未在快取中找到。"),
            Err(e) => eprintln!("取得串流URL時發生錯誤: {:?}", e),
        }
    });
}

use actix_web::{web, App, HttpServer, HttpResponse, Responder};
use redis::aio::Connection;
use redis::AsyncCommands;
use std::sync::Arc;
use tokio::sync::Mutex;

struct AppState {
    cache: Arc<Mutex<Connection>>,
}

async fn stream_handler(state: web::Data<AppState>) -> impl Responder {
    let mut con = state.cache.lock().await;
    let stream_url: Option<String> = con.get("active_stream").await.ok();
    
    match stream_url {
        Some(url) => HttpResponse::Ok().body(format!("正在串流: {}", url)),
        None => HttpResponse::NotFound().body("沒有可用的串流"),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let client = redis::Client::open("redis://127.0.0.1/").expect("無法連線到Redis");
    let connection = client.get_async_connection().await.expect("Redis連線錯誤");

    let state = web::Data::new(AppState {
        cache: Arc::new(Mutex::new(connection)),
    });

    HttpServer::new(move || {
        App::new()
            .app_data(state.clone())
            .route("/stream", web::get().to(stream_handler))
    })
    .bind("127.0.0.1:8080")?
    .run()
    .await
}

use actix_web::{web, App, HttpServer, HttpResponse, Responder};
use redis::{aio::ConnectionManager, AsyncCommands, Client};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::RwLock;

#[derive(Debug, Serialize, Deserialize, Clone)]
struct StreamNode {
    id: String,
    region: String,
    load: f32,
    capacity: u32,
    url: String,
}

struct AppState {
    redis: Arc<RwLock<ConnectionManager>>,
    node_id: String,
}

async fn register_node(state: &AppState, node: &StreamNode) -> Result<(), redis::RedisError> {
    let mut con = state.redis.write().await;
    let node_json = serde_json::to_string(node).unwrap();
    con.set_ex(format!("node:{}", node.id), node_json, 60).await?;
    con.zadd("available_nodes", node.id.clone(), node.load as f64).await?;
    Ok(())
}

async fn get_optimal_node(con: &mut ConnectionManager) -> Option<StreamNode> {
    let node_id: String = con.zrange_first("available_nodes").await.ok()?;
    let node_json: String = con.get(format!("node:{}", node_id)).await.ok()?;
    serde_json::from_str(&node_json).ok()
}

async fn stream_request(state: web::Data<AppState>) -> impl Responder {
    let mut con = state.redis.write().await;
    
    match get_optimal_node(&mut con).await {
        Some(node) => HttpResponse::Ok().json(node),
        None => HttpResponse::ServiceUnavailable().body("無可用串流節點")
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_string());
    let node_id = std::env::var("NODE_ID").unwrap_or_else(|_| "node1".to_string());
    let region = std::env::var("REGION").unwrap_or_else(|_| "asia-east1".to_string());
    
    let client = Client::open(redis_url).expect("無法連線Redis");
    let manager = ConnectionManager::new(client).await.expect("Redis連線錯誤");
    
    let state = web::Data::new(AppState {
        redis: Arc::new(RwLock::new(manager)),
        node_id: node_id.clone(),
    });
    
    // 註冊此節點
    let node = StreamNode {
        id: node_id,
        region,
        load: 0.0,
        capacity: 5000,
        url: format!("rtmp://streaming.example.com/{}", node_id),
    };
    
    let state_clone = state.clone();
    tokio::spawn(async move {
        let _ = register_node(&state_clone, &node).await;
        
        // 定期更新節點狀態
        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
        loop {
            interval.tick().await;
            let _ = register_node(&state_clone, &node).await;
        }
    });
    
    HttpServer::new(move || {
        App::new()
            .app_data(state.clone())
            .route("/stream", web::get().to(stream_request))
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}

from prometheus_client import start_http_server, Summary, Gauge
import requests
import time

REQUEST_TIME = Summary("request_processing_seconds", "Time spent processing requests")
STREAM_STATUS = Gauge("stream_available", "Indicates if a stream is available (1) or not (0)")

@REQUEST_TIME.time()
def monitor_stream():
    try:
        response = requests.get("http://127.0.0.1:8080/stream", timeout=3)
        if response.status_code == 200:
            STREAM_STATUS.set(1)
        else:
            STREAM_STATUS.set(0)
        print(f"Streaming Status: {response.text}")
    except Exception as e:
        STREAM_STATUS.set(0)
        print(f"Monitoring Error: {e}")

if __name__ == "__main__":
    start_http_server(8000)
    while True:
        monitor_stream()
        time.sleep(5)

# Build Stage
FROM rust:1.71 AS builder
WORKDIR /app

# Install dependencies
RUN apt-get update && apt-get install -y pkg-config libssl-dev

# Copy project files
COPY Cargo.toml Cargo.lock ./
RUN mkdir src && echo "fn main() {}" > src/main.rs  # Precompile dependencies
RUN cargo fetch && cargo build --release

# Copy actual source files and build the application
COPY src ./src
RUN cargo build --release --locked

# Production Stage
FROM debian:bullseye-slim
WORKDIR /app

# Install runtime dependencies
RUN apt-get update && apt-get install -y libssl-dev redis-tools && apt-get clean

# Copy compiled Rust binary from builder
COPY --from=builder /app/target/release/streaming-server /app/

# Expose the server port
EXPOSE 8080

# Start the streaming server
CMD ["/app/streaming-server"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rust-streaming-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rust-streaming
  template:
    metadata:
      labels:
        app: rust-streaming
    spec:
      containers:
      - name: rust-streaming
        image: rust-streaming-engine:latest
        ports:
        - containerPort: 8080
        resources:
          limits:
            cpu: "500m"
            memory: "512Mi"
          requests:
            cpu: "250m"
            memory: "256Mi"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10

kubectl apply -f rust-streaming-deployment.yaml

kubectl scale deployment rust-streaming-server --replicas=5

kubectl get pods

upstream streaming_servers {
    server 192.168.1.2:8080 weight=3;  # 優先分配流量給效能較佳的節點
    server 192.168.1.3:8080 max_fails=2 fail_timeout=10s;
    server 192.168.1.4:8080;
    ip_hash;  # 確保相同使用者的請求始終被導向同一伺服器
}

server {
    listen 80;
    location / {
        proxy_pass http://streaming_servers;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
    }
}

use actix_web::{get, App, HttpResponse, HttpServer, Responder};
use redis::AsyncCommands;
use std::env;

#[get("/stream/{stream_id}")]
async fn get_stream(stream_id: actix_web::web::Path<String>) -> impl Responder {
    let redis_url = env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1/".to_string());
    let client = redis::Client::open(redis_url).expect("Failed to connect to Redis");
    let mut con = client.get_async_connection().await.expect("Redis connection failed");

    match con.get::<_, Option<String>>(&stream_id).await {
        Ok(Some(url)) => HttpResponse::Ok().body(url),
        _ => HttpResponse::NotFound().body("Stream not found"),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| App::new().service(get_stream))
        .bind("0.0.0.0:8080")?
        .run()
        .await
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rust-streaming-server
spec:
  replicas: 5  # 自動擴充套件的初始節點數
  selector:
    matchLabels:
      app: rust-streaming
  template:
    metadata:
      labels:
        app: rust-streaming
    spec:
      containers:
      - name: rust-streaming
        image: rust-streaming-engine:latest
        ports:
        - containerPort: 8080
        resources:
          limits:
            cpu: "500m"
            memory: "512Mi"
          requests:
            cpu: "250m"
            memory: "256Mi"
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10

use wasm_bindgen::prelude::*;
use web_sys::{console, HtmlVideoElement};

#[wasm_bindgen]
pub fn start_streaming(video_id: &str, stream_url: &str) {
    let window = web_sys::window().expect("No global `window` exists");
    let document = window.document().expect("No document found");
    let video_element = document
        .get_element_by_id(video_id)
        .expect("Video element not found")
        .dyn_into::<HtmlVideoElement>()
        .expect("Failed to cast to video element");

    video_element.set_src(stream_url);
    video_element.set_autoplay(true);
    video_element.set_controls(true);
    console::log_1(&"Streaming started on WebAssembly!".into());
}

use warp::Filter;
use tokio::fs;

#[tokio::main]
async fn main() {
    let wasm_route = warp::path("wasm")
        .and(warp::fs::dir("./wasm_output"));

    let routes = wasm_route.with(warp::cors().allow_any_origin());
    warp::serve(routes).run(([0, 0, 0, 0], 8080)).await;
}

import requests
import time

def get_network_speed():
    try:
        return requests.get("http://speedtest.example.com").json()["speed"]
    except:
        return 10  # 預設為中低速度值

def select_stream_quality(speed):
    if speed > 50:
        return "http://cdn.example.com/high_quality.m3u8"
    elif speed > 10:
        return "http://cdn.example.com/medium_quality.m3u8"
    else:
        return "http://cdn.example.com/low_quality.m3u8"

while True:
    network_speed = get_network_speed()
    selected_stream = select_stream_quality(network_speed)
    print(f"Selected Stream: {selected_stream}")
    time.sleep(30)  # 每30秒重新檢測一次

# 第一階段：構建Rust串流伺服器
FROM rust:1.71 AS builder
WORKDIR /app

# 安裝必要依賴
RUN apt-get update && apt-get install -y \
    ffmpeg \
    libavcodec-dev \
    libavformat-dev \
    libavutil-dev \
    pkg-config \
    libssl-dev

# 複製Rust專案檔案
COPY Cargo.toml Cargo.lock ./
RUN mkdir src && echo "fn main() {}" > src/main.rs  # 建立虛擬main.rs以解析依賴
RUN cargo fetch && cargo build --release

# 複製實際原始碼並編譯應用程式
COPY src ./src
RUN cargo build --release --locked

# 第二階段：在最小化映像中佈署Rust串流伺服器
FROM debian:bullseye-slim
WORKDIR /app

# 安裝僅執行時所需的依賴
RUN apt-get update && apt-get install -y \
    ffmpeg \
    libavcodec-dev \
    libavformat-dev \
    libavutil-dev \
    redis-tools && \
    apt-get clean

# 從builder階段複製編譯好的Rust二進位檔
COPY --from=builder /app/target/release/rust-streaming-server /app/

# 設定許可權並開放串流伺服器埠
RUN chmod +x /app/rust-streaming-server
EXPOSE 8080

# 啟動Rust串流伺服器
CMD ["/app/rust-streaming-server"]

docker build -t rust-streaming-engine .
docker run -d -p 8080:8080 --name streaming-server rust-streaming-engine

use redis::{AsyncCommands, Client};
use tokio::time::{self, Duration};
use std::env;

async fn cache_stream_url(stream_id: &str, url: &str) -> redis::RedisResult<()> {
    let client = Client::open(env::var("REDIS_URL").unwrap_or("redis://127.0.0.1/".to_string()))?;
    let mut con = client.get_async_connection().await?;
    con.set_ex(stream_id, url, 3600).await // 快取一小時
}

async fn get_stream_url(stream_id: &str) -> redis::RedisResult<Option<String>> {
    let client = Client::open(env::var("REDIS_URL").unwrap_or("redis://127.0.0.1/".to_string()))?;
    let mut con = client.get_async_connection().await?;
    con.get(stream_id).await
}

#[tokio::main]
async fn main() {
    let stream_id = "stream123";
    let stream_url = "http://cdn.example.com/stream123.m3u8";

    if let Err(e) = cache_stream_url(stream_id, stream_url).await {
        eprintln!("快取串流 URL 失敗: {:?}", e);
    } else {
        println!("串流 URL 已快取: {}", stream_url);
    }

    match get_stream_url(stream_id).await {
        Ok(Some(url)) => println!("從快取得 URL: {}", url),
        Ok(None) => println!("快取中找不到串流 URL"),
        Err(e) => eprintln!("取得串流 URL 時發生錯誤: {:?}", e),
    }
}

version: "3.8"
services:
  streaming-server:
    build: .
    ports:
      - "8080:8080"
    depends_on:
      - redis
    environment:
      REDIS_URL: "redis://redis:6379"

  webrtc-server:
    image: webrtc-rs
    ports:
      - "9000:9000"

  redis:
    image: redis:latest
    restart: always
    ports:
      - "6379:6379"

docker-compose up --build -d

use webrtc::api::APIBuilder;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use std::sync::Arc;
use tokio::sync::Mutex;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let api = APIBuilder::new().build();
    let config = RTCConfiguration::default();
    let peer_connection = Arc::new(Mutex::new(api.new_peer_connection(config).await?));

    let peer_clone = peer_connection.clone();
    tokio::spawn(async move {
        let mut pc = peer_clone.lock().await;
        pc.on_peer_connection_state_change(Box::new(|state: RTCPeerConnectionState| {
            println!("點對點連線狀態變更: {:?}", state);
            Box::pin(async {})
        }));
    });

    let offer = peer_connection.lock().await.create_offer(None).await?;
    peer_connection.lock().await.set_local_description(offer.clone()).await?;

    println!("WebRTC Offer SDP:\n{}", offer.sdp);

    Ok(())
}

use redis::{Client, Commands, Connection, RedisError, RedisResult};
use std::sync::{Arc, Mutex};
use std::time::Duration;

struct RedisCacheManager {
    pool: Arc<Mutex<Vec<Connection>>>,
    url: String,
}

impl RedisCacheManager {
    fn new(redis_url: &str, pool_size: usize) -> Result<Self, RedisError> {
        let client = Client::open(redis_url)?;
        let mut connections = Vec::with_capacity(pool_size);
        
        for _ in 0..pool_size {
            connections.push(client.get_connection()?);
        }
        
        Ok(Self {
            pool: Arc::new(Mutex::new(connections)),
            url: redis_url.to_string(),
        })
    }
    
    fn get_connection(&self) -> Result<Connection, RedisError> {
        let mut pool = self.pool.lock().unwrap();
        
        if let Some(conn) = pool.pop() {
            Ok(conn)
        } else {
            Client::open(&self.url)?.get_connection()
        }
    }
    
    fn return_connection(&self, conn: Connection) {
        let mut pool = self.pool.lock().unwrap();
        pool.push(conn);
    }
    
    fn cache_stream_url(&self, stream_id: &str, url: &str, ttl_seconds: u64) -> RedisResult<()> {
        let mut conn = self.get_connection()?;
        let result = conn.set_ex(stream_id, url, ttl_seconds as usize);
        self.return_connection(conn);
        result
    }
    
    fn get_stream_url(&self, stream_id: &str) -> RedisResult<Option<String>> {
        let mut conn = self.get_connection()?;
        let result = conn.get(stream_id);
        self.return_connection(conn);
        result
    }
}

fn main() {
    // 建立連線池，提高效能
    let cache_manager = match RedisCacheManager::new("redis://127.0.0.1/", 10) {
        Ok(manager) => manager,
        Err(e) => {
            eprintln!("無法建立Redis連線池: {:?}", e);
            return;
        }
    };
    
    // 快取串流URL，設定一小時過期
    let stream_id = "live_stream_123";
    let stream_url = "https://cdn.example.com/streams/live_123.m3u8";
    
    if let Err(e) = cache_manager.cache_stream_url(stream_id, stream_url, 3600) {
        eprintln!("快取串流URL失敗: {:?}", e);
    } else {
        println!("成功快取串流URL: {}", stream_url);
    }
    
    // 從快取中讀取串流URL
    match cache_manager.get_stream_url(stream_id) {
        Ok(Some(url)) => println!("從快取得URL: {}", url),
        Ok(None) => println!("快取中找不到此串流URL"),
        Err(e) => println!("查詢快取時發生錯誤: {:?}", e),
    }
}

version: "3.8"

services:
  streaming-server:
    build:
      context: ./streaming-server
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    depends_on:
      - redis
      - webrtc-server
    environment:
      REDIS_URL: "redis://redis:6379"
      WEBRTC_SERVER_URL: "http://webrtc-server:9000"
      LOG_LEVEL: "info"
    restart: unless-stopped
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 2G

  webrtc-server:
    build:
      context: ./webrtc-server
      dockerfile: Dockerfile
    ports:
      - "9000:9000"
      - "10000-10100:10000-10100/udp"  # WebRTC媒體流埠
    environment:
      TURN_SECRET: ${TURN_SECRET}
      STUN_SERVER: "stun:stun.l.google.com:19302"
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    command: redis-server --appendonly yes --maxmemory 1gb --maxmemory-policy allkeys-lru
    restart: unless-stopped

volumes:
  redis-data:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rust-streaming-server
  namespace: streaming
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rust-streaming
  template:
    metadata:
      labels:
        app: rust-streaming
    spec:
      containers:
      - name: rust-streaming
        image: streaming-registry.example.com/rust-streaming:v1.2.3
        ports:
        - containerPort: 8080
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2"
            memory: "2Gi"
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 15
          periodSeconds: 20
        env:
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: streaming-secrets
              key: redis-url
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: rust-streaming-hpa
  namespace: streaming
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: rust-streaming-server
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60

apiVersion: apps/v1
kind: 有狀態集合
metadata:
  name: redis-cluster
  namespace: streaming
spec:
  serviceName: "redis-cluster"
  replicas: 6
  selector:
    matchLabels:
      app: redis-cluster
  template:
    metadata:
      labels:
        app: redis-cluster
    spec:
      containers:
      - name: redis
        image: redis:7
        command:
          - redis-server
          - "/conf/redis.conf"
        ports:
        - containerPort: 6379
        - containerPort: 16379
        volumeMounts:
        - name: redis-config
          mountPath: /conf
        - name: redis-data
          mountPath: /data
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "1"
            memory: "4Gi"
      volumes:
      - name: redis-config
        configMap:
          name: redis-cluster-config
  volumeClaimTemplates:
  - metadata:
      name: redis-data
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 10Gi
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: redis-cluster-config
  namespace: streaming
data:
  redis.conf: |
    cluster-enabled yes
    cluster-config-file /data/nodes.conf
    cluster-node-timeout 5000
    appendonly yes
    maxmemory 3gb
    maxmemory-policy volatile-lru