119 lines
2.6 KiB
Go
119 lines
2.6 KiB
Go
package sse
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
|
|
"ruben/inventory2/logging"
|
|
"ruben/inventory2/server/auth"
|
|
"ruben/inventory2/server/response"
|
|
"ruben/inventory2/server/sse"
|
|
)
|
|
|
|
const (
|
|
maxNumOpenConnectionsPerUser = 3
|
|
)
|
|
|
|
type (
|
|
sseRouter struct {
|
|
log *logging.Logger
|
|
sse *sse.Queue
|
|
|
|
users map[string][maxNumOpenConnectionsPerUser]context.CancelFunc
|
|
lock sync.Mutex
|
|
}
|
|
)
|
|
|
|
func Routes(
|
|
r gin.IRouter,
|
|
logger *logging.Logger,
|
|
sq *sse.Queue,
|
|
) {
|
|
s := &sseRouter{
|
|
log: logger,
|
|
sse: sq,
|
|
users: make(map[string][maxNumOpenConnectionsPerUser]context.CancelFunc),
|
|
}
|
|
|
|
r.GET("/", response.Handler(s.serveEvents))
|
|
}
|
|
|
|
func (r *sseRouter) serveEvents(c *gin.Context) (response.Response, error) {
|
|
acct := auth.GetIdentity(c).Account
|
|
acctID := acct.AccountID
|
|
userID := acct.UserID
|
|
email := acct.Email
|
|
|
|
log := r.log.WithGroup("serveEvents").With(
|
|
"accountID", acctID,
|
|
"userID", userID,
|
|
"email", email,
|
|
"userAgent", c.Request.UserAgent(),
|
|
)
|
|
log.Info("user connected to sse queue")
|
|
|
|
ctx := r.closeOutstandingConnectionsForUserAndStoreCancelFuncForUser(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) closeOutstandingConnectionsForUserAndStoreCancelFuncForUser(ctx context.Context, userID string) context.Context {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
|
|
closeConnFuncs := r.users[userID]
|
|
defer func() {
|
|
r.users[userID] = closeConnFuncs
|
|
}()
|
|
|
|
// close existing connection
|
|
|
|
for i := range maxNumOpenConnectionsPerUser {
|
|
if fn := closeConnFuncs[i]; fn == nil {
|
|
// not hit limit on connections.
|
|
// save the cancellation func and done
|
|
closeConnFuncs[i] = cancel
|
|
return ctx
|
|
}
|
|
}
|
|
|
|
// close the oldest connection, shift all cancellation funcs down, and push the new one in
|
|
closeConnFuncs[0]()
|
|
for i := range maxNumOpenConnectionsPerUser - 1 {
|
|
closeConnFuncs[i] = closeConnFuncs[i+1]
|
|
}
|
|
closeConnFuncs[maxNumOpenConnectionsPerUser-1] = cancel
|
|
|
|
return ctx
|
|
}
|