Merge pull request #17 from JIeJaitt/stream_server

README.md

@@ -1,10 +1,13 @@
 # StmSrv: streaming server
 
+> **Note**  
+> 本项目目前仅用于学习
+
 流媒体点播服务
 
 ![](image/Overall-architecture.png)
 
-https://blog.csdn.net/weixin_52723461/article/details/120277314?csdn_share_tail=%7B%22type%22%3A%22blog%22%2C%22rType%22%3A%22article%22%2C%22rId%22%3A%22120277314%22%2C%22source%22%3A%22weixin_52723461%22%7D
+https://jiejaitt.blog.csdn.net/article/details/120277314
 
 # LICENSE
 [MIT](tps://github.com/JIeJaitt/stmsrv/blob/5aea553bd697a9906484eae470eac5b10123e9f8/LICENSE)

api/dbops/api_test.go

@@ -75,7 +75,7 @@ func TestVideoWorkFlow(t *testing.T) {
 }
 
 func testAddVideoInfo(t *testing.T) {
-	vi, err := AddNewVideo(1, "my-video")
+	vi, err := AddNewVideo(1, "my-videos")
 	if err != nil {
 		t.Errorf("Error of AddVideoInfo: %v", err)
 	}
@@ -113,7 +113,7 @@ func TestComments(t *testing.T) {
 func testAddComments(t *testing.T) {
 	vid := "12345"
 	aid := 1
-	content := "I like this video"
+	content := "I like this videos"
 
 	err := AddNewComments(vid, aid, content)
 	if err != nil {

streamserver/defs.go

@@ -0,0 +1,6 @@
+package main
+
+const (
+	VIDEO_DIR       = "../videos/"
+	MAX_UPLOAD_SIZE = 1024 * 1024 * 500 // 500MB
+)

streamserver/handlers.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"github.com/julienschmidt/httprouter"
+	"html/template"
+	"io"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"time"
+)
+
+func testPageHandler(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
+	t, _ := template.ParseFiles("../videos/upload.html")
+	t.Execute(w, nil)
+}
+
+func streamHandler(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
+	// 从 URL 中获取 vid-id
+	vid := p.ByName("vid-id")
+	// 获取文件目录
+	vl := VIDEO_DIR + vid
+
+	// 打开文件
+	video, err := os.Open(vl)
+	if err != nil {
+		log.Printf("Error when try to open file: %v", err)
+		sendErrorResponse(w, http.StatusInternalServerError, "Internal error.")
+		return
+	}
+
+	w.Header().Set("Content-Type", "video/mp4")
+	// 把文件作为二进制流传输给客户端
+	http.ServeContent(w, r, "", time.Now(), video)
+	defer video.Close()
+}
+
+// uploadHandler是一个处理上传文件请求的处理函数
+func uploadHandler(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
+	// 设置请求体的最大字节数限制为MAX_UPLOAD_SIZE
+	r.Body = http.MaxBytesReader(w, r.Body, MAX_UPLOAD_SIZE)
+
+	// 解析多部分表单数据，其中包含上传的文件
+	err := r.ParseMultipartForm(MAX_UPLOAD_SIZE)
+	if err != nil {
+		// 发生错误时，向客户端发送400错误响应
+		sendErrorResponse(w, http.StatusBadRequest, "Failed to parse multipart form")
+		return
+	}
+
+	// 从表单中获取上传的文件
+	file, _, err := r.FormFile("file")
+	if err != nil {
+		// 发生错误时，向客户端发送400错误响应
+		sendErrorResponse(w, http.StatusBadRequest, "Failed to get file from form")
+		return
+	}
+	defer file.Close()
+
+	// 读取文件的内容
+	data, err := io.ReadAll(file)
+	if err != nil {
+		// 发生错误时，记录错误日志并向客户端发送500错误响应
+		log.Printf("Failed to read file: %v", err)
+		sendErrorResponse(w, http.StatusInternalServerError, "Failed to read file")
+		return
+	}
+
+	// 获取URL中的vid-id参数
+	vid := p.ByName("vid-id")
+
+	// 构建文件的完整路径
+	filePath := VIDEO_DIR + vid
+
+	// 将文件内容写入到指定路径的文件中
+	err = ioutil.WriteFile(filePath, data, 0666)
+	if err != nil {
+		// 发生错误时，记录错误日志并向客户端发送500错误响应
+		log.Printf("Failed to write file: %v", err)
+		sendErrorResponse(w, http.StatusInternalServerError, "Failed to write file")
+		return
+	}
+
+	// 向客户端发送201状态码，表示文件上传成功
+	w.WriteHeader(http.StatusCreated)
+	// 向客户端发送成功消息
+	io.WriteString(w, "File uploaded successfully")
+}

streamserver/limiter.go

@@ -0,0 +1,34 @@
+package main
+
+import "log"
+
+// ConnLimiter 是一个结构体，用于限制并发连接数
+type ConnLimiter struct {
+	concurrentConn int      // 并发连接数
+	bucket         chan int // 令牌桶通道
+}
+
+// NewConnLimiter 是 ConnLimiter 的构造函数
+func NewConnLimiter(cc int) *ConnLimiter {
+	return &ConnLimiter{
+		concurrentConn: cc,
+		bucket:         make(chan int, cc), // 缓冲区大小为 cc
+	}
+}
+
+func (cl *ConnLimiter) GetConn() bool {
+	// 如果通道已满，说明已经达到并发连接数的上限，返回 false
+	if len(cl.bucket) >= cl.concurrentConn {
+		log.Printf("Reached the rate limitation.")
+		return false
+	}
+	// 如果通道未满，说明还有空闲的连接，返回 true
+	cl.bucket <- 1
+	return true
+}
+
+// ReleaseConn 用于释放连接
+func (cl *ConnLimiter) ReleaseConn() {
+	c := <-cl.bucket
+	log.Printf("New connection coming: %d", c)
+}

streamserver/main.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"github.com/julienschmidt/httprouter"
+	"net/http"
+)
+
+type middleWareHandler struct {
+	r *httprouter.Router
+	l *ConnLimiter
+}
+
+func NewMiddleWareHandler(r *httprouter.Router, cc int) http.Handler {
+	m := middleWareHandler{}
+	m.r = r
+	m.l = NewConnLimiter(cc)
+	return m
+}
+
+func (m middleWareHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if !m.l.GetConn() {
+		sendErrorResponse(w, http.StatusTooManyRequests, "Too many requests.")
+		return
+	}
+	m.r.ServeHTTP(w, r)
+	defer m.l.ReleaseConn()
+}
+
+func RegisterHandler() *httprouter.Router {
+	router := httprouter.New()
+	router.GET("/videos/:vid-id", streamHandler)
+	router.POST("/upload/:vid-id", uploadHandler)
+	router.GET("/testpage", testPageHandler)
+	return router
+}
+
+func main() {
+	r := RegisterHandler()
+	mh := NewMiddleWareHandler(r, 2)
+	http.ListenAndServe(":9000", mh)
+}

streamserver/response.go

@@ -0,0 +1,13 @@
+package main
+
+import (
+	"io"
+	"net/http"
+)
+
+func sendErrorResponse(w http.ResponseWriter, sc int, errMsg string) {
+	// 通过 sc(status code) 设置状态码
+	w.WriteHeader(sc)
+	// 通过 io.WriteString 写入错误信息返回给客户端
+	io.WriteString(w, errMsg)
+}

videos/upload.html

@@ -0,0 +1,16 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>Test Upload a File</title>
+</head>
+<body>
+    <form enctype="multipart/form-data" action="http://127.0.0.1:9000/upload/ddd" method="post">
+        {{/* 1. File input */}}
+        <input type="file" name="file" />
+
+        {{/* 2. Submit button */}}
+        <input type="submit" value="upload file" />
+    </form>
+</body>
+</html>