98 lines
2.1 KiB
Go
98 lines
2.1 KiB
Go
package sse
|
|
|
|
import (
|
|
"context"
|
|
"ruben/inventory2/internal/logging"
|
|
"ruben/inventory2/internal/server/middleware"
|
|
"ruben/inventory2/internal/server/response"
|
|
"ruben/inventory2/internal/server/sse"
|
|
"sync"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
)
|
|
|
|
type (
|
|
sseRouter struct {
|
|
log *logging.Logger
|
|
sse *sse.Queue
|
|
|
|
users map[string]context.CancelFunc
|
|
lock sync.Mutex
|
|
}
|
|
)
|
|
|
|
func Routes(
|
|
r gin.IRouter,
|
|
logger *logging.Logger,
|
|
sq *sse.Queue,
|
|
auth *middleware.Auth,
|
|
) {
|
|
s := &sseRouter{
|
|
log: logger,
|
|
sse: sq,
|
|
users: make(map[string]context.CancelFunc),
|
|
}
|
|
|
|
r.GET("/", auth.AuthenticateAndAddIdentityGin(), response.Handler(s.serveEvents))
|
|
}
|
|
|
|
func (r *sseRouter) serveEvents(c *gin.Context) (response.Response, error) {
|
|
acct := middleware.GetIdentity(c).Account
|
|
acctID := acct.AccountID
|
|
userID := acct.UserID
|
|
email := acct.Email
|
|
|
|
log := r.log.WithGroup("serveEvents").With(
|
|
"accountID", acctID,
|
|
"userID", userID,
|
|
"email", email,
|
|
)
|
|
log.Info("user connected to sse queue")
|
|
|
|
ctx := r.closeExistingConnectionsForUserAndStoreCancelFuncForUser(c, userID)
|
|
|
|
w := c.Writer
|
|
|
|
hdr := w.Header()
|
|
hdr.Set("Access-Control-Allow-Origin", "*")
|
|
hdr.Set("Access-Control-Expose-Headers", "Content-Type")
|
|
hdr.Set("Content-Type", "text/event-stream")
|
|
hdr.Set("Connection", "keep-alive")
|
|
hdr.Set("Cache-Control", "no-cache")
|
|
w.Flush()
|
|
|
|
err := r.sse.Listen(ctx, acctID, func(ctx context.Context, e *sse.Event) error {
|
|
log.Debugf("sending event of type %s", e.Type)
|
|
|
|
e.Write(w)
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
log.Errorf("no longer connected to sse queue due to error: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
// send a 'close' message so the front end doesn't try to reconnect
|
|
(&sse.Event{Type: "close"}).Write(w)
|
|
|
|
log.Info("user disconnecting sse queue")
|
|
return response.Status(200), nil
|
|
}
|
|
|
|
func (r *sseRouter) closeExistingConnectionsForUserAndStoreCancelFuncForUser(ctx context.Context, userID string) context.Context {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
|
|
// close existing connection
|
|
if closeConn, ok := r.users[userID]; ok {
|
|
closeConn()
|
|
delete(r.users, userID)
|
|
}
|
|
|
|
// store reference to cancel func
|
|
ctx, r.users[userID] = context.WithCancel(ctx)
|
|
|
|
return ctx
|
|
}
|