package fusenrender import ( "context" "encoding/json" "fmt" "io" "log" "net/http" "time" "github.com/gorilla/websocket" "github.com/lni/dragonboat/v4" "github.com/lni/dragonboat/v4/client" ) var UidCreater = NewUniqueId(1) var PopChannel chan *QueueItem = make(chan *QueueItem, 100) // chan *QueueItem = make(chan *QueueItem, 1) var upgrader = websocket.Upgrader{ ReadBufferSize: 10240, WriteBufferSize: 10240, } func HttpListen(ns *dragonboat.NodeHost, port int) { // http.HandleFunc("/api/render/render_notify", callbackHandler) http.HandleFunc("/api/render/queue/push", pushRenderTaskHandler) http.HandleFunc("/api/ws/render/queue/pop", queueHandler) log.Printf(":%d", port) http.ListenAndServe(fmt.Sprintf(":%d", port), nil) } func queueHandler(w http.ResponseWriter, r *http.Request) { var count int var err error conn, err := upgrader.Upgrade(w, r, nil) if err != nil { log.Println(err) return } defer conn.Close() raddr := conn.RemoteAddr() // laddr := conn.LocalAddr() log.Printf("%s 建立连接 当前unity处理机器数量 %d\n", raddr, DequeueHandler.RefCountAdd(1)) defer func() { log.Printf("%s 退出连接 当前unity处理机器数量 %d\n", raddr, DequeueHandler.RefCountAdd(-1)) }() for { item := <-PopChannel if item == nil { continue } var wdata []byte switch data := item.Data.(type) { case string: wdata = []byte(data) case []byte: wdata = data default: wdata, err = json.Marshal(data) if err != nil { log.Println(err) continue } } err = conn.SetWriteDeadline(time.Now().Add(time.Second * 6)) if err != nil { log.Println(err) log.Println("重新回队") stateClient.PushItem(nil, item) return } // 写回消息 err = conn.WriteMessage(websocket.BinaryMessage, wdata) if err != nil { log.Println(err) log.Println("重新回队") stateClient.PushItem(nil, item) return } count++ log.Println("count:", count) // 打印消息 log.Printf("source: [%s] 数据 推送到unity [%s]\n", item.Source, raddr.String()) } } type Form struct { UserID int `json:"user_id"` GuestID int `json:"guest_id"` APIType int `json:"api_type"` UploadBucket int `json:"upload_bucket"` FileKey string `json:"file_key"` FileData []byte `json:"file_data"` MetaData string `json:"meta_data"` } // RequestCallback结构体 type RequestCallback struct { Sign string `json:"sign"` Time int64 `json:"time"` Info Info `json:"info"` } // Info结构体 type Info struct { TaskID string `json:"task_id"` Image string `json:"image"` } func pushRenderTaskHandler(w http.ResponseWriter, r *http.Request) { // 1. 读取Body内容 body, err := io.ReadAll(r.Body) if err != nil { panic(err) } // 2. 定义结构体 item := QueueItem{} // 3. 解析JSON到结构体 err = json.Unmarshal(body, &item) if err != nil { panic(err) } stateClient.PushItem(nil, &item) } type StateClient struct { nh *dragonboat.NodeHost } var stateClient *StateClient func (cli *StateClient) GetNoOPSession() *client.Session { return cli.nh.GetNoOPSession(shardID) } func (cli *StateClient) PushItem(cs *client.Session, item *QueueItem) { if cs == nil { cs = cli.GetNoOPSession() } cmd := &CmdEnqueue{Command: Command{Group: "unity3d"}} cmd.Item = item data, err := FsPasser.PackToBytes(cmd) if err != nil { log.Println(err) } ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) _, err = cli.nh.SyncPropose(ctx, cs, data) if err != nil { log.Println(err) } // log.Println("enqueue", len(result.Data)) cancel() } func (cli *StateClient) PopItem(cs *client.Session, group string) (*QueueItem, error) { if cs == nil { cs = cli.GetNoOPSession() } cmd := &CmdDequeue{Command{Group: "unity3d"}} data, err := FsPasser.PackToBytes(cmd) if err != nil { log.Println(err) } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) result, err := cli.nh.SyncPropose(ctx, cs, data) cancel() if err != nil { return nil, err } if len(result.Data) == 0 { return nil, nil } var item QueueItem err = item.Decode(result.Data) if err != nil { log.Println(err) } return &item, nil }