removed intermediate /internal directory
This commit is contained in:
@@ -0,0 +1,160 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"ruben/inventory2/logging"
|
||||
)
|
||||
|
||||
type (
|
||||
UpdateNotificationPublisher struct {
|
||||
log *logging.Logger
|
||||
queue sender
|
||||
getAccountID func(*gin.Context) int64
|
||||
trimBasePath string
|
||||
basePathPattern string
|
||||
}
|
||||
|
||||
// sender is satisfied by *Queue
|
||||
sender interface {
|
||||
Send(ctx context.Context, e Event) error
|
||||
}
|
||||
)
|
||||
|
||||
func (q *Queue) NewUpdateNotificationPublisher(
|
||||
log *logging.Logger,
|
||||
getAccountID func(*gin.Context) int64,
|
||||
) *UpdateNotificationPublisher {
|
||||
return &UpdateNotificationPublisher{
|
||||
log: log,
|
||||
queue: q,
|
||||
getAccountID: getAccountID,
|
||||
}
|
||||
}
|
||||
|
||||
// Trim produces an *UpdateNotificationPublisher with the pathPattern
|
||||
// trimmed from events otherwise published by p.
|
||||
func (p *UpdateNotificationPublisher) Trim(pathPattern string) *UpdateNotificationPublisher {
|
||||
p2 := *p
|
||||
p2.trimBasePath = path.Join(p2.trimBasePath, pathPattern)
|
||||
return &p2
|
||||
}
|
||||
|
||||
// Group produces an *UpdateNotificationPublisher with the pathPattern
|
||||
// appended to the base path used by p, if any, in publishing events.
|
||||
// If no base path was set to p, then pathPattern becomes the base path.
|
||||
func (p *UpdateNotificationPublisher) Group(pathPattern string) *UpdateNotificationPublisher {
|
||||
p2 := *p
|
||||
p2.basePathPattern = path.Join(p2.basePathPattern, pathPattern)
|
||||
return &p2
|
||||
}
|
||||
|
||||
// Publish constructs a gin middleware.
|
||||
// It must used with and after middleware puts in the Identity into the gin.Context.
|
||||
func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFunc {
|
||||
trimBasePathSegs := getPathSegments(p.trimBasePath)
|
||||
trimmedBasePathPattern := path.Join(p.basePathPattern, pathPattern)
|
||||
trimmedBasePathPatternSegs := getPathSegments(trimmedBasePathPattern)
|
||||
fullBasePathPatternSegs := append(trimBasePathSegs, trimmedBasePathPatternSegs...)
|
||||
|
||||
return func(c *gin.Context) {
|
||||
reqPathSegs := getPathSegments(c.Request.URL.Path)
|
||||
|
||||
c.Next()
|
||||
if len(c.Errors) > 0 {
|
||||
return
|
||||
}
|
||||
if len(reqPathSegs) < len(fullBasePathPatternSegs) {
|
||||
p.log.Errorf("req path is less that the full path: %v, %v", c.Request.URL.Path, path.Join(p.trimBasePath, p.basePathPattern, pathPattern))
|
||||
return
|
||||
}
|
||||
|
||||
// if the path is a subpath, then emit events along the subpath.
|
||||
|
||||
for i, s := range trimmedBasePathPatternSegs {
|
||||
if isWildcard := s[0] == ':'; isWildcard {
|
||||
continue
|
||||
}
|
||||
|
||||
rs := reqPathSegs[i]
|
||||
if isSubpath := s != rs; isSubpath {
|
||||
continue
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
topEventParts := reqPathSegs[len(trimBasePathSegs):len(fullBasePathPatternSegs)]
|
||||
topEvent := strings.Join(topEventParts, "_")
|
||||
|
||||
events := make([]string, len(reqPathSegs)-len(fullBasePathPatternSegs)+1)
|
||||
events[0] = topEvent
|
||||
parentEvent := topEvent
|
||||
for i, s := range reqPathSegs[len(fullBasePathPatternSegs):] {
|
||||
e := parentEvent + "_" + s
|
||||
events[i+1] = e
|
||||
parentEvent = e
|
||||
}
|
||||
|
||||
acctID := p.getAccountID(c)
|
||||
for _, e := range events {
|
||||
go func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
|
||||
if err := p.queue.Send(ctx, Event{
|
||||
AccountID: acctID,
|
||||
Type: e,
|
||||
Data: []byte(fmt.Sprintf(`{"eventType": %q}`, e)),
|
||||
}); err != nil {
|
||||
p.log.Errorf("failed to send sse event to listener: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *UpdateNotificationPublisher) Push(ctx context.Context, acctID int64, eventTypes ...string) error {
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(len(eventTypes))
|
||||
|
||||
errs := make([]error, len(eventTypes))
|
||||
for i, e := range eventTypes {
|
||||
i := i
|
||||
e := e
|
||||
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if err := p.queue.Send(ctx, Event{
|
||||
AccountID: acctID,
|
||||
Type: e,
|
||||
Data: []byte(fmt.Sprintf(`{"eventType": %q}`, e)),
|
||||
}); err != nil {
|
||||
errs[i] = fmt.Errorf("failed to send sse event to listener: %w", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
return errors.Join(errs...)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getPathSegments(p string) []string {
|
||||
p = path.Clean(p)
|
||||
if p == "" || p == "." || p == "/" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return strings.Split(strings.Trim(p, "/"), "/")
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"maps"
|
||||
"net/http"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type (
|
||||
Queue struct {
|
||||
in chan Event
|
||||
out map[int64]map[int]chan Event
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
Event struct {
|
||||
AccountID int64
|
||||
Type string
|
||||
Data []byte
|
||||
}
|
||||
)
|
||||
|
||||
func NewQueue() *Queue {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &Queue{
|
||||
in: make(chan Event),
|
||||
out: make(map[int64]map[int]chan Event),
|
||||
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) Start(ctx context.Context) error {
|
||||
defer q.cancel()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// we're done piping events
|
||||
return nil
|
||||
case e := <-q.in:
|
||||
if e.AccountID == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// share event with listeners on the account
|
||||
q.lock.Lock()
|
||||
for _, out := range q.out[e.AccountID] {
|
||||
out <- e
|
||||
}
|
||||
q.lock.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) Listen(ctx context.Context, acctID int64, fn func(context.Context, *Event) error) error {
|
||||
// create new out pipe and append it to the queue
|
||||
|
||||
out := make(chan Event, 1)
|
||||
|
||||
q.lock.Lock()
|
||||
outs := q.out[acctID]
|
||||
if outs == nil {
|
||||
outs = make(map[int]chan Event)
|
||||
q.out[acctID] = outs
|
||||
}
|
||||
|
||||
var maxID int
|
||||
for n := range maps.Keys(outs) {
|
||||
maxID = max(maxID, n)
|
||||
}
|
||||
id := maxID + 1
|
||||
|
||||
outs[id] = out
|
||||
q.lock.Unlock()
|
||||
|
||||
// delete the pipe when done listening
|
||||
defer func() {
|
||||
q.lock.Lock()
|
||||
delete(q.out[acctID], id)
|
||||
if len(q.out[acctID]) == 0 {
|
||||
delete(q.out, acctID)
|
||||
}
|
||||
q.lock.Unlock()
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-q.ctx.Done():
|
||||
// queue is shut down
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
// done listening to events
|
||||
return nil
|
||||
case e := <-out:
|
||||
// pass event to the caller
|
||||
if err := fn(ctx, &e); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) Send(ctx context.Context, e Event) error {
|
||||
select {
|
||||
case <-q.ctx.Done():
|
||||
// the queue has closed
|
||||
return fmt.Errorf("event queue closed: %w", q.ctx.Err())
|
||||
case <-ctx.Done():
|
||||
// sender ran out of time
|
||||
return fmt.Errorf("provided context canceled: %w", ctx.Err())
|
||||
// push the event onto the queue
|
||||
case q.in <- e:
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Event) Write(w http.ResponseWriter) {
|
||||
fmt.Fprintf(
|
||||
w,
|
||||
"event: %s\ndata: %s\n\n",
|
||||
e.Type,
|
||||
bytes.ReplaceAll(e.Data, []byte("\n"), []byte(" ")),
|
||||
)
|
||||
w.(http.Flusher).Flush()
|
||||
}
|
||||
Reference in New Issue
Block a user