修复部分问题

ExplodingDragon · ExplodingDragon · commit 35df220beab8 · 2025-11-20T17:33:18.000+08:00
diff --git a/examples/js_ws/index.js b/examples/js_ws/index.js
@@ -1,4 +1,4 @@
-async function run() {
+(async ()=>{
     let ws = websocket();
     let shouldExit = false;
     while (!shouldExit) {
@@ -17,5 +17,4 @@ async function run() {
                 break;
         }
     }
-}
-run().then(r => {});
+})()
diff --git a/examples/js_ws_event/event.js b/examples/js_ws_event/event.js
@@ -1,28 +1,31 @@
-const name = (await request.getQuery("name"))?.trim();
+const name = (request.getQuery("name"))?.trim();
 
 if (!name) {
     throw new Error(`Missing or empty name parameter`);
 }
 
 const ws = websocket();
 
-try {
-    // 事件处理
-    event.subscribe("messages").on((msg) => {
-        ws.writeText(msg);
-    });
+async function eventPull() {
+    while (true) {
+        const data  = await event.pull('messages')
+        ws.writeText(data);
+    }
+}
 
-    // 主循环
-    for await (const data of ws.readText()) {
+async function messagePull() {
+    while (true) {
+        const data  = await ws.readText()
         if (data === "exit") break;
-
         if (data?.trim()) {
             await event.put("messages", JSON.stringify({
-                name,
+                name:name,
                 data: data.trim()
             }));
         }
     }
-} finally {
-    ws.close();
-}
+}
+
+(async () => {
+    await Promise.all(eventPull(), messagePull())
+})()
diff --git a/pkg/core/filter.go b/pkg/core/filter.go
@@ -51,7 +51,9 @@ func NextCallWrapper(call FilterCall, parentCall NextCall, stack Filter) NextCal
 	return func(ctx FilterContext, writer http.ResponseWriter, request *http.Request) error {
 		zap.L().Debug(fmt.Sprintf("call filter(%s) before", stack.Type), zap.Any("filter", stack))
 		err := call(ctx, writer, request, parentCall)
-		zap.L().Debug(fmt.Sprintf("call filter(%s) after", stack.Type), zap.Any("filter", stack), zap.Error(err))
+		zap.L().Debug(fmt.Sprintf("call filter(%s) after", stack.Type),
+			zap.Any("filter", stack),
+			zap.Error(err))
 		return err
 	}
 }
