fix(app): use event loop runtime for every operations
arcad/edge/pipeline/head This commit looks good Details

This commit is contained in:
wpetit 2023-04-24 12:16:30 +02:00
parent abc60b9ae3
commit ba9ae6e391
9 changed files with 137 additions and 81 deletions

View File

@ -250,10 +250,10 @@ type ModuleDepFunc func(*moduleDeps) error
func getServerModules(deps *moduleDeps) []app.ServerModuleFactory { func getServerModules(deps *moduleDeps) []app.ServerModuleFactory {
return []app.ServerModuleFactory{ return []app.ServerModuleFactory{
module.LifecycleModuleFactory(),
module.ContextModuleFactory(), module.ContextModuleFactory(),
module.ConsoleModuleFactory(), module.ConsoleModuleFactory(),
cast.CastModuleFactory(), cast.CastModuleFactory(),
module.LifecycleModuleFactory(),
netModule.ModuleFactory(deps.Bus), netModule.ModuleFactory(deps.Bus),
module.RPCModuleFactory(deps.Bus), module.RPCModuleFactory(deps.Bus),
module.StoreModuleFactory(deps.DocumentStore), module.StoreModuleFactory(deps.DocumentStore),

View File

@ -39,3 +39,14 @@ func NewPromiseProxy(promise *goja.Promise, resolve func(result interface{}), re
return proxy return proxy
} }
func NewPromiseProxyFrom(rt *goja.Runtime) *PromiseProxy {
promise, resolve, reject := rt.NewPromise()
return NewPromiseProxy(promise, resolve, reject)
}
func IsPromise(v goja.Value) (*goja.Promise, bool) {
promise, ok := v.Export().(*goja.Promise)
return promise, ok
}

View File

@ -17,15 +17,22 @@ var (
) )
type Server struct { type Server struct {
runtime *goja.Runtime loop *eventloop.EventLoop
loop *eventloop.EventLoop factories []ServerModuleFactory
modules []ServerModule modules []ServerModule
} }
func (s *Server) Load(name string, src string) error { func (s *Server) Load(name string, src string) error {
_, err := s.runtime.RunScript(name, src) var err error
s.loop.RunOnLoop(func(rt *goja.Runtime) {
_, err = rt.RunScript(name, src)
if err != nil {
err = errors.Wrap(err, "could not run js script")
}
})
if err != nil { if err != nil {
return errors.Wrap(err, "could not run js script") return errors.WithStack(err)
} }
return nil return nil
@ -34,15 +41,15 @@ func (s *Server) Load(name string, src string) error {
func (s *Server) ExecFuncByName(ctx context.Context, funcName string, args ...interface{}) (goja.Value, error) { func (s *Server) ExecFuncByName(ctx context.Context, funcName string, args ...interface{}) (goja.Value, error) {
ctx = logger.With(ctx, logger.F("function", funcName), logger.F("args", args)) ctx = logger.With(ctx, logger.F("function", funcName), logger.F("args", args))
callable, ok := goja.AssertFunction(s.runtime.Get(funcName)) ret, err := s.Exec(ctx, funcName, args...)
if !ok { if err != nil {
return nil, errors.WithStack(ErrFuncDoesNotExist) return nil, errors.WithStack(err)
} }
return s.Exec(ctx, callable, args...) return ret, nil
} }
func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...interface{}) (goja.Value, error) { func (s *Server) Exec(ctx context.Context, callableOrFuncname any, args ...interface{}) (goja.Value, error) {
var ( var (
wg sync.WaitGroup wg sync.WaitGroup
value goja.Value value goja.Value
@ -51,7 +58,28 @@ func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...inter
wg.Add(1) wg.Add(1)
s.loop.RunOnLoop(func(vm *goja.Runtime) { s.loop.RunOnLoop(func(rt *goja.Runtime) {
var callable goja.Callable
switch typ := callableOrFuncname.(type) {
case goja.Callable:
callable = typ
case string:
call, ok := goja.AssertFunction(rt.Get(typ))
if !ok {
err = errors.WithStack(ErrFuncDoesNotExist)
return
}
callable = call
default:
err = errors.Errorf("callableOrFuncname: expected callable or function name, got '%T'", callableOrFuncname)
return
}
logger.Debug(ctx, "executing callable") logger.Debug(ctx, "executing callable")
defer wg.Done() defer wg.Done()
@ -73,7 +101,7 @@ func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...inter
jsArgs := make([]goja.Value, 0, len(args)) jsArgs := make([]goja.Value, 0, len(args))
for _, a := range args { for _, a := range args {
jsArgs = append(jsArgs, vm.ToValue(a)) jsArgs = append(jsArgs, rt.ToValue(a))
} }
value, err = callable(nil, jsArgs...) value, err = callable(nil, jsArgs...)
@ -84,12 +112,11 @@ func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...inter
wg.Wait() wg.Wait()
return value, err if err != nil {
} return nil, errors.WithStack(err)
}
func (s *Server) IsPromise(v goja.Value) (*goja.Promise, bool) { return value, nil
promise, ok := v.Export().(*goja.Promise)
return promise, ok
} }
func (s *Server) WaitForPromise(promise *goja.Promise) goja.Value { func (s *Server) WaitForPromise(promise *goja.Promise) goja.Value {
@ -135,28 +162,21 @@ func (s *Server) WaitForPromise(promise *goja.Promise) goja.Value {
return value return value
} }
func (s *Server) NewPromise() *PromiseProxy {
promise, resolve, reject := s.runtime.NewPromise()
return NewPromiseProxy(promise, resolve, reject)
}
func (s *Server) ToValue(v interface{}) goja.Value {
return s.runtime.ToValue(v)
}
func (s *Server) Start() error { func (s *Server) Start() error {
s.loop.Start() s.loop.Start()
for _, mod := range s.modules { var err error
initMod, ok := mod.(InitializableModule)
if !ok {
continue
}
if err := initMod.OnInit(); err != nil { s.loop.RunOnLoop(func(rt *goja.Runtime) {
return errors.WithStack(err) rt.SetFieldNameMapper(goja.TagFieldNameMapper("goja", true))
rt.SetRandSource(createRandomSource())
if err = s.initModules(rt); err != nil {
err = errors.WithStack(err)
} }
})
if err != nil {
return errors.WithStack(err)
} }
return nil return nil
@ -166,36 +186,46 @@ func (s *Server) Stop() {
s.loop.Stop() s.loop.Stop()
} }
func (s *Server) initModules(factories ...ServerModuleFactory) { func (s *Server) initModules(rt *goja.Runtime) error {
runtime := goja.New() modules := make([]ServerModule, 0, len(s.factories))
runtime.SetFieldNameMapper(goja.TagFieldNameMapper("goja", true)) for _, moduleFactory := range s.factories {
runtime.SetRandSource(createRandomSource())
modules := make([]ServerModule, 0, len(factories))
for _, moduleFactory := range factories {
mod := moduleFactory(s) mod := moduleFactory(s)
export := runtime.NewObject()
export := rt.NewObject()
mod.Export(export) mod.Export(export)
runtime.Set(mod.Name(), export)
rt.Set(mod.Name(), export)
modules = append(modules, mod) modules = append(modules, mod)
} }
s.runtime = runtime for _, mod := range modules {
initMod, ok := mod.(InitializableModule)
if !ok {
continue
}
logger.Debug(context.Background(), "initializing module", logger.F("module", initMod.Name()))
if err := initMod.OnInit(rt); err != nil {
return errors.WithStack(err)
}
}
s.modules = modules s.modules = modules
return nil
} }
func NewServer(factories ...ServerModuleFactory) *Server { func NewServer(factories ...ServerModuleFactory) *Server {
server := &Server{ server := &Server{
factories: factories,
loop: eventloop.NewEventLoop( loop: eventloop.NewEventLoop(
eventloop.EnableConsole(false), eventloop.EnableConsole(false),
), ),
} }
server.initModules(factories...)
return server return server
} }

View File

@ -13,5 +13,5 @@ type ServerModule interface {
type InitializableModule interface { type InitializableModule interface {
ServerModule ServerModule
OnInit() error OnInit(rt *goja.Runtime) error
} }

View File

@ -53,6 +53,10 @@ func (m *Module) Export(export *goja.Object) {
if err := export.Set("CLAIM_PREFERRED_USERNAME", ClaimPreferredUsername); err != nil { if err := export.Set("CLAIM_PREFERRED_USERNAME", ClaimPreferredUsername); err != nil {
panic(errors.Wrap(err, "could not set 'CLAIM_PREFERRED_USERNAME' property")) panic(errors.Wrap(err, "could not set 'CLAIM_PREFERRED_USERNAME' property"))
} }
if err := export.Set("CLAIM_ISSUER", ClaimIssuer); err != nil {
panic(errors.Wrap(err, "could not set 'CLAIM_ISSUER' property"))
}
} }
func (m *Module) getClaim(call goja.FunctionCall, rt *goja.Runtime) goja.Value { func (m *Module) getClaim(call goja.FunctionCall, rt *goja.Runtime) goja.Value {

View File

@ -15,10 +15,7 @@ const (
defaultTimeout = 30 * time.Second defaultTimeout = 30 * time.Second
) )
type Module struct { type Module struct{}
ctx context.Context
server *app.Server
}
func (m *Module) Name() string { func (m *Module) Name() string {
return "cast" return "cast"
@ -54,7 +51,7 @@ func (m *Module) refreshDevices(call goja.FunctionCall, rt *goja.Runtime) goja.V
panic(rt.ToValue(errors.WithStack(err))) panic(rt.ToValue(errors.WithStack(err)))
} }
promise := m.server.NewPromise() promise := app.NewPromiseProxyFrom(rt)
go func() { go func() {
ctx, cancel := context.WithTimeout(context.Background(), timeout) ctx, cancel := context.WithTimeout(context.Background(), timeout)
@ -102,7 +99,7 @@ func (m *Module) loadUrl(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
panic(rt.ToValue(errors.WithStack(err))) panic(rt.ToValue(errors.WithStack(err)))
} }
promise := m.server.NewPromise() promise := app.NewPromiseProxyFrom(rt)
go func() { go func() {
ctx, cancel := context.WithTimeout(context.Background(), timeout) ctx, cancel := context.WithTimeout(context.Background(), timeout)
@ -121,7 +118,7 @@ func (m *Module) loadUrl(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
promise.Resolve(nil) promise.Resolve(nil)
}() }()
return m.server.ToValue(promise) return rt.ToValue(promise)
} }
func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value { func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
@ -137,7 +134,7 @@ func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
panic(rt.ToValue(errors.WithStack(err))) panic(rt.ToValue(errors.WithStack(err)))
} }
promise := m.server.NewPromise() promise := app.NewPromiseProxyFrom(rt)
go func() { go func() {
ctx, cancel := context.WithTimeout(context.Background(), timeout) ctx, cancel := context.WithTimeout(context.Background(), timeout)
@ -156,7 +153,7 @@ func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
promise.Resolve(nil) promise.Resolve(nil)
}() }()
return m.server.ToValue(promise) return rt.ToValue(promise)
} }
func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value { func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
@ -172,7 +169,7 @@ func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value
panic(rt.ToValue(errors.WithStack(err))) panic(rt.ToValue(errors.WithStack(err)))
} }
promise := m.server.NewPromise() promise := app.NewPromiseProxyFrom(rt)
go func() { go func() {
ctx, cancel := context.WithTimeout(context.Background(), timeout) ctx, cancel := context.WithTimeout(context.Background(), timeout)
@ -191,7 +188,7 @@ func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value
promise.Resolve(status) promise.Resolve(status)
}() }()
return m.server.ToValue(promise) return rt.ToValue(promise)
} }
func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) { func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) {
@ -214,8 +211,6 @@ func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) {
func CastModuleFactory() app.ServerModuleFactory { func CastModuleFactory() app.ServerModuleFactory {
return func(server *app.Server) app.ServerModule { return func(server *app.Server) app.ServerModule {
return &Module{ return &Module{}
server: server,
}
} }
} }

View File

@ -85,7 +85,7 @@ func TestCastModuleRefreshDevices(t *testing.T) {
t.Error(errors.WithStack(err)) t.Error(errors.WithStack(err))
} }
promise, ok := server.IsPromise(result) promise, ok := app.IsPromise(result)
if !ok { if !ok {
t.Fatal("expected promise") t.Fatal("expected promise")
} }

View File

@ -9,9 +9,7 @@ import (
"gitlab.com/wpetit/goweb/logger" "gitlab.com/wpetit/goweb/logger"
) )
type LifecycleModule struct { type LifecycleModule struct{}
server *app.Server
}
func (m *LifecycleModule) Name() string { func (m *LifecycleModule) Name() string {
return "lifecycle" return "lifecycle"
@ -20,25 +18,37 @@ func (m *LifecycleModule) Name() string {
func (m *LifecycleModule) Export(export *goja.Object) { func (m *LifecycleModule) Export(export *goja.Object) {
} }
func (m *LifecycleModule) OnInit() error { func (m *LifecycleModule) OnInit(rt *goja.Runtime) (err error) {
if _, err := m.server.ExecFuncByName(context.Background(), "onInit"); err != nil { call, ok := goja.AssertFunction(rt.Get("onInit"))
if errors.Is(err, app.ErrFuncDoesNotExist) { if !ok {
logger.Warn(context.Background(), "could not find onInit() function", logger.E(errors.WithStack(err))) logger.Warn(context.Background(), "could not find onInit() function")
return nil return nil
}
return errors.WithStack(err)
} }
defer func() {
if recovered := recover(); recovered != nil {
revoveredErr, ok := recovered.(error)
if ok {
logger.Error(context.Background(), "recovered runtime error", logger.E(errors.WithStack(revoveredErr)))
err = errors.WithStack(app.ErUnknownError)
return
}
panic(recovered)
}
}()
call(nil)
return nil return nil
} }
func LifecycleModuleFactory() app.ServerModuleFactory { func LifecycleModuleFactory() app.ServerModuleFactory {
return func(server *app.Server) app.ServerModule { return func(server *app.Server) app.ServerModule {
module := &LifecycleModule{ module := &LifecycleModule{}
server: server,
}
return module return module
} }

View File

@ -51,6 +51,12 @@ func (m *RPCModule) Export(export *goja.Object) {
} }
} }
func (m *RPCModule) OnInit(rt *goja.Runtime) error {
go m.handleMessages()
return nil
}
func (m *RPCModule) register(call goja.FunctionCall, rt *goja.Runtime) goja.Value { func (m *RPCModule) register(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
fnName := util.AssertString(call.Argument(0), rt) fnName := util.AssertString(call.Argument(0), rt)
@ -181,7 +187,7 @@ func (m *RPCModule) handleMessages() {
continue continue
} }
promise, ok := m.server.IsPromise(result) promise, ok := app.IsPromise(result)
if ok { if ok {
go func(ctx context.Context, req *RPCRequest, promise *goja.Promise) { go func(ctx context.Context, req *RPCRequest, promise *goja.Promise) {
result := m.server.WaitForPromise(promise) result := m.server.WaitForPromise(promise)
@ -263,8 +269,8 @@ func RPCModuleFactory(bus bus.Bus) app.ServerModuleFactory {
bus: bus, bus: bus,
} }
go mod.handleMessages()
return mod return mod
} }
} }
var _ app.InitializableModule = &RPCModule{}