fix(via_test): remove signal sync test that was bocking test execution

2025-11-25 22:54:00 -01:00
parent 9776fe0a49
commit 6edace647e
5 changed files with 189 additions and 104 deletions
--- a/context.go
+++ b/context.go
@@ -25,6 +25,7 @@ type Context struct {
 	actionRegistry    map[string]func()
 	signals           *sync.Map
 	mutex             sync.RWMutex
+	ctxDisposedChan   chan struct{}
 }

 // View defines the UI rendered by this context.
@@ -58,7 +59,7 @@ func (c *Context) View(f func() h.H) {
 //			)
 //		})
 //	})
-func (c *Context) Component(f func(c *Context)) func() h.H {
+func (c *Context) Component(initCtx func(c *Context)) func() h.H {
 	id := c.id + "/_component/" + genRandID()
 	compCtx := newContext(id, c.route, c.app)
 	if c.isComponent() {
@@ -66,7 +67,7 @@ func (c *Context) Component(f func(c *Context)) func() h.H {
 	} else {
 		compCtx.parentPageCtx = c
 	}
-	f(compCtx)
+	initCtx(compCtx)
 	c.componentRegistry[id] = compCtx
 	return compCtx.view
 }
@@ -114,6 +115,21 @@ func (c *Context) getActionFn(id string) (func(), error) {
 	return nil, fmt.Errorf("action '%s' not found", id)
 }

+// Routine uses the given initialization handler to define a safe concurrent goroutine
+// that is tied to *Context. The returned *Routine instance provides methods
+// to start, stop or update the routine.
+func (c *Context) Routine(initRoutine func(*Routine)) *Routine {
+	var cn chan struct{}
+	if c.isComponent() { // components use the chan on the parent page ctx
+		cn = c.parentPageCtx.ctxDisposedChan
+	} else {
+		cn = c.ctxDisposedChan
+	}
+	r := newRoutine(cn)
+	initRoutine(r)
+	return r
+}
+
 // Signal creates a reactive signal and initializes it with the given value.
 // Use Bind() to link the value of input elements to the signal and Text() to
 // display the signal value and watch the UI update live as the input changes.
@@ -202,6 +218,8 @@ func (c *Context) getPatchChan() chan patch {
 }

 func (c *Context) prepareSignalsForPatch() map[string]any {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
 	updatedSigs := make(map[string]any)
 	c.signals.Range(func(sigID, value any) bool {
 		if sig, ok := value.(*signal); ok {
@@ -218,31 +236,42 @@ func (c *Context) prepareSignalsForPatch() map[string]any {
 	return updatedSigs
 }

+// sendPatch queues a patch on this *Context sse stream. If the sse is closed or queue is full, the patch
+// is dropped to prevent runtime blocks.
+func (c *Context) sendPatch(p patch) {
+	patchChan := c.getPatchChan()
+	if patchChan == nil {
+		c.app.logWarn(c, "view out of sync: sse stream closed")
+	}
+	select {
+	case patchChan <- p: //queue patch
+	default: // closed or buffer full - drop patch without blocking
+		c.app.logWarn(c, "view out of sync: sse stream closed or queue is full")
+	}
+}
+
 // Sync pushes the current view state and signal changes to the browser immediately
 // over the live SSE event stream.
 func (c *Context) Sync() {
-	patchChan := c.getPatchChan()
 	elemsPatch := bytes.NewBuffer(make([]byte, 0))
 	if err := c.view().Render(elemsPatch); err != nil {
 		c.app.logErr(c, "sync view failed: %v", err)
 		return
 	}
-	patchChan <- patch{patchTypeElements, elemsPatch.String()}
+	c.sendPatch(patch{patchTypeElements, elemsPatch.String()})

-	c.mutex.RLock()
-	defer c.mutex.RUnlock()
 	updatedSigs := c.prepareSignalsForPatch()

 	if len(updatedSigs) != 0 {
 		outgoingSigs, _ := json.Marshal(updatedSigs)
-		patchChan <- patch{patchTypeSignals, string(outgoingSigs)}
+		c.sendPatch(patch{patchTypeSignals, string(outgoingSigs)})
 	}
 }

 // SyncElements pushes an immediate html patch over the live SSE stream to the
 // browser that merges with the DOM
 //
-// For the merge to occur, the top level element in the patch needs to have
+// For the merge to occur, each top lever element in the patch needs to have
 // an ID that matches the ID of an element that already sits in the view.
 //
 // Example:
@@ -254,58 +283,44 @@ func (c *Context) Sync() {
 //		h.P(h.Text("Hello from Via!"))
 //	)
 //
-// Then, the merge will only occur if the ID of the top level element in the patch
+// Then, the merge will only occur if the ID of one of the top level elements in the patch
 // matches 'my-element'.
-func (c *Context) SyncElements(elem h.H) {
-	if elem == nil {
-		c.app.logErr(c, "sync elements failed: view func is nil")
-		return
+func (c *Context) SyncElements(elem ...h.H) {
+	b := bytes.NewBuffer(nil)
+	for idx, el := range elem {
+		if el == nil {
+			c.app.logWarn(c, "sync elements failed: element at idx=%d is nil", idx)
+			continue
+		}
+		if err := el.Render(b); err != nil {
+			c.app.logWarn(c, "sync elements failed: element at idx=%d has invalid html", idx)
+			continue
+		}
 	}
-	patchChan := c.getPatchChan()
-	if patchChan == nil {
-		c.app.logWarn(c, "sync elements failed: no sse stream")
-		return
-	}
-	b := bytes.NewBuffer(make([]byte, 0))
-	_ = elem.Render(b)
-	patchChan <- patch{patchTypeElements, b.String()}
+	c.sendPatch(patch{patchTypeElements, b.String()})
 }

 // SyncSignals pushes the current signal changes to the browser immediately
 // over the live SSE event stream.
 func (c *Context) SyncSignals() {
-	patchChan := c.getPatchChan()
-	if patchChan == nil {
-		c.app.logWarn(c, "signals out of sync: no sse stream")
-		return
-	}
-	updatedSigs := make(map[string]any)
-
-	c.signals.Range(func(key, val any) bool {
-		// We know the types.
-		sig, _ := val.(*signal) // adjust *Signal to your actual signal type
-		id, _ := key.(string)
-		if sig.err != nil {
-			c.app.logWarn(c, "signal out of sync'%s': %v", sig.id, sig.err)
-		}
-		if sig.changed && sig.err == nil {
-			updatedSigs[id] = fmt.Sprintf("%v", sig.val)
-			sig.changed = false
-		}
-		return true // continue iteration
-	})
+	updatedSigs := c.prepareSignalsForPatch()
 	if len(updatedSigs) != 0 {
 		outgoingSignals, _ := json.Marshal(updatedSigs)
-		patchChan <- patch{patchTypeSignals, string(outgoingSignals)}
+		c.sendPatch(patch{patchTypeSignals, string(outgoingSignals)})
 	}
 }

 func (c *Context) ExecScript(s string) {
 	if s == "" {
+		c.app.logWarn(c, "exec script failed: empty script")
 		return
 	}
-	patchChan := c.getPatchChan()
-	patchChan <- patch{patchTypeScript, s}
+	c.sendPatch(patch{patchTypeScript, s})
+}
+
+// stopAllRoutines safely stops all go routines tied to this Context. Prevents goroutine leakage.
+func (c *Context) stopAllRoutines() {
+	close(c.ctxDisposedChan)
 }

 func newContext(id string, route string, v *V) *Context {
@@ -320,6 +335,6 @@ func newContext(id string, route string, v *V) *Context {
 		componentRegistry: make(map[string]*Context),
 		actionRegistry:    make(map[string]func()),
 		signals:           new(sync.Map),
-		patchChan:         make(chan patch, 100),
+		ctxDisposedChan:   make(chan struct{}),
 	}
 }
--- a/internal/examples/realtimechart/main.go
+++ b/internal/examples/realtimechart/main.go
@@ -54,40 +54,50 @@ func chartCompFn(c *via.Context) {
 	labels := make([]string, 1000)

 	isLive := true
+
 	isLiveSig := c.Signal("on")
-	refreshRate := c.Signal(24)
-	tkr := time.NewTicker(1000 / time.Duration(refreshRate.Int()) * time.Millisecond)

-	updateRefreshRate := c.Action(func() {
-		tkr.Reset(1000 / time.Duration(refreshRate.Int()) * time.Millisecond)
-	})
+	refreshRate := c.Signal("1")

-	toggleIsLive := c.Action(func() {
-		isLive = isLiveSig.Bool()
-	})
+	computedTickDuration := func() time.Duration {
+		return 1000 / time.Duration(refreshRate.Int()) * time.Millisecond
+	}

-	go func() {
-		defer tkr.Stop()
-		for range tkr.C {
+	updateData := c.Routine(func(r *via.Routine) {
+
+		r.OnInterval(computedTickDuration(), func() {
 			labels = append(labels[1:], time.Now().Format("15:04:05.000"))
 			data = append(data[1:], rand.Float64()*10)
 			labelsTxt, _ := json.Marshal(labels)
 			dataTxt, _ := json.Marshal(data)

-			if isLive {
-				c.ExecScript(fmt.Sprintf(`
-					if (myChart)
-						myChart.setOption({
-							xAxis: [{data: %s}],
-							series:[{data: %s}]
-						},{
-							notMerge:false,
-							lazyUpdate:true
-						});
+			c.ExecScript(fmt.Sprintf(`
+				if (myChart)
+					myChart.setOption({
+						xAxis: [{data: %s}],
+						series:[{data: %s}]
+					},{
+						notMerge:false,
+						lazyUpdate:true
+					});
 				`, labelsTxt, dataTxt))
-			}
+		})
+
+	})
+	updateData.Start()
+
+	updateRefreshRate := c.Action(func() {
+		updateData.UpdateInterval(computedTickDuration())
+	})
+
+	toggleIsLive := c.Action(func() {
+		isLive = isLiveSig.Bool()
+		if isLive {
+			updateData.Start()
+		} else {
+			updateData.Stop()
 		}
-	}()
+	})

 	c.View(func() h.H {
 		return h.Div(
@@ -96,7 +106,7 @@ func chartCompFn(c *via.Context) {
 				var myChart = echarts.init(document.getElementById('chart'), prefersDark.matches ? 'dark' : 'light');
 				var option = {
 					backgroundColor: prefersDark.matches ? 'transparent' : '#ffffff',
-					animationDurationUpdate: 0,  //  affects updates/redraws
+					animationDurationUpdate: 0, // affects updates/redraws
 					tooltip: {
 						trigger: 'axis',
 						position: function (pt) {
--- a/routine.go
+++ b/routine.go
@@ -0,0 +1,76 @@
+package via
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Routine allows for defining concurrent goroutines safely. Goroutines started by *Routine
+// are tied to the *Context lifecycle.
+type Routine struct {
+	mu             sync.Mutex
+	ctxDisposed    chan struct{}
+	localInterrupt chan struct{}
+	isRunning      atomic.Bool
+	routineFn      func()
+	tckDuration    time.Duration
+	tkr            *time.Ticker
+}
+
+// OnInterval starts a go routine that sets a time.Ticker with the given duration and executes
+// the given func() on every tick. Use *Routine.UpdateInterval to update the interval.
+// If the routine is running, it is stopped.
+func (r *Routine) OnInterval(d time.Duration, fn func()) {
+	if r.isRunning.Load() == true {
+		r.Stop()
+	}
+	r.tckDuration = d
+	r.routineFn = func() {
+		r.tkr = time.NewTicker(r.tckDuration)
+		defer r.tkr.Stop() // clean up the ticker when routine stops
+		for {
+			select {
+			case <-r.ctxDisposed: // dispose of the routine when ctx is disposed
+				return
+			case <-r.localInterrupt: // dispose of the routine on interrupt signal
+				return
+			case <-r.tkr.C:
+				fn()
+			}
+		}
+	}
+}
+
+// UpdateInterval sets a new interval duration for the internal *time.Ticker. If the provided
+// duration is equal of less than 0, UpdateInterval does nothing.
+func (r *Routine) UpdateInterval(d time.Duration) {
+	r.tckDuration = d
+	r.tkr.Reset(d)
+
+}
+
+// Start executes the predifined goroutine. If no predifined goroutine exists, or it already
+// started, Start does nothing.
+func (r *Routine) Start() {
+	if !r.isRunning.CompareAndSwap(false, true) || r.routineFn == nil {
+		return
+	}
+	go r.routineFn()
+}
+
+// Stop interrupts the predifined goroutine. If no predifined goroutine exists, or it already
+// ustopped, Stop does nothing.
+func (r *Routine) Stop() {
+	if !r.isRunning.CompareAndSwap(true, false) || r.routineFn == nil {
+		return
+	}
+	r.localInterrupt <- struct{}{}
+}
+
+func newRoutine(ctxDisposedChan chan struct{}) *Routine {
+	return &Routine{
+		ctxDisposed:    ctxDisposedChan,
+		localInterrupt: make(chan struct{}),
+	}
+}
--- a/via.go
+++ b/via.go
@@ -145,6 +145,7 @@ func (v *V) Page(route string, initContextFn func(c *Context)) {
 		c := newContext("", "", v)
 		initContextFn(c)
 		c.view()
+		c.stopAllRoutines()
 	}()

 	// save page init function allows devmode to restore persisted ctx later
@@ -205,17 +206,17 @@ func (v *V) currSessionNum() int {
 	return len(v.contextRegistry)
 }

-func (v *V) unregisterCtx(id string) {
+func (v *V) unregisterCtx(c *Context) {
 	v.contextRegistryMutex.Lock()
 	defer v.contextRegistryMutex.Unlock()
-	if id == "" {
+	if c.id == "" {
+		v.logErr(c, "unregister ctx failed: ctx contains empty id")
 		return
 	}
-	v.logDebug(nil, "ctx '%s' removed from registry", id)
+	v.logDebug(c, "ctx removed from registry")
+	c.stopAllRoutines()
+	delete(v.contextRegistry, c.id)
 	v.logDebug(nil, "number of sessions in registry: %d", v.currSessionNum())
-
-	delete(v.contextRegistry, id)
-	v.currSessionNum()
 }

 func (v *V) getCtx(id string) (*Context, error) {
@@ -385,11 +386,16 @@ func New() *V {

 		v.logDebug(c, "SSE connection established")

-		if v.cfg.DevMode {
-			c.Sync()
-		} else {
-			c.SyncSignals()
-		}
+		c.patchChan = make(chan patch, 1000)
+		defer close(c.patchChan)
+
+		go func() {
+			if v.cfg.DevMode {
+				c.Sync()
+			} else {
+				c.SyncSignals()
+			}
+		}()

 		for {
 			select {
@@ -398,7 +404,7 @@ func New() *V {
 				return
 			case patch, ok := <-c.patchChan:
 				if !ok {
-					continue
+					return
 				}
 				switch patch.typ {
 				case patchTypeElements:
@@ -465,7 +471,7 @@ func New() *V {
 		if v.cfg.DevMode {
 			v.devModeRemovePersisted(c)
 		}
-		v.unregisterCtx(c.id)
+		v.unregisterCtx(c)

 	})
 	return v
--- a/via_test.go
+++ b/via_test.go
@@ -1,7 +1,6 @@
 package via

 import (
-	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -87,27 +86,6 @@ func TestConfig(t *testing.T) {
 	assert.Equal(t, "Test", v.cfg.DocumentTitle)
 }

-func TestSyncSignals(t *testing.T) {
-	var ctx *Context
-	var sig *signal
-	v := New()
-	v.Page("/", func(c *Context) {
-		ctx = c
-		sig = c.Signal("initial")
-		c.View(func() h.H { return h.Div() })
-	})
-
-	req := httptest.NewRequest("GET", "/", nil)
-	w := httptest.NewRecorder()
-	v.mux.ServeHTTP(w, req)
-
-	sig.SetValue("updated")
-	ctx.SyncSignals()
-
-	patch := <-ctx.patchChan
-	assert.Equal(t, patch.content, fmt.Sprintf(`{"%s":"updated"}`, sig.ID()))
-}
-
 func TestPage_PanicsOnNoView(t *testing.T) {
 	assert.Panics(t, func() {
 		v := New()