Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
4a7acbb630 | ||
|
|
a7ace9099f | ||
|
|
d8318af9c4 | ||
|
|
30cc6d88e6 | ||
|
|
88bd0f31df | ||
|
|
82a3314089 | ||
|
|
73f4e4009b | ||
|
|
c77ccc0796 |
4
.gitignore
vendored
4
.gitignore
vendored
@@ -47,3 +47,7 @@ internal/examples/picocss/picocss
|
|||||||
internal/examples/plugins/plugins
|
internal/examples/plugins/plugins
|
||||||
internal/examples/realtimechart/realtimechart
|
internal/examples/realtimechart/realtimechart
|
||||||
internal/examples/shakespeare/shakespeare
|
internal/examples/shakespeare/shakespeare
|
||||||
|
internal/examples/nats-chatroom/nats-chatroom
|
||||||
|
|
||||||
|
# NATS data directory
|
||||||
|
data/
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
// actionTrigger represents a trigger to an event handler fn
|
// actionTrigger represents a trigger to an event handler fn
|
||||||
|
|||||||
@@ -1,15 +1,17 @@
|
|||||||
package via
|
package via
|
||||||
|
|
||||||
import "github.com/alexedwards/scs/v2"
|
import (
|
||||||
|
"github.com/alexedwards/scs/v2"
|
||||||
|
"github.com/rs/zerolog"
|
||||||
|
)
|
||||||
|
|
||||||
type LogLevel int
|
func ptr(l zerolog.Level) *zerolog.Level { return &l }
|
||||||
|
|
||||||
const (
|
var (
|
||||||
undefined LogLevel = iota
|
LogLevelDebug = ptr(zerolog.DebugLevel)
|
||||||
LogLevelError
|
LogLevelInfo = ptr(zerolog.InfoLevel)
|
||||||
LogLevelWarn
|
LogLevelWarn = ptr(zerolog.WarnLevel)
|
||||||
LogLevelInfo
|
LogLevelError = ptr(zerolog.ErrorLevel)
|
||||||
LogLevelDebug
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// Plugin is a func that can mutate the given *via.V app runtime. It is useful to integrate popular JS/CSS UI libraries or tools.
|
// Plugin is a func that can mutate the given *via.V app runtime. It is useful to integrate popular JS/CSS UI libraries or tools.
|
||||||
@@ -23,9 +25,12 @@ type Options struct {
|
|||||||
// The http server address. e.g. ':3000'
|
// The http server address. e.g. ':3000'
|
||||||
ServerAddress string
|
ServerAddress string
|
||||||
|
|
||||||
// Level of the logs to write to stdout.
|
// LogLevel sets the minimum log level. nil keeps the default (Info).
|
||||||
// Options: Error, Warn, Info, Debug.
|
LogLevel *zerolog.Level
|
||||||
LogLvl LogLevel
|
|
||||||
|
// Logger overrides the default logger entirely. When set, LogLevel and
|
||||||
|
// DevMode have no effect on logging.
|
||||||
|
Logger *zerolog.Logger
|
||||||
|
|
||||||
// The title of the HTML document.
|
// The title of the HTML document.
|
||||||
DocumentTitle string
|
DocumentTitle string
|
||||||
@@ -45,4 +50,8 @@ type Options struct {
|
|||||||
// DatastarPath is the URL path where the script is served.
|
// DatastarPath is the URL path where the script is served.
|
||||||
// Defaults to "/_datastar.js" if empty.
|
// Defaults to "/_datastar.js" if empty.
|
||||||
DatastarPath string
|
DatastarPath string
|
||||||
|
|
||||||
|
// PubSub enables publish/subscribe messaging. Use vianats.New() for an
|
||||||
|
// embedded NATS backend, or supply any PubSub implementation.
|
||||||
|
PubSub PubSub
|
||||||
}
|
}
|
||||||
|
|||||||
73
context.go
73
context.go
@@ -5,13 +5,12 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
|
||||||
"maps"
|
"maps"
|
||||||
"reflect"
|
"reflect"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Context is the living bridge between Go and the browser.
|
// Context is the living bridge between Go and the browser.
|
||||||
@@ -31,6 +30,9 @@ type Context struct {
|
|||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
ctxDisposedChan chan struct{}
|
ctxDisposedChan chan struct{}
|
||||||
reqCtx context.Context
|
reqCtx context.Context
|
||||||
|
subscriptions []Subscription
|
||||||
|
subsMu sync.Mutex
|
||||||
|
disposeOnce sync.Once
|
||||||
}
|
}
|
||||||
|
|
||||||
// View defines the UI rendered by this context.
|
// View defines the UI rendered by this context.
|
||||||
@@ -349,11 +351,23 @@ func (c *Context) ReplaceURLf(format string, a ...any) {
|
|||||||
c.ReplaceURL(fmt.Sprintf(format, a...))
|
c.ReplaceURL(fmt.Sprintf(format, a...))
|
||||||
}
|
}
|
||||||
|
|
||||||
// stopAllRoutines stops all go routines tied to this Context preventing goroutine leaks.
|
// dispose idempotently tears down this context: unsubscribes all pubsub
|
||||||
|
// subscriptions and closes ctxDisposedChan to stop routines and exit the SSE loop.
|
||||||
|
func (c *Context) dispose() {
|
||||||
|
c.disposeOnce.Do(func() {
|
||||||
|
c.unsubscribeAll()
|
||||||
|
c.stopAllRoutines()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// stopAllRoutines closes ctxDisposedChan, broadcasting to all listening
|
||||||
|
// goroutines (OnIntervalRoutine, SSE loop) that this context is done.
|
||||||
func (c *Context) stopAllRoutines() {
|
func (c *Context) stopAllRoutines() {
|
||||||
select {
|
select {
|
||||||
case c.ctxDisposedChan <- struct{}{}:
|
case <-c.ctxDisposedChan:
|
||||||
|
// already closed
|
||||||
default:
|
default:
|
||||||
|
close(c.ctxDisposedChan)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -403,9 +417,58 @@ func (c *Context) Session() *Session {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Publish sends data to the given subject via the configured PubSub backend.
|
||||||
|
// Returns an error if no PubSub is configured. No-ops during panic-check init.
|
||||||
|
func (c *Context) Publish(subject string, data []byte) error {
|
||||||
|
if c.id == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if c.app.pubsub == nil {
|
||||||
|
return fmt.Errorf("pubsub not configured")
|
||||||
|
}
|
||||||
|
return c.app.pubsub.Publish(subject, data)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Subscribe creates a subscription on the configured PubSub backend.
|
||||||
|
// The subscription is tracked for automatic cleanup when the context is disposed.
|
||||||
|
// Returns an error if no PubSub is configured. No-ops during panic-check init.
|
||||||
|
func (c *Context) Subscribe(subject string, handler func(data []byte)) (Subscription, error) {
|
||||||
|
if c.id == "" {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if c.app.pubsub == nil {
|
||||||
|
return nil, fmt.Errorf("pubsub not configured")
|
||||||
|
}
|
||||||
|
sub, err := c.app.pubsub.Subscribe(subject, handler)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Track on page context for cleanup (components use parent, like signals/actions)
|
||||||
|
target := c
|
||||||
|
if c.isComponent() {
|
||||||
|
target = c.parentPageCtx
|
||||||
|
}
|
||||||
|
target.subsMu.Lock()
|
||||||
|
target.subscriptions = append(target.subscriptions, sub)
|
||||||
|
target.subsMu.Unlock()
|
||||||
|
return sub, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// unsubscribeAll cleans up all tracked subscriptions for this context and its components.
|
||||||
|
func (c *Context) unsubscribeAll() {
|
||||||
|
c.subsMu.Lock()
|
||||||
|
subs := c.subscriptions
|
||||||
|
c.subscriptions = nil
|
||||||
|
c.subsMu.Unlock()
|
||||||
|
for _, sub := range subs {
|
||||||
|
sub.Unsubscribe()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func newContext(id string, route string, v *V) *Context {
|
func newContext(id string, route string, v *V) *Context {
|
||||||
if v == nil {
|
if v == nil {
|
||||||
log.Fatal("create context failed: app pointer is nil")
|
panic("create context failed: app pointer is nil")
|
||||||
}
|
}
|
||||||
|
|
||||||
return &Context{
|
return &Context{
|
||||||
|
|||||||
33
go.mod
33
go.mod
@@ -1,4 +1,4 @@
|
|||||||
module github.com/go-via/via
|
module github.com/ryanhamamura/via
|
||||||
|
|
||||||
go 1.25.4
|
go 1.25.4
|
||||||
|
|
||||||
@@ -6,20 +6,37 @@ require maragu.dev/gomponents v1.2.0
|
|||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/DATA-DOG/go-sqlmock v1.5.2
|
github.com/DATA-DOG/go-sqlmock v1.5.2
|
||||||
|
github.com/alexedwards/scs/sqlite3store v0.0.0-20251002162104-209de6e426de
|
||||||
|
github.com/alexedwards/scs/v2 v2.9.0
|
||||||
|
github.com/delaneyj/toolbelt v0.9.1
|
||||||
github.com/mattn/go-sqlite3 v1.14.32
|
github.com/mattn/go-sqlite3 v1.14.32
|
||||||
|
github.com/nats-io/nats.go v1.48.0
|
||||||
|
github.com/rs/zerolog v1.34.0
|
||||||
github.com/starfederation/datastar-go v1.0.3
|
github.com/starfederation/datastar-go v1.0.3
|
||||||
github.com/stretchr/testify v1.10.0
|
github.com/stretchr/testify v1.11.1
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/CAFxX/httpcompression v0.0.9 // indirect
|
github.com/CAFxX/httpcompression v0.0.9 // indirect
|
||||||
github.com/alexedwards/scs/v2 v2.9.0 // indirect
|
|
||||||
github.com/andybalholm/brotli v1.2.0 // indirect
|
github.com/andybalholm/brotli v1.2.0 // indirect
|
||||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
github.com/antithesishq/antithesis-sdk-go v0.5.0 // indirect
|
||||||
github.com/klauspost/compress v1.18.0 // indirect
|
github.com/cenkalti/backoff v2.2.1+incompatible // indirect
|
||||||
github.com/kr/pretty v0.1.0 // indirect
|
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
github.com/google/go-tpm v0.9.7 // indirect
|
||||||
|
github.com/klauspost/compress v1.18.2 // indirect
|
||||||
|
github.com/kr/text v0.2.0 // indirect
|
||||||
|
github.com/mattn/go-colorable v0.1.13 // indirect
|
||||||
|
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||||
|
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76 // indirect
|
||||||
|
github.com/nats-io/jwt/v2 v2.8.0 // indirect
|
||||||
|
github.com/nats-io/nats-server/v2 v2.12.2 // indirect
|
||||||
|
github.com/nats-io/nkeys v0.4.12 // indirect
|
||||||
|
github.com/nats-io/nuid v1.0.1 // indirect
|
||||||
|
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
|
||||||
|
github.com/rogpeppe/go-internal v1.14.1 // indirect
|
||||||
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
||||||
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
|
golang.org/x/crypto v0.45.0 // indirect
|
||||||
|
golang.org/x/sys v0.38.0 // indirect
|
||||||
|
golang.org/x/time v0.14.0 // indirect
|
||||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||||
)
|
)
|
||||||
|
|||||||
75
go.sum
75
go.sum
@@ -2,32 +2,71 @@ github.com/CAFxX/httpcompression v0.0.9 h1:0ue2X8dOLEpxTm8tt+OdHcgA+gbDge0OqFQWG
|
|||||||
github.com/CAFxX/httpcompression v0.0.9/go.mod h1:XX8oPZA+4IDcfZ0A71Hz0mZsv/YJOgYygkFhizVPilM=
|
github.com/CAFxX/httpcompression v0.0.9/go.mod h1:XX8oPZA+4IDcfZ0A71Hz0mZsv/YJOgYygkFhizVPilM=
|
||||||
github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
|
github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
|
||||||
github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
|
github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
|
||||||
|
github.com/alexedwards/scs/sqlite3store v0.0.0-20251002162104-209de6e426de h1:c72K9HLu6K442et0j3BUL/9HEYaUJouLkkVANdmqTOo=
|
||||||
|
github.com/alexedwards/scs/sqlite3store v0.0.0-20251002162104-209de6e426de/go.mod h1:Iyk7S76cxGaiEX/mSYmTZzYehp4KfyylcLaV3OnToss=
|
||||||
github.com/alexedwards/scs/v2 v2.9.0 h1:xa05mVpwTBm1iLeTMNFfAWpKUm4fXAW7CeAViqBVS90=
|
github.com/alexedwards/scs/v2 v2.9.0 h1:xa05mVpwTBm1iLeTMNFfAWpKUm4fXAW7CeAViqBVS90=
|
||||||
github.com/alexedwards/scs/v2 v2.9.0/go.mod h1:ToaROZxyKukJKT/xLcVQAChi5k6+Pn1Gvmdl7h3RRj8=
|
github.com/alexedwards/scs/v2 v2.9.0/go.mod h1:ToaROZxyKukJKT/xLcVQAChi5k6+Pn1Gvmdl7h3RRj8=
|
||||||
github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
|
github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
|
||||||
github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
|
github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
|
||||||
github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
|
github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
|
||||||
|
github.com/antithesishq/antithesis-sdk-go v0.5.0 h1:cudCFF83pDDANcXFzkQPUHHedfnnIbUO3JMr9fqwFJs=
|
||||||
|
github.com/antithesishq/antithesis-sdk-go v0.5.0/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E=
|
||||||
|
github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
|
||||||
|
github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
|
||||||
|
github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
|
||||||
|
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
|
||||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
|
||||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||||
|
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
|
||||||
|
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||||
|
github.com/delaneyj/toolbelt v0.9.1 h1:QJComn2qoaQ4azl5uRkGpdHSO9e+JtoxDTXCiQHvH8o=
|
||||||
|
github.com/delaneyj/toolbelt v0.9.1/go.mod h1:eNXpPuThjTD4tpRNCBl4JEz9jdg9LpyzNuyG+stnIbs=
|
||||||
|
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
|
||||||
github.com/google/brotli/go/cbrotli v0.0.0-20230829110029-ed738e842d2f h1:jopqB+UTSdJGEJT8tEqYyE29zN91fi2827oLET8tl7k=
|
github.com/google/brotli/go/cbrotli v0.0.0-20230829110029-ed738e842d2f h1:jopqB+UTSdJGEJT8tEqYyE29zN91fi2827oLET8tl7k=
|
||||||
github.com/google/brotli/go/cbrotli v0.0.0-20230829110029-ed738e842d2f/go.mod h1:nOPhAkwVliJdNTkj3gXpljmWhjc4wCaVqbMJcPKWP4s=
|
github.com/google/brotli/go/cbrotli v0.0.0-20230829110029-ed738e842d2f/go.mod h1:nOPhAkwVliJdNTkj3gXpljmWhjc4wCaVqbMJcPKWP4s=
|
||||||
|
github.com/google/go-tpm v0.9.7 h1:u89J4tUUeDTlH8xxC3CTW7OHZjbjKoHdQ9W7gCUhtxA=
|
||||||
|
github.com/google/go-tpm v0.9.7/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||||
github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=
|
github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=
|
||||||
github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
|
github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
|
||||||
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
|
github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk=
|
||||||
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
|
github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
|
||||||
github.com/klauspost/pgzip v1.2.6/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
|
github.com/klauspost/pgzip v1.2.6/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
|
||||||
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
|
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
||||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||||
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
|
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
|
||||||
|
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
|
||||||
|
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
|
||||||
|
github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||||
|
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
|
||||||
|
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||||
|
github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
|
||||||
github.com/mattn/go-sqlite3 v1.14.32 h1:JD12Ag3oLy1zQA+BNn74xRgaBbdhbNIDYvQUEuuErjs=
|
github.com/mattn/go-sqlite3 v1.14.32 h1:JD12Ag3oLy1zQA+BNn74xRgaBbdhbNIDYvQUEuuErjs=
|
||||||
github.com/mattn/go-sqlite3 v1.14.32/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
|
github.com/mattn/go-sqlite3 v1.14.32/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
|
||||||
|
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76 h1:KGuD/pM2JpL9FAYvBrnBBeENKZNh6eNtjqytV6TYjnk=
|
||||||
|
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||||
|
github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
|
||||||
|
github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
|
||||||
|
github.com/nats-io/nats-server/v2 v2.12.2 h1:4TEQd0Y4zvcW0IsVxjlXnRso1hBkQl3TS0BI+SxgPhE=
|
||||||
|
github.com/nats-io/nats-server/v2 v2.12.2/go.mod h1:j1AAttYeu7WnvD8HLJ+WWKNMSyxsqmZ160pNtCQRMyE=
|
||||||
|
github.com/nats-io/nats.go v1.48.0 h1:pSFyXApG+yWU/TgbKCjmm5K4wrHu86231/w84qRVR+U=
|
||||||
|
github.com/nats-io/nats.go v1.48.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
|
||||||
|
github.com/nats-io/nkeys v0.4.12 h1:nssm7JKOG9/x4J8II47VWCL1Ds29avyiQDRn0ckMvDc=
|
||||||
|
github.com/nats-io/nkeys v0.4.12/go.mod h1:MT59A1HYcjIcyQDJStTfaOY6vhy9XTUjOFo+SVsvpBg=
|
||||||
|
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||||
|
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||||
github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
|
github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
|
||||||
github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
|
github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
|
||||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||||
|
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||||
|
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||||
|
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
|
||||||
|
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
|
||||||
|
github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
|
||||||
|
github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY=
|
||||||
|
github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ=
|
||||||
github.com/starfederation/datastar-go v1.0.3 h1:DnzgsJ6tDHDM6y5Nxsk0AGW/m8SyKch2vQg3P1xGTcU=
|
github.com/starfederation/datastar-go v1.0.3 h1:DnzgsJ6tDHDM6y5Nxsk0AGW/m8SyKch2vQg3P1xGTcU=
|
||||||
github.com/starfederation/datastar-go v1.0.3/go.mod h1:stm83LQkhZkwa5GzzdPEN6dLuu8FVwxIv0w1DYkbD3w=
|
github.com/starfederation/datastar-go v1.0.3/go.mod h1:stm83LQkhZkwa5GzzdPEN6dLuu8FVwxIv0w1DYkbD3w=
|
||||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||||
@@ -36,8 +75,8 @@ github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpE
|
|||||||
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||||
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
|
||||||
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
||||||
github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
|
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||||
github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
|
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||||
github.com/ulikunitz/xz v0.5.11/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14=
|
github.com/ulikunitz/xz v0.5.11/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14=
|
||||||
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
|
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
|
||||||
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
|
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
|
||||||
@@ -45,9 +84,19 @@ github.com/valyala/gozstd v1.20.1 h1:xPnnnvjmaDDitMFfDxmQ4vpx0+3CdTg2o3lALvXTU/g
|
|||||||
github.com/valyala/gozstd v1.20.1/go.mod h1:y5Ew47GLlP37EkTB+B4s7r6A5rdaeB7ftbl9zoYiIPQ=
|
github.com/valyala/gozstd v1.20.1/go.mod h1:y5Ew47GLlP37EkTB+B4s7r6A5rdaeB7ftbl9zoYiIPQ=
|
||||||
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
|
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
|
||||||
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
|
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
|
||||||
|
golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
|
||||||
|
golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
|
||||||
|
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||||
|
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||||
|
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||||
|
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||||
|
golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
|
||||||
|
golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||||
|
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
|
||||||
|
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
|
||||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||||
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
|
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
|
||||||
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
|
||||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
|
|||||||
@@ -3,8 +3,8 @@ package main
|
|||||||
import (
|
import (
|
||||||
"math/rand"
|
"math/rand"
|
||||||
|
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
@@ -22,7 +22,7 @@ func main() {
|
|||||||
v.Config(via.Options{
|
v.Config(via.Options{
|
||||||
DevMode: true,
|
DevMode: true,
|
||||||
DocumentTitle: "ViaChat",
|
DocumentTitle: "ViaChat",
|
||||||
LogLvl: via.LogLevelInfo,
|
LogLevel: via.LogLevelInfo,
|
||||||
})
|
})
|
||||||
|
|
||||||
v.AppendToHead(
|
v.AppendToHead(
|
||||||
|
|||||||
@@ -1,8 +1,8 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Counter struct{ Count int }
|
type Counter struct{ Count int }
|
||||||
|
|||||||
@@ -1,8 +1,8 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -1,8 +1,8 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
// "github.com/go-via/via-plugin-picocss/picocss"
|
// "github.com/go-via/via-plugin-picocss/picocss"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Counter struct{ Count int }
|
type Counter struct{ Count int }
|
||||||
@@ -14,7 +14,7 @@ func main() {
|
|||||||
v.Config(via.Options{
|
v.Config(via.Options{
|
||||||
DocumentTitle: "Live Reload Demo",
|
DocumentTitle: "Live Reload Demo",
|
||||||
DevMode: true,
|
DevMode: true,
|
||||||
LogLvl: via.LogLevelDebug,
|
LogLevel: via.LogLevelDebug,
|
||||||
Plugins: []via.Plugin{
|
Plugins: []via.Plugin{
|
||||||
// picocss.Default
|
// picocss.Default
|
||||||
},
|
},
|
||||||
|
|||||||
109
internal/examples/nats-chatroom/README.md
Normal file
109
internal/examples/nats-chatroom/README.md
Normal file
@@ -0,0 +1,109 @@
|
|||||||
|
# NATS Chatroom Example (Embedded)
|
||||||
|
|
||||||
|
A chatroom built with Via and an **embedded NATS server**, demonstrating pub/sub messaging as an alternative to the custom `Rooms` implementation in `../chatroom`.
|
||||||
|
|
||||||
|
Uses `delaneyj/toolbelt/embeddednats` to run NATS inside the same binary - no external server required.
|
||||||
|
|
||||||
|
## Key Differences from Original Chatroom
|
||||||
|
|
||||||
|
| Aspect | Original (`../chatroom`) | This Example |
|
||||||
|
|--------|-------------------------|--------------|
|
||||||
|
| Pub/sub | Custom `Rooms` struct (~160 lines) | NATS subjects |
|
||||||
|
| Member tracking | Manual `map[TU]Syncable` | NATS handles subscribers |
|
||||||
|
| Publish timing | Ticker every 100ms + dirty flag | Instant delivery |
|
||||||
|
| Durability | None (in-memory) | JetStream persists to disk |
|
||||||
|
| Multi-instance | Not supported | Works across server instances |
|
||||||
|
| External deps | None | **None** (NATS embedded in binary) |
|
||||||
|
|
||||||
|
## Run the Example
|
||||||
|
|
||||||
|
```bash
|
||||||
|
go run ./internal/examples/nats-chatroom
|
||||||
|
```
|
||||||
|
|
||||||
|
That's it. No separate NATS server needed.
|
||||||
|
|
||||||
|
Open multiple browser tabs at http://localhost:7331 to see messages broadcast across all clients.
|
||||||
|
|
||||||
|
## How Embedded NATS Works
|
||||||
|
|
||||||
|
```go
|
||||||
|
// Start embedded NATS server (JetStream enabled by default)
|
||||||
|
ns, err := embeddednats.New(ctx,
|
||||||
|
embeddednats.WithDirectory("./data/nats"),
|
||||||
|
)
|
||||||
|
ns.WaitForServer()
|
||||||
|
|
||||||
|
// Get client connection to embedded server
|
||||||
|
nc, err := ns.Client()
|
||||||
|
```
|
||||||
|
|
||||||
|
Data is persisted to `./data/nats/` for JetStream durability.
|
||||||
|
|
||||||
|
## Architecture
|
||||||
|
|
||||||
|
```
|
||||||
|
┌─────────────────────────────────────────────────────────┐
|
||||||
|
│ Single Binary │
|
||||||
|
│ │
|
||||||
|
│ Browser A Embedded NATS Browser B │
|
||||||
|
│ │ │ │ │
|
||||||
|
│ │-- Via Action ---> │ │ │
|
||||||
|
│ │ (Send msg) │ │ │
|
||||||
|
│ │ │ │ │
|
||||||
|
│ │ nc.Publish() │ │
|
||||||
|
│ │ "chat.room.Go" │ │
|
||||||
|
│ │ │ │ │
|
||||||
|
│ │<-- Subscribe -----|---- Subscribe --->│ │
|
||||||
|
│ │ callback │ callback │ │
|
||||||
|
│ │ │ │ │
|
||||||
|
│ │-- c.Sync() ------>│<--- c.Sync() -----| │
|
||||||
|
│ │ (SSE) │ (SSE) │ │
|
||||||
|
│ │
|
||||||
|
└─────────────────────────────────────────────────────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
## JetStream Durability
|
||||||
|
|
||||||
|
Messages persist to disk via JetStream:
|
||||||
|
|
||||||
|
```go
|
||||||
|
js.AddStream(&nats.StreamConfig{
|
||||||
|
Name: "CHAT",
|
||||||
|
Subjects: []string{"chat.>"},
|
||||||
|
MaxMsgs: 1000, // Keep last 1000 messages
|
||||||
|
MaxAge: 24 * time.Hour,
|
||||||
|
})
|
||||||
|
```
|
||||||
|
|
||||||
|
Stop and restart the app - chat history survives.
|
||||||
|
|
||||||
|
## Code Comparison
|
||||||
|
|
||||||
|
**Original chatroom - 160+ lines of custom pub/sub:**
|
||||||
|
- `Rooms` struct with named rooms
|
||||||
|
- `Room` with member tracking, mutex, dirty flag
|
||||||
|
- Ticker-based publish loop
|
||||||
|
- Manual join/leave channels
|
||||||
|
|
||||||
|
**This example - ~60 lines of NATS integration:**
|
||||||
|
- `embeddednats.New()` starts the server
|
||||||
|
- `nc.Subscribe(subject, handler)` for receiving
|
||||||
|
- `nc.Publish(subject, data)` for sending
|
||||||
|
- NATS handles delivery, no polling
|
||||||
|
|
||||||
|
## Next Steps
|
||||||
|
|
||||||
|
If this pattern proves useful, it could be promoted to a Via plugin:
|
||||||
|
|
||||||
|
```go
|
||||||
|
// Hypothetical future API
|
||||||
|
v.Config(via.WithEmbeddedNATS("./data/nats"))
|
||||||
|
|
||||||
|
// In page init
|
||||||
|
c.Subscribe("events.user.*", func(data []byte) {
|
||||||
|
c.Sync()
|
||||||
|
})
|
||||||
|
|
||||||
|
c.Publish("events.user.login", userData)
|
||||||
|
```
|
||||||
303
internal/examples/nats-chatroom/main.go
Normal file
303
internal/examples/nats-chatroom/main.go
Normal file
@@ -0,0 +1,303 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"log"
|
||||||
|
"math/rand"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/nats-io/nats.go"
|
||||||
|
"github.com/ryanhamamura/via"
|
||||||
|
"github.com/ryanhamamura/via/h"
|
||||||
|
"github.com/ryanhamamura/via/vianats"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
WithSignal = via.WithSignal
|
||||||
|
)
|
||||||
|
|
||||||
|
// ChatMessage represents a message in a chat room
|
||||||
|
type ChatMessage struct {
|
||||||
|
User UserInfo `json:"user"`
|
||||||
|
Message string `json:"message"`
|
||||||
|
Time int64 `json:"time"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// UserInfo identifies a chat participant
|
||||||
|
type UserInfo struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
Emoji string `json:"emoji"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (u *UserInfo) Avatar() h.H {
|
||||||
|
return h.Div(h.Class("avatar"), h.Attr("title", u.Name), h.Text(u.Emoji))
|
||||||
|
}
|
||||||
|
|
||||||
|
var roomNames = []string{"Go", "Rust", "Python", "JavaScript", "Clojure"}
|
||||||
|
|
||||||
|
func main() {
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
ps, err := vianats.New(ctx, "./data/nats")
|
||||||
|
if err != nil {
|
||||||
|
log.Fatalf("Failed to start embedded NATS: %v", err)
|
||||||
|
}
|
||||||
|
defer ps.Close()
|
||||||
|
|
||||||
|
// Create JetStream stream for message durability
|
||||||
|
js := ps.JetStream()
|
||||||
|
js.AddStream(&nats.StreamConfig{
|
||||||
|
Name: "CHAT",
|
||||||
|
Subjects: []string{"chat.>"},
|
||||||
|
Retention: nats.LimitsPolicy,
|
||||||
|
MaxMsgs: 1000,
|
||||||
|
MaxAge: 24 * time.Hour,
|
||||||
|
})
|
||||||
|
|
||||||
|
v := via.New()
|
||||||
|
v.Config(via.Options{
|
||||||
|
DevMode: true,
|
||||||
|
DocumentTitle: "NATS Chat",
|
||||||
|
LogLevel: via.LogLevelInfo,
|
||||||
|
ServerAddress: ":7331",
|
||||||
|
PubSub: ps,
|
||||||
|
})
|
||||||
|
|
||||||
|
v.AppendToHead(
|
||||||
|
h.Link(h.Rel("stylesheet"), h.Href("https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css")),
|
||||||
|
h.StyleEl(h.Raw(`
|
||||||
|
body { margin: 0; }
|
||||||
|
main {
|
||||||
|
display: flex;
|
||||||
|
flex-direction: column;
|
||||||
|
height: 100vh;
|
||||||
|
}
|
||||||
|
nav[role="tab-control"] ul li a[aria-current="page"] {
|
||||||
|
background-color: var(--pico-primary-background);
|
||||||
|
color: var(--pico-primary-inverse);
|
||||||
|
border-bottom: 2px solid var(--pico-primary);
|
||||||
|
}
|
||||||
|
.chat-message { display: flex; gap: 0.75rem; margin-bottom: 0.5rem; }
|
||||||
|
.avatar {
|
||||||
|
width: 2rem;
|
||||||
|
height: 2rem;
|
||||||
|
border-radius: 50%;
|
||||||
|
background: var(--pico-muted-border-color);
|
||||||
|
display: grid;
|
||||||
|
place-items: center;
|
||||||
|
font-size: 1.5rem;
|
||||||
|
flex-shrink: 0;
|
||||||
|
}
|
||||||
|
.bubble { flex: 1; }
|
||||||
|
.bubble p { margin: 0; }
|
||||||
|
.chat-history {
|
||||||
|
flex: 1;
|
||||||
|
overflow-y: auto;
|
||||||
|
padding: 1rem;
|
||||||
|
padding-bottom: calc(88px + env(safe-area-inset-bottom));
|
||||||
|
}
|
||||||
|
.chat-input {
|
||||||
|
position: fixed;
|
||||||
|
left: 0;
|
||||||
|
right: 0;
|
||||||
|
bottom: 0;
|
||||||
|
background: var(--pico-background-color);
|
||||||
|
display: flex;
|
||||||
|
align-items: center;
|
||||||
|
gap: 0.75rem;
|
||||||
|
padding: 0.75rem 1rem calc(0.75rem + env(safe-area-inset-bottom));
|
||||||
|
border-top: 1px solid var(--pico-muted-border-color);
|
||||||
|
}
|
||||||
|
.chat-input fieldset {
|
||||||
|
flex: 1;
|
||||||
|
margin: 0;
|
||||||
|
}
|
||||||
|
.nats-badge {
|
||||||
|
background: #27AAE1;
|
||||||
|
color: white;
|
||||||
|
padding: 0.25rem 0.5rem;
|
||||||
|
border-radius: 4px;
|
||||||
|
font-size: 0.75rem;
|
||||||
|
margin-left: auto;
|
||||||
|
}
|
||||||
|
`)),
|
||||||
|
h.Script(h.Raw(`
|
||||||
|
function scrollChatToBottom() {
|
||||||
|
const chatHistory = document.querySelector('.chat-history');
|
||||||
|
if (chatHistory) chatHistory.scrollTop = chatHistory.scrollHeight;
|
||||||
|
}
|
||||||
|
`)),
|
||||||
|
)
|
||||||
|
|
||||||
|
v.Page("/", func(c *via.Context) {
|
||||||
|
currentUser := randUser()
|
||||||
|
roomSignal := c.Signal("Go")
|
||||||
|
statement := c.Signal("")
|
||||||
|
|
||||||
|
var messages []ChatMessage
|
||||||
|
var messagesMu sync.Mutex
|
||||||
|
currentRoom := "Go"
|
||||||
|
|
||||||
|
var currentSub via.Subscription
|
||||||
|
|
||||||
|
subscribeToRoom := func(room string) {
|
||||||
|
if currentSub != nil {
|
||||||
|
currentSub.Unsubscribe()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Replay history from JetStream before subscribing for real-time
|
||||||
|
subject := "chat.room." + room
|
||||||
|
if hist, err := js.SubscribeSync(subject, nats.DeliverAll(), nats.OrderedConsumer()); err == nil {
|
||||||
|
for {
|
||||||
|
msg, err := hist.NextMsg(200 * time.Millisecond)
|
||||||
|
if err != nil {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
var chatMsg ChatMessage
|
||||||
|
if json.Unmarshal(msg.Data, &chatMsg) == nil {
|
||||||
|
messages = append(messages, chatMsg)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
hist.Unsubscribe()
|
||||||
|
if len(messages) > 50 {
|
||||||
|
messages = messages[len(messages)-50:]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
sub, _ := c.Subscribe(subject, func(data []byte) {
|
||||||
|
var msg ChatMessage
|
||||||
|
if err := json.Unmarshal(data, &msg); err != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
messagesMu.Lock()
|
||||||
|
messages = append(messages, msg)
|
||||||
|
if len(messages) > 50 {
|
||||||
|
messages = messages[len(messages)-50:]
|
||||||
|
}
|
||||||
|
messagesMu.Unlock()
|
||||||
|
c.Sync()
|
||||||
|
})
|
||||||
|
currentSub = sub
|
||||||
|
currentRoom = room
|
||||||
|
}
|
||||||
|
|
||||||
|
subscribeToRoom("Go")
|
||||||
|
|
||||||
|
switchRoom := c.Action(func() {
|
||||||
|
newRoom := roomSignal.String()
|
||||||
|
if newRoom != currentRoom {
|
||||||
|
messagesMu.Lock()
|
||||||
|
messages = nil
|
||||||
|
messagesMu.Unlock()
|
||||||
|
subscribeToRoom(newRoom)
|
||||||
|
c.Sync()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
say := c.Action(func() {
|
||||||
|
msg := statement.String()
|
||||||
|
if msg == "" {
|
||||||
|
msg = randomDevQuote()
|
||||||
|
}
|
||||||
|
statement.SetValue("")
|
||||||
|
|
||||||
|
data, _ := json.Marshal(ChatMessage{
|
||||||
|
User: currentUser,
|
||||||
|
Message: msg,
|
||||||
|
Time: time.Now().UnixMilli(),
|
||||||
|
})
|
||||||
|
c.Publish("chat.room."+currentRoom, data)
|
||||||
|
})
|
||||||
|
|
||||||
|
c.View(func() h.H {
|
||||||
|
var tabs []h.H
|
||||||
|
for _, name := range roomNames {
|
||||||
|
isCurrent := name == currentRoom
|
||||||
|
tabs = append(tabs, h.Li(
|
||||||
|
h.A(
|
||||||
|
h.If(isCurrent, h.Attr("aria-current", "page")),
|
||||||
|
h.Text(name),
|
||||||
|
switchRoom.OnClick(WithSignal(roomSignal, name)),
|
||||||
|
),
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
messagesMu.Lock()
|
||||||
|
chatHistoryChildren := []h.H{
|
||||||
|
h.Class("chat-history"),
|
||||||
|
h.Script(h.Raw(`new MutationObserver(()=>scrollChatToBottom()).observe(document.querySelector('.chat-history'), {childList:true})`)),
|
||||||
|
}
|
||||||
|
for _, msg := range messages {
|
||||||
|
chatHistoryChildren = append(chatHistoryChildren,
|
||||||
|
h.Div(h.Class("chat-message"),
|
||||||
|
h.Div(h.Class("avatar"), h.Attr("title", msg.User.Name), h.Text(msg.User.Emoji)),
|
||||||
|
h.Div(h.Class("bubble"),
|
||||||
|
h.P(h.Text(msg.Message)),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
messagesMu.Unlock()
|
||||||
|
|
||||||
|
return h.Main(h.Class("container"),
|
||||||
|
h.Nav(
|
||||||
|
h.Attr("role", "tab-control"),
|
||||||
|
h.Ul(tabs...),
|
||||||
|
h.Span(h.Class("nats-badge"), h.Text("NATS")),
|
||||||
|
),
|
||||||
|
h.Div(chatHistoryChildren...),
|
||||||
|
h.Div(
|
||||||
|
h.Class("chat-input"),
|
||||||
|
currentUser.Avatar(),
|
||||||
|
h.FieldSet(
|
||||||
|
h.Attr("role", "group"),
|
||||||
|
h.Input(
|
||||||
|
h.Type("text"),
|
||||||
|
h.Placeholder(currentUser.Name+" says..."),
|
||||||
|
statement.Bind(),
|
||||||
|
h.Attr("autofocus"),
|
||||||
|
say.OnKeyDown("Enter"),
|
||||||
|
),
|
||||||
|
h.Button(h.Text("Send"), say.OnClick()),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
|
log.Println("Starting NATS chatroom on :7331 (embedded NATS server)")
|
||||||
|
v.Start()
|
||||||
|
}
|
||||||
|
|
||||||
|
func randUser() UserInfo {
|
||||||
|
adjectives := []string{"Happy", "Clever", "Brave", "Swift", "Gentle", "Wise", "Bold", "Calm", "Eager", "Fierce"}
|
||||||
|
animals := []string{"Panda", "Tiger", "Eagle", "Dolphin", "Fox", "Wolf", "Bear", "Hawk", "Otter", "Lion"}
|
||||||
|
emojis := []string{"🐼", "🐯", "🦅", "🐬", "🦊", "🐺", "🐻", "🦅", "🦦", "🦁"}
|
||||||
|
|
||||||
|
idx := rand.Intn(len(animals))
|
||||||
|
return UserInfo{
|
||||||
|
Name: adjectives[rand.Intn(len(adjectives))] + " " + animals[idx],
|
||||||
|
Emoji: emojis[idx],
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var quoteIdx = rand.Intn(len(devQuotes))
|
||||||
|
var devQuotes = []string{
|
||||||
|
"Just use NATS.",
|
||||||
|
"Pub/sub all the things!",
|
||||||
|
"Messages are the new API.",
|
||||||
|
"JetStream for durability.",
|
||||||
|
"No more polling.",
|
||||||
|
"Event-driven architecture FTW.",
|
||||||
|
"Decouple everything.",
|
||||||
|
"NATS is fast.",
|
||||||
|
"Subjects are like topics.",
|
||||||
|
"Request-reply is cool.",
|
||||||
|
}
|
||||||
|
|
||||||
|
func randomDevQuote() string {
|
||||||
|
quoteIdx = (quoteIdx + 1) % len(devQuotes)
|
||||||
|
return devQuotes[quoteIdx]
|
||||||
|
}
|
||||||
@@ -3,9 +3,9 @@ package main
|
|||||||
import (
|
import (
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
// "github.com/go-via/via-plugin-picocss/picocss"
|
// "github.com/go-via/via-plugin-picocss/picocss"
|
||||||
. "github.com/go-via/via/h"
|
. "github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
// "github.com/go-via/via-plugin-picocss/picocss"
|
// "github.com/go-via/via-plugin-picocss/picocss"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
type Counter struct{ Count int }
|
type Counter struct{ Count int }
|
||||||
|
|||||||
@@ -4,8 +4,8 @@ import (
|
|||||||
_ "embed"
|
_ "embed"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Example of a Via application with a plugin that adds PicoCSS. The plugin
|
// Example of a Via application with a plugin that adds PicoCSS. The plugin
|
||||||
|
|||||||
@@ -5,16 +5,16 @@ import (
|
|||||||
"math/rand"
|
"math/rand"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
// "github.com/go-via/via-plugin-picocss/picocss"
|
// "github.com/go-via/via-plugin-picocss/picocss"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
v := via.New()
|
v := via.New()
|
||||||
|
|
||||||
v.Config(via.Options{
|
v.Config(via.Options{
|
||||||
LogLvl: via.LogLevelDebug,
|
LogLevel: via.LogLevelDebug,
|
||||||
DevMode: true,
|
DevMode: true,
|
||||||
Plugins: []via.Plugin{
|
Plugins: []via.Plugin{
|
||||||
// picocss.Default,
|
// picocss.Default,
|
||||||
|
|||||||
@@ -1,13 +1,33 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/go-via/via"
|
"database/sql"
|
||||||
"github.com/go-via/via/h"
|
"log"
|
||||||
|
|
||||||
|
_ "github.com/mattn/go-sqlite3"
|
||||||
|
"github.com/ryanhamamura/via"
|
||||||
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
// Open SQLite database for persistent sessions
|
||||||
|
db, err := sql.Open("sqlite3", "sessions.db")
|
||||||
|
if err != nil {
|
||||||
|
log.Fatalf("failed to open database: %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
// Create session manager with SQLite store
|
||||||
|
sm, err := via.NewSQLiteSessionManager(db)
|
||||||
|
if err != nil {
|
||||||
|
log.Fatalf("failed to create session manager: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
v := via.New()
|
v := via.New()
|
||||||
v.Config(via.Options{ServerAddress: ":7331"})
|
v.Config(via.Options{
|
||||||
|
ServerAddress: ":7331",
|
||||||
|
SessionManager: sm,
|
||||||
|
})
|
||||||
|
|
||||||
// Login page
|
// Login page
|
||||||
v.Page("/login", func(c *via.Context) {
|
v.Page("/login", func(c *via.Context) {
|
||||||
|
|||||||
@@ -6,8 +6,8 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/go-via/via"
|
"github.com/ryanhamamura/via"
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
_ "github.com/mattn/go-sqlite3"
|
_ "github.com/mattn/go-sqlite3"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -54,7 +54,7 @@ func main() {
|
|||||||
v.Config(via.Options{
|
v.Config(via.Options{
|
||||||
DevMode: true,
|
DevMode: true,
|
||||||
DocumentTitle: "Search",
|
DocumentTitle: "Search",
|
||||||
LogLvl: via.LogLevelWarn,
|
LogLevel: via.LogLevelWarn,
|
||||||
})
|
})
|
||||||
|
|
||||||
v.AppendToHead(
|
v.AppendToHead(
|
||||||
|
|||||||
195
nats_test.go
Normal file
195
nats_test.go
Normal file
@@ -0,0 +1,195 @@
|
|||||||
|
package via
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/ryanhamamura/via/h"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
)
|
||||||
|
|
||||||
|
type mockHandler struct {
|
||||||
|
id int64
|
||||||
|
fn func([]byte)
|
||||||
|
active atomic.Bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// mockPubSub implements PubSub for testing without NATS.
|
||||||
|
type mockPubSub struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
subs map[string][]*mockHandler
|
||||||
|
nextID atomic.Int64
|
||||||
|
}
|
||||||
|
|
||||||
|
func newMockPubSub() *mockPubSub {
|
||||||
|
return &mockPubSub{subs: make(map[string][]*mockHandler)}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockPubSub) Publish(subject string, data []byte) error {
|
||||||
|
m.mu.Lock()
|
||||||
|
handlers := make([]*mockHandler, len(m.subs[subject]))
|
||||||
|
copy(handlers, m.subs[subject])
|
||||||
|
m.mu.Unlock()
|
||||||
|
for _, h := range handlers {
|
||||||
|
if h.active.Load() {
|
||||||
|
h.fn(data)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockPubSub) Subscribe(subject string, handler func(data []byte)) (Subscription, error) {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
mh := &mockHandler{
|
||||||
|
id: m.nextID.Add(1),
|
||||||
|
fn: handler,
|
||||||
|
}
|
||||||
|
mh.active.Store(true)
|
||||||
|
m.subs[subject] = append(m.subs[subject], mh)
|
||||||
|
return &mockSub{handler: mh}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *mockPubSub) Close() error { return nil }
|
||||||
|
|
||||||
|
type mockSub struct {
|
||||||
|
handler *mockHandler
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *mockSub) Unsubscribe() error {
|
||||||
|
s.handler.active.Store(false)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPubSub_RoundTrip(t *testing.T) {
|
||||||
|
ps := newMockPubSub()
|
||||||
|
v := New()
|
||||||
|
v.Config(Options{PubSub: ps})
|
||||||
|
|
||||||
|
var received []byte
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(1)
|
||||||
|
|
||||||
|
c := newContext("test-ctx", "/", v)
|
||||||
|
c.View(func() h.H { return h.Div() })
|
||||||
|
|
||||||
|
_, err := c.Subscribe("test.topic", func(data []byte) {
|
||||||
|
received = data
|
||||||
|
wg.Done()
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
err = c.Publish("test.topic", []byte("hello"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
wg.Wait()
|
||||||
|
assert.Equal(t, []byte("hello"), received)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPubSub_MultipleSubscribers(t *testing.T) {
|
||||||
|
ps := newMockPubSub()
|
||||||
|
v := New()
|
||||||
|
v.Config(Options{PubSub: ps})
|
||||||
|
|
||||||
|
var mu sync.Mutex
|
||||||
|
var results []string
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(2)
|
||||||
|
|
||||||
|
c1 := newContext("ctx-1", "/", v)
|
||||||
|
c1.View(func() h.H { return h.Div() })
|
||||||
|
c2 := newContext("ctx-2", "/", v)
|
||||||
|
c2.View(func() h.H { return h.Div() })
|
||||||
|
|
||||||
|
c1.Subscribe("broadcast", func(data []byte) {
|
||||||
|
mu.Lock()
|
||||||
|
results = append(results, "c1:"+string(data))
|
||||||
|
mu.Unlock()
|
||||||
|
wg.Done()
|
||||||
|
})
|
||||||
|
|
||||||
|
c2.Subscribe("broadcast", func(data []byte) {
|
||||||
|
mu.Lock()
|
||||||
|
results = append(results, "c2:"+string(data))
|
||||||
|
mu.Unlock()
|
||||||
|
wg.Done()
|
||||||
|
})
|
||||||
|
|
||||||
|
c1.Publish("broadcast", []byte("msg"))
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
assert.Len(t, results, 2)
|
||||||
|
assert.Contains(t, results, "c1:msg")
|
||||||
|
assert.Contains(t, results, "c2:msg")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPubSub_SubscriptionCleanupOnDispose(t *testing.T) {
|
||||||
|
ps := newMockPubSub()
|
||||||
|
v := New()
|
||||||
|
v.Config(Options{PubSub: ps})
|
||||||
|
|
||||||
|
c := newContext("cleanup-ctx", "/", v)
|
||||||
|
c.View(func() h.H { return h.Div() })
|
||||||
|
|
||||||
|
c.Subscribe("room.1", func(data []byte) {})
|
||||||
|
c.Subscribe("room.2", func(data []byte) {})
|
||||||
|
|
||||||
|
assert.Len(t, c.subscriptions, 2)
|
||||||
|
|
||||||
|
c.unsubscribeAll()
|
||||||
|
assert.Empty(t, c.subscriptions)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPubSub_ManualUnsubscribe(t *testing.T) {
|
||||||
|
ps := newMockPubSub()
|
||||||
|
v := New()
|
||||||
|
v.Config(Options{PubSub: ps})
|
||||||
|
|
||||||
|
c := newContext("unsub-ctx", "/", v)
|
||||||
|
c.View(func() h.H { return h.Div() })
|
||||||
|
|
||||||
|
called := false
|
||||||
|
sub, err := c.Subscribe("topic", func(data []byte) {
|
||||||
|
called = true
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
sub.Unsubscribe()
|
||||||
|
|
||||||
|
c.Publish("topic", []byte("ignored"))
|
||||||
|
time.Sleep(10 * time.Millisecond)
|
||||||
|
assert.False(t, called)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPubSub_NoOpWhenNotConfigured(t *testing.T) {
|
||||||
|
v := New()
|
||||||
|
|
||||||
|
c := newContext("noop-ctx", "/", v)
|
||||||
|
c.View(func() h.H { return h.Div() })
|
||||||
|
|
||||||
|
err := c.Publish("topic", []byte("data"))
|
||||||
|
assert.Error(t, err)
|
||||||
|
|
||||||
|
sub, err := c.Subscribe("topic", func(data []byte) {})
|
||||||
|
assert.Error(t, err)
|
||||||
|
assert.Nil(t, sub)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPubSub_NoOpDuringPanicCheck(t *testing.T) {
|
||||||
|
ps := newMockPubSub()
|
||||||
|
v := New()
|
||||||
|
v.Config(Options{PubSub: ps})
|
||||||
|
|
||||||
|
// Panic-check context has id=""
|
||||||
|
c := newContext("", "/", v)
|
||||||
|
|
||||||
|
err := c.Publish("topic", []byte("data"))
|
||||||
|
assert.NoError(t, err)
|
||||||
|
|
||||||
|
sub, err := c.Subscribe("topic", func(data []byte) {})
|
||||||
|
assert.NoError(t, err)
|
||||||
|
assert.Nil(t, sub)
|
||||||
|
}
|
||||||
14
pubsub.go
Normal file
14
pubsub.go
Normal file
@@ -0,0 +1,14 @@
|
|||||||
|
package via
|
||||||
|
|
||||||
|
// PubSub is an interface for publish/subscribe messaging backends.
|
||||||
|
// The vianats sub-package provides an embedded NATS implementation.
|
||||||
|
type PubSub interface {
|
||||||
|
Publish(subject string, data []byte) error
|
||||||
|
Subscribe(subject string, handler func(data []byte)) (Subscription, error)
|
||||||
|
Close() error
|
||||||
|
}
|
||||||
|
|
||||||
|
// Subscription represents an active subscription that can be manually unsubscribed.
|
||||||
|
type Subscription interface {
|
||||||
|
Unsubscribe() error
|
||||||
|
}
|
||||||
23
session.go
23
session.go
@@ -2,11 +2,34 @@ package via
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"database/sql"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/alexedwards/scs/sqlite3store"
|
||||||
"github.com/alexedwards/scs/v2"
|
"github.com/alexedwards/scs/v2"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// NewSQLiteSessionManager creates a session manager using SQLite for persistence.
|
||||||
|
// Creates the sessions table if it doesn't exist.
|
||||||
|
// The returned manager can be configured further (Lifetime, Cookie settings, etc.)
|
||||||
|
// before passing to Options.SessionManager.
|
||||||
|
func NewSQLiteSessionManager(db *sql.DB) (*scs.SessionManager, error) {
|
||||||
|
_, err := db.Exec(`
|
||||||
|
CREATE TABLE IF NOT EXISTS sessions (
|
||||||
|
token TEXT PRIMARY KEY,
|
||||||
|
data BLOB NOT NULL,
|
||||||
|
expiry REAL NOT NULL
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS sessions_expiry_idx ON sessions(expiry);
|
||||||
|
`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
sm := scs.New()
|
||||||
|
sm.Store = sqlite3store.New(db)
|
||||||
|
return sm, nil
|
||||||
|
}
|
||||||
|
|
||||||
// Session provides access to the user's session data.
|
// Session provides access to the user's session data.
|
||||||
// Session data persists across page views for the same browser.
|
// Session data persists across page views for the same browser.
|
||||||
type Session struct {
|
type Session struct {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Signal represents a value that is reactive in the browser. Signals
|
// Signal represents a value that is reactive in the browser. Signals
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import (
|
|||||||
// "net/http/httptest"
|
// "net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
170
via.go
170
via.go
@@ -7,22 +7,26 @@
|
|||||||
package via
|
package via
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
_ "embed"
|
_ "embed"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"log"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
"os"
|
"os"
|
||||||
|
ossignal "os/signal"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"syscall"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/alexedwards/scs/v2"
|
"github.com/alexedwards/scs/v2"
|
||||||
"github.com/go-via/via/h"
|
"github.com/rs/zerolog"
|
||||||
|
"github.com/ryanhamamura/via/h"
|
||||||
"github.com/starfederation/datastar-go/datastar"
|
"github.com/starfederation/datastar-go/datastar"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -34,63 +38,66 @@ var datastarJS []byte
|
|||||||
type V struct {
|
type V struct {
|
||||||
cfg Options
|
cfg Options
|
||||||
mux *http.ServeMux
|
mux *http.ServeMux
|
||||||
|
server *http.Server
|
||||||
|
logger zerolog.Logger
|
||||||
contextRegistry map[string]*Context
|
contextRegistry map[string]*Context
|
||||||
contextRegistryMutex sync.RWMutex
|
contextRegistryMutex sync.RWMutex
|
||||||
documentHeadIncludes []h.H
|
documentHeadIncludes []h.H
|
||||||
documentFootIncludes []h.H
|
documentFootIncludes []h.H
|
||||||
devModePageInitFnMap map[string]func(*Context)
|
devModePageInitFnMap map[string]func(*Context)
|
||||||
sessionManager *scs.SessionManager
|
sessionManager *scs.SessionManager
|
||||||
|
pubsub PubSub
|
||||||
datastarPath string
|
datastarPath string
|
||||||
datastarContent []byte
|
datastarContent []byte
|
||||||
datastarOnce sync.Once
|
datastarOnce sync.Once
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (v *V) logEvent(evt *zerolog.Event, c *Context) *zerolog.Event {
|
||||||
|
if c != nil && c.id != "" {
|
||||||
|
evt = evt.Str("via-ctx", c.id)
|
||||||
|
}
|
||||||
|
return evt
|
||||||
|
}
|
||||||
|
|
||||||
func (v *V) logFatal(format string, a ...any) {
|
func (v *V) logFatal(format string, a ...any) {
|
||||||
log.Printf("[fatal] msg=%q", fmt.Sprintf(format, a...))
|
v.logEvent(v.logger.WithLevel(zerolog.FatalLevel), nil).Msgf(format, a...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (v *V) logErr(c *Context, format string, a ...any) {
|
func (v *V) logErr(c *Context, format string, a ...any) {
|
||||||
cRef := ""
|
v.logEvent(v.logger.Error(), c).Msgf(format, a...)
|
||||||
if c != nil && c.id != "" {
|
|
||||||
cRef = fmt.Sprintf("via-ctx=%q ", c.id)
|
|
||||||
}
|
|
||||||
log.Printf("[error] %smsg=%q", cRef, fmt.Sprintf(format, a...))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (v *V) logWarn(c *Context, format string, a ...any) {
|
func (v *V) logWarn(c *Context, format string, a ...any) {
|
||||||
cRef := ""
|
v.logEvent(v.logger.Warn(), c).Msgf(format, a...)
|
||||||
if c != nil && c.id != "" {
|
|
||||||
cRef = fmt.Sprintf("via-ctx=%q ", c.id)
|
|
||||||
}
|
|
||||||
if v.cfg.LogLvl >= LogLevelWarn {
|
|
||||||
log.Printf("[warn] %smsg=%q", cRef, fmt.Sprintf(format, a...))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (v *V) logInfo(c *Context, format string, a ...any) {
|
func (v *V) logInfo(c *Context, format string, a ...any) {
|
||||||
cRef := ""
|
v.logEvent(v.logger.Info(), c).Msgf(format, a...)
|
||||||
if c != nil && c.id != "" {
|
|
||||||
cRef = fmt.Sprintf("via-ctx=%q ", c.id)
|
|
||||||
}
|
|
||||||
if v.cfg.LogLvl >= LogLevelInfo {
|
|
||||||
log.Printf("[info] %smsg=%q", cRef, fmt.Sprintf(format, a...))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (v *V) logDebug(c *Context, format string, a ...any) {
|
func (v *V) logDebug(c *Context, format string, a ...any) {
|
||||||
cRef := ""
|
v.logEvent(v.logger.Debug(), c).Msgf(format, a...)
|
||||||
if c != nil && c.id != "" {
|
}
|
||||||
cRef = fmt.Sprintf("via-ctx=%q ", c.id)
|
|
||||||
}
|
func newConsoleLogger(level zerolog.Level) zerolog.Logger {
|
||||||
if v.cfg.LogLvl == LogLevelDebug {
|
return zerolog.New(zerolog.ConsoleWriter{Out: os.Stderr, TimeFormat: "15:04:05"}).
|
||||||
log.Printf("[debug] %smsg=%q", cRef, fmt.Sprintf(format, a...))
|
With().Timestamp().Logger().Level(level)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Config overrides the default configuration with the given options.
|
// Config overrides the default configuration with the given options.
|
||||||
func (v *V) Config(cfg Options) {
|
func (v *V) Config(cfg Options) {
|
||||||
if cfg.LogLvl != undefined {
|
if cfg.Logger != nil {
|
||||||
v.cfg.LogLvl = cfg.LogLvl
|
v.logger = *cfg.Logger
|
||||||
|
} else if cfg.LogLevel != nil || cfg.DevMode != v.cfg.DevMode {
|
||||||
|
level := zerolog.InfoLevel
|
||||||
|
if cfg.LogLevel != nil {
|
||||||
|
level = *cfg.LogLevel
|
||||||
|
}
|
||||||
|
if cfg.DevMode {
|
||||||
|
v.logger = newConsoleLogger(level)
|
||||||
|
} else {
|
||||||
|
v.logger = zerolog.New(os.Stderr).With().Timestamp().Logger().Level(level)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if cfg.DocumentTitle != "" {
|
if cfg.DocumentTitle != "" {
|
||||||
v.cfg.DocumentTitle = cfg.DocumentTitle
|
v.cfg.DocumentTitle = cfg.DocumentTitle
|
||||||
@@ -117,6 +124,9 @@ func (v *V) Config(cfg Options) {
|
|||||||
if cfg.DatastarPath != "" {
|
if cfg.DatastarPath != "" {
|
||||||
v.datastarPath = cfg.DatastarPath
|
v.datastarPath = cfg.DatastarPath
|
||||||
}
|
}
|
||||||
|
if cfg.PubSub != nil {
|
||||||
|
v.pubsub = cfg.PubSub
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// AppendToHead appends the given h.H nodes to the head of the base HTML document.
|
// AppendToHead appends the given h.H nodes to the head of the base HTML document.
|
||||||
@@ -249,14 +259,82 @@ func (v *V) getCtx(id string) (*Context, error) {
|
|||||||
return nil, fmt.Errorf("ctx '%s' not found", id)
|
return nil, fmt.Errorf("ctx '%s' not found", id)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start starts the Via HTTP server on the given address.
|
// Start starts the Via HTTP server and blocks until a SIGINT or SIGTERM
|
||||||
|
// signal is received, then performs a graceful shutdown.
|
||||||
func (v *V) Start() {
|
func (v *V) Start() {
|
||||||
v.logInfo(nil, "via started at [%s]", v.cfg.ServerAddress)
|
|
||||||
handler := http.Handler(v.mux)
|
handler := http.Handler(v.mux)
|
||||||
if v.sessionManager != nil {
|
if v.sessionManager != nil {
|
||||||
handler = v.sessionManager.LoadAndSave(v.mux)
|
handler = v.sessionManager.LoadAndSave(v.mux)
|
||||||
}
|
}
|
||||||
log.Fatalf("[fatal] %v", http.ListenAndServe(v.cfg.ServerAddress, handler))
|
v.server = &http.Server{
|
||||||
|
Addr: v.cfg.ServerAddress,
|
||||||
|
Handler: handler,
|
||||||
|
}
|
||||||
|
|
||||||
|
errCh := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
errCh <- v.server.ListenAndServe()
|
||||||
|
}()
|
||||||
|
|
||||||
|
v.logInfo(nil, "via started at [%s]", v.cfg.ServerAddress)
|
||||||
|
|
||||||
|
sigCh := make(chan os.Signal, 1)
|
||||||
|
ossignal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||||||
|
|
||||||
|
select {
|
||||||
|
case sig := <-sigCh:
|
||||||
|
v.logInfo(nil, "received signal %v, shutting down", sig)
|
||||||
|
case err := <-errCh:
|
||||||
|
if err != nil && err != http.ErrServerClosed {
|
||||||
|
v.logger.Fatal().Err(err).Msg("http server failed")
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
v.shutdown()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Shutdown gracefully shuts down the server and all contexts.
|
||||||
|
// Safe for programmatic or test use.
|
||||||
|
func (v *V) Shutdown() {
|
||||||
|
v.shutdown()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (v *V) shutdown() {
|
||||||
|
v.logInfo(nil, "draining all contexts")
|
||||||
|
v.drainAllContexts()
|
||||||
|
|
||||||
|
if v.server != nil {
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := v.server.Shutdown(ctx); err != nil {
|
||||||
|
v.logErr(nil, "http server shutdown error: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if v.pubsub != nil {
|
||||||
|
if err := v.pubsub.Close(); err != nil {
|
||||||
|
v.logErr(nil, "pubsub close error: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
v.logInfo(nil, "shutdown complete")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (v *V) drainAllContexts() {
|
||||||
|
v.contextRegistryMutex.Lock()
|
||||||
|
contexts := make([]*Context, 0, len(v.contextRegistry))
|
||||||
|
for _, c := range v.contextRegistry {
|
||||||
|
contexts = append(contexts, c)
|
||||||
|
}
|
||||||
|
v.contextRegistry = make(map[string]*Context)
|
||||||
|
v.contextRegistryMutex.Unlock()
|
||||||
|
|
||||||
|
for _, c := range contexts {
|
||||||
|
v.logDebug(c, "disposing context")
|
||||||
|
c.dispose()
|
||||||
|
}
|
||||||
|
v.logInfo(nil, "drained %d context(s)", len(contexts))
|
||||||
}
|
}
|
||||||
|
|
||||||
// HTTPServeMux returns the underlying HTTP request multiplexer to enable user extentions, middleware and
|
// HTTPServeMux returns the underlying HTTP request multiplexer to enable user extentions, middleware and
|
||||||
@@ -280,7 +358,7 @@ func (v *V) ensureDatastarHandler() {
|
|||||||
func (v *V) devModePersist(c *Context) {
|
func (v *V) devModePersist(c *Context) {
|
||||||
p := filepath.Join(".via", "devmode", "ctx.json")
|
p := filepath.Join(".via", "devmode", "ctx.json")
|
||||||
if err := os.MkdirAll(filepath.Dir(p), 0755); err != nil {
|
if err := os.MkdirAll(filepath.Dir(p), 0755); err != nil {
|
||||||
log.Fatalf("failed to create directory for devmode files: %v", err)
|
v.logFatal("failed to create directory for devmode files: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// load persisted list from file, or empty list if file not found
|
// load persisted list from file, or empty list if file not found
|
||||||
@@ -394,6 +472,7 @@ func New() *V {
|
|||||||
|
|
||||||
v := &V{
|
v := &V{
|
||||||
mux: mux,
|
mux: mux,
|
||||||
|
logger: newConsoleLogger(zerolog.InfoLevel),
|
||||||
contextRegistry: make(map[string]*Context),
|
contextRegistry: make(map[string]*Context),
|
||||||
devModePageInitFnMap: make(map[string]func(*Context)),
|
devModePageInitFnMap: make(map[string]func(*Context)),
|
||||||
sessionManager: scs.New(),
|
sessionManager: scs.New(),
|
||||||
@@ -402,16 +481,11 @@ func New() *V {
|
|||||||
cfg: Options{
|
cfg: Options{
|
||||||
DevMode: false,
|
DevMode: false,
|
||||||
ServerAddress: ":3000",
|
ServerAddress: ":3000",
|
||||||
LogLvl: LogLevelInfo,
|
|
||||||
DocumentTitle: "⚡ Via",
|
DocumentTitle: "⚡ Via",
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
v.mux.HandleFunc("GET /_sse", func(w http.ResponseWriter, r *http.Request) {
|
v.mux.HandleFunc("GET /_sse", func(w http.ResponseWriter, r *http.Request) {
|
||||||
isReconnect := false
|
|
||||||
if r.Header.Get("last-event-id") == "via" {
|
|
||||||
isReconnect = true
|
|
||||||
}
|
|
||||||
var sigs map[string]any
|
var sigs map[string]any
|
||||||
_ = datastar.ReadSignals(r, &sigs)
|
_ = datastar.ReadSignals(r, &sigs)
|
||||||
cID, _ := sigs["via-ctx"].(string)
|
cID, _ := sigs["via-ctx"].(string)
|
||||||
@@ -436,11 +510,7 @@ func New() *V {
|
|||||||
v.logDebug(c, "SSE connection established")
|
v.logDebug(c, "SSE connection established")
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
if isReconnect || v.cfg.DevMode {
|
|
||||||
c.Sync()
|
c.Sync()
|
||||||
return
|
|
||||||
}
|
|
||||||
c.SyncSignals()
|
|
||||||
}()
|
}()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
@@ -448,10 +518,10 @@ func New() *V {
|
|||||||
case <-sse.Context().Done():
|
case <-sse.Context().Done():
|
||||||
v.logDebug(c, "SSE connection ended")
|
v.logDebug(c, "SSE connection ended")
|
||||||
return
|
return
|
||||||
case patch, ok := <-c.patchChan:
|
case <-c.ctxDisposedChan:
|
||||||
if !ok {
|
v.logDebug(c, "context disposed, closing SSE")
|
||||||
continue
|
return
|
||||||
}
|
case patch := <-c.patchChan:
|
||||||
switch patch.typ {
|
switch patch.typ {
|
||||||
case patchTypeElements:
|
case patchTypeElements:
|
||||||
if err := sse.PatchElements(patch.content); err != nil {
|
if err := sse.PatchElements(patch.content); err != nil {
|
||||||
@@ -522,7 +592,7 @@ func New() *V {
|
|||||||
v.mux.HandleFunc("POST /_session/close", func(w http.ResponseWriter, r *http.Request) {
|
v.mux.HandleFunc("POST /_session/close", func(w http.ResponseWriter, r *http.Request) {
|
||||||
body, err := io.ReadAll(r.Body)
|
body, err := io.ReadAll(r.Body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("Error reading body: %v", err)
|
v.logErr(nil, "error reading body: %v", err)
|
||||||
w.WriteHeader(http.StatusBadRequest)
|
w.WriteHeader(http.StatusBadRequest)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -533,7 +603,7 @@ func New() *V {
|
|||||||
v.logErr(c, "failed to handle session close: %v", err)
|
v.logErr(c, "failed to handle session close: %v", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
c.stopAllRoutines()
|
c.dispose()
|
||||||
v.logDebug(c, "session close event triggered")
|
v.logDebug(c, "session close event triggered")
|
||||||
if v.cfg.DevMode {
|
if v.cfg.DevMode {
|
||||||
v.devModeRemovePersisted(c)
|
v.devModeRemovePersisted(c)
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/go-via/via/h"
|
"github.com/ryanhamamura/via/h"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
78
vianats/vianats.go
Normal file
78
vianats/vianats.go
Normal file
@@ -0,0 +1,78 @@
|
|||||||
|
// Package vianats provides an embedded NATS server with JetStream as a
|
||||||
|
// pub/sub backend for Via applications.
|
||||||
|
package vianats
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"github.com/delaneyj/toolbelt/embeddednats"
|
||||||
|
"github.com/nats-io/nats.go"
|
||||||
|
"github.com/ryanhamamura/via"
|
||||||
|
)
|
||||||
|
|
||||||
|
// NATS implements via.PubSub using an embedded NATS server with JetStream.
|
||||||
|
type NATS struct {
|
||||||
|
server *embeddednats.Server
|
||||||
|
nc *nats.Conn
|
||||||
|
js nats.JetStreamContext
|
||||||
|
}
|
||||||
|
|
||||||
|
// New starts an embedded NATS server with JetStream enabled and returns a
|
||||||
|
// ready-to-use NATS instance. The server stores data in dataDir and shuts
|
||||||
|
// down when ctx is cancelled.
|
||||||
|
func New(ctx context.Context, dataDir string) (*NATS, error) {
|
||||||
|
ns, err := embeddednats.New(ctx, embeddednats.WithDirectory(dataDir))
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("vianats: start server: %w", err)
|
||||||
|
}
|
||||||
|
ns.WaitForServer()
|
||||||
|
|
||||||
|
nc, err := ns.Client()
|
||||||
|
if err != nil {
|
||||||
|
ns.Close()
|
||||||
|
return nil, fmt.Errorf("vianats: connect client: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
js, err := nc.JetStream()
|
||||||
|
if err != nil {
|
||||||
|
nc.Close()
|
||||||
|
ns.Close()
|
||||||
|
return nil, fmt.Errorf("vianats: init jetstream: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &NATS{server: ns, nc: nc, js: js}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Publish sends data to the given subject using core NATS publish.
|
||||||
|
// JetStream captures messages automatically if a matching stream exists.
|
||||||
|
func (n *NATS) Publish(subject string, data []byte) error {
|
||||||
|
return n.nc.Publish(subject, data)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Subscribe creates a core NATS subscription for real-time fan-out delivery.
|
||||||
|
func (n *NATS) Subscribe(subject string, handler func(data []byte)) (via.Subscription, error) {
|
||||||
|
sub, err := n.nc.Subscribe(subject, func(msg *nats.Msg) {
|
||||||
|
handler(msg.Data)
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return sub, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close shuts down the client connection and embedded server.
|
||||||
|
func (n *NATS) Close() error {
|
||||||
|
n.nc.Close()
|
||||||
|
return n.server.Close()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Conn returns the underlying NATS connection for advanced usage.
|
||||||
|
func (n *NATS) Conn() *nats.Conn {
|
||||||
|
return n.nc
|
||||||
|
}
|
||||||
|
|
||||||
|
// JetStream returns the JetStream context for stream configuration and replay.
|
||||||
|
func (n *NATS) JetStream() nats.JetStreamContext {
|
||||||
|
return n.js
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user