diff --git a/pkg/filters/goja/goja.go b/pkg/filters/goja/goja.go
@@ -1,8 +1,9 @@
 package goja
 
 import (
-	"context"
+	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"path/filepath"
@@ -14,6 +15,7 @@ import (
 	"github.com/dop251/goja_nodejs/eventloop"
 	"github.com/dop251/goja_nodejs/require"
 	"github.com/dop251/goja_nodejs/url"
+	"go.uber.org/zap"
 	"gopkg.d7z.net/gitea-pages/pkg/core"
 )
 
@@ -47,75 +49,102 @@ func FilterInstGoJa(gl core.Params) (core.FilterInstance, error) {
 			if err != nil {
 				return err
 			}
-			prg, err := goja.Compile("main.js", js, false)
+
+			debug := NewDebug(global.EnableDebug && param.Debug && request.URL.Query().
+				Get("debug") == "true", request, w)
+			program, err := goja.Compile("main.js", js, false)
 			if err != nil {
-				return err
+				return debug.Flush(err)
 			}
-			debug := NewDebug(global.EnableDebug && param.Debug && request.URL.Query().Get("debug") == "true", request, w)
-			registry := newRegistry(ctx)
-			registry.RegisterNativeModule(console.ModuleName, console.RequireWithPrinter(debug))
-			loop := eventloop.NewEventLoop(eventloop.WithRegistry(registry), eventloop.EnableConsole(true))
-			stop := make(chan struct{}, 1)
-			shutdown := make(chan struct{}, 1)
-			defer close(shutdown)
-			timeout, timeoutCancelFunc := context.WithTimeout(ctx, global.Timeout)
-			defer timeoutCancelFunc()
-			count := 0
+			registry := newRegistry(ctx, debug)
+			jsLoop := eventloop.NewEventLoop(eventloop.WithRegistry(registry),
+				eventloop.EnableConsole(true))
+
+			jsLoop.Start()
+			defer jsLoop.Stop()
+
 			closers := NewClosers()
 			defer closers.Close()
-			go func() {
-				defer func() {
-					shutdown <- struct{}{}
+
+			stop := make(chan error, 1)
+			defer close(stop)
+
+			jsLoop.RunOnLoop(func(vm *goja.Runtime) {
+				err := func() error {
+					url.Enable(vm)
+					if err = RequestInject(ctx, vm, request); err != nil {
+						return err
+					}
+					if err = ResponseInject(vm, debug, request); err != nil {
+						return err
+					}
+					if err = KVInject(ctx, vm); err != nil {
+						return err
+					}
+					if err = EventInject(ctx, vm, jsLoop); err != nil {
+						return err
+					}
+					if global.EnableWebsocket {
+						var closer io.Closer
+						closer, err = WebsocketInject(ctx, vm, debug, request, jsLoop)
+						if err != nil {
+							return err
+						}
+						closers.AddCloser(closer.Close)
+					}
+					return nil
 				}()
-				select {
-				case <-timeout.Done():
-				case <-stop:
-				}
-				count = loop.Stop()
-			}()
-			loop.Run(func(vm *goja.Runtime) {
-				url.Enable(vm)
-				if err = RequestInject(ctx, vm, request); err != nil {
-					panic(err)
+				if err != nil {
+					stop <- errors.Join(err, errors.New("js init failed"))
+					return
 				}
-				if err = ResponseInject(vm, debug, request); err != nil {
-					panic(err)
+				result, err := vm.RunProgram(program)
+				if err != nil {
+					stop <- err
+					return
 				}
-				if err = KVInject(ctx, vm); err != nil {
-					panic(err)
-				}
-				if err = EventInject(ctx, vm); err != nil {
-					panic(err)
-				}
-				if global.EnableWebsocket {
-					var closer io.Closer
-					closer, err = WebsocketInject(ctx, vm, debug, request, loop, timeoutCancelFunc)
-					if err != nil {
-						panic(err)
+				export := result.Export()
+				if export != nil {
+					if promise, ok := export.(*goja.Promise); ok {
+						go func() {
+							for {
+								switch promise.State() {
+								case goja.PromiseStateFulfilled, goja.PromiseStateRejected:
+									result := promise.Result().Export()
+									switch data := result.(type) {
+									case error:
+										stop <- data
+									default:
+										marshal, _ := json.Marshal(result)
+										zap.L().Debug(fmt.Sprintf("js promise result %s", string(marshal)),
+											zap.Any("result", promise.Result().ExportType()))
+										stop <- nil
+									}
+									return
+								default:
+									time.Sleep(time.Millisecond * 5)
+								}
+							}
+						}()
 					}
-					closers.AddCloser(closer.Close)
 				}
-				_, err = vm.RunProgram(prg)
 			})
-			stop <- struct{}{}
-			close(stop)
-			<-shutdown
-			if count != 0 {
-				err = errors.Join(context.DeadlineExceeded, err)
-			}
-			return debug.Flush(err)
+			resultErr := <-stop
+			return debug.Flush(resultErr)
 		}, nil
 	}, nil
 }
 
-func newRegistry(ctx core.FilterContext) *require.Registry {
+func newRegistry(ctx core.FilterContext, printer console.Printer) *require.Registry {
 	registry := require.NewRegistry(
 		require.WithLoader(func(path string) ([]byte, error) {
 			return ctx.PageVFS.Read(ctx, path)
 		}),
 		require.WithPathResolver(func(base, path string) string {
 			return filepath.Join(base, filepath.FromSlash(path))
 		}))
+	registry.RegisterNativeModule(console.ModuleName, console.RequireWithPrinter(printer))
+
 	return registry
 }
 
diff --git a/pkg/filters/goja/var_event.go b/pkg/filters/goja/var_event.go
@@ -2,39 +2,33 @@ package goja
 
 import (
 	"github.com/dop251/goja"
+	"github.com/dop251/goja_nodejs/eventloop"
 	"gopkg.d7z.net/gitea-pages/pkg/core"
 )
 
-func EventInject(ctx core.FilterContext, jsCtx *goja.Runtime) error {
+func EventInject(ctx core.FilterContext, jsCtx *goja.Runtime, loop *eventloop.EventLoop) error {
 	return jsCtx.GlobalObject().Set("event", map[string]interface{}{
-		"subscribe": func(key string) (map[string]any, error) {
-			subscribe, err := ctx.Event.Subscribe(ctx, key)
-			if err != nil {
-				return nil, err
-			}
-			return map[string]any{
-				"on": func(f func(string)) {
-					go func() {
-					z:
-						for {
-							select {
-							case <-ctx.Done():
-								break z
-							case data := <-subscribe:
-								f(data)
-							}
-						}
-					}()
-				},
-				"get": func() (string, error) {
-					select {
-					case <-ctx.Done():
-						return "", ctx.Err()
-					case data := <-subscribe:
-						return data, nil
-					}
-				},
-			}, nil
+		"load": func(key string) *goja.Promise {
+			promise, resolve, reject := jsCtx.NewPromise()
+			go func() {
+				subscribe, err := ctx.Event.Subscribe(ctx, key)
+				if err != nil {
+					loop.RunOnLoop(func(runtime *goja.Runtime) {
+						_ = reject(runtime.ToValue(err))
+					})
+				}
+				select {
+				case data := <-subscribe:
+					loop.RunOnLoop(func(runtime *goja.Runtime) {
+						_ = resolve(runtime.ToValue(data))
+					})
+				case <-ctx.Done():
+					loop.RunOnLoop(func(runtime *goja.Runtime) {
+						_ = reject(runtime.ToValue(ctx.Err()))
+					})
+				}
+			}()
+			return promise
 		},
 		"put": func(key, value string) error {
 			return ctx.Event.Publish(ctx, key, value)
diff --git a/pkg/filters/goja/var_websocket.go b/pkg/filters/goja/var_websocket.go
@@ -1,7 +1,6 @@
 package goja
 
 import (
-	"context"
 	"io"
 	"net/http"
 	"time"
@@ -14,15 +13,14 @@ import (
 	"gopkg.d7z.net/gitea-pages/pkg/core"
 )
 
-func WebsocketInject(ctx core.FilterContext, jsCtx *goja.Runtime, w http.ResponseWriter, request *http.Request, loop *eventloop.EventLoop, cancelFunc context.CancelFunc) (io.Closer, error) {
+func WebsocketInject(ctx core.FilterContext, jsCtx *goja.Runtime, w http.ResponseWriter, request *http.Request, loop *eventloop.EventLoop) (io.Closer, error) {
 	closers := NewClosers()
 	return closers, jsCtx.GlobalObject().Set("websocket", func() (any, error) {
 		upgrader := websocket.Upgrader{}
 		conn, err := upgrader.Upgrade(w, request, nil)
 		if err != nil {
 			return nil, err
 		}
-		cancelFunc()
 		go func() {
 			ticker := time.NewTicker(15 * time.Second)
 			defer ticker.Stop()
@@ -63,32 +61,35 @@ func WebsocketInject(ctx core.FilterContext, jsCtx *goja.Runtime, w http.Respons
 			},
 			"TypeTextMessage":   websocket.TextMessage,
 			"TypeBinaryMessage": websocket.BinaryMessage,
-			"readText": func() goja.Value {
+			"readText": func() *goja.Promise {
 				promise, resolve, reject := jsCtx.NewPromise()
 				go func() {
 					select {
 					case <-ctx.Done():
+						loop.RunOnLoop(func(runtime *goja.Runtime) {
+							_ = reject(runtime.ToValue(ctx.Err()))
+						})
 						return
 					default:
 					}
 					defer func() {
 						if r := recover(); r != nil {
 							zap.L().Debug("websocket panic", zap.Any("panic", r))
-							loop.Run(func(runtime *goja.Runtime) {
+							loop.RunOnLoop(func(runtime *goja.Runtime) {
 								_ = reject(runtime.ToValue(r))
 							})
 						}
 					}()
 					_, p, err := conn.ReadMessage()
-					loop.Run(func(runtime *goja.Runtime) {
+					loop.RunOnLoop(func(runtime *goja.Runtime) {
 						if err != nil {
 							_ = reject(runtime.ToValue(err))
 						} else {
 							_ = resolve(runtime.ToValue(string(p)))
 						}
 					})
 				}()
-				return promise.Result()
+				return promise
 			},
 			"read": func() (any, error) {
 				messageType, p, err := conn.ReadMessage()

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-async function run() {`
	`1`	`+(async ()=>{`
`2`	`2`	`let ws = websocket();`
`3`	`3`	`let shouldExit = false;`
`4`	`4`	`while (!shouldExit) {`
`@@ -17,5 +17,4 @@ async function run() {`
`17`	`17`	`break;`
`18`	`18`	`}`
`19`	`19`	`}`
`20`		`-}`
`21`		`-run().then(r => {});`
	`20`	`+})()`
Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,9 @@ func NextCallWrapper(call FilterCall, parentCall NextCall, stack Filter) NextCal`
`51`	`51`	`return func(ctx FilterContext, writer http.ResponseWriter, request *http.Request) error {`
`52`	`52`	`zap.L().Debug(fmt.Sprintf("call filter(%s) before", stack.Type), zap.Any("filter", stack))`
`53`	`53`	`err := call(ctx, writer, request, parentCall)`
`54`		`- zap.L().Debug(fmt.Sprintf("call filter(%s) after", stack.Type), zap.Any("filter", stack), zap.Error(err))`
	`54`	`+ zap.L().Debug(fmt.Sprintf("call filter(%s) after", stack.Type),`
	`55`	`+ zap.Any("filter", stack),`
	`56`	`+ zap.Error(err))`
`55`	`57`	`return err`
`56`	`58`	`}`
`57`	`59`	`}`