diff --git a/internal/payload/outgoing/activity.go b/internal/payload/outgoing/activity.go index 02d55683c..03b165c7f 100644 --- a/internal/payload/outgoing/activity.go +++ b/internal/payload/outgoing/activity.go @@ -9,6 +9,7 @@ type ( Kind string `json:"kind,omitempty"` MessageID uint64 `json:"messageID,string,omitempty"` ChannelID uint64 `json:"channelID,string,omitempty"` + Present bool `json:"present"` } ) diff --git a/internal/payload/outgoing/connection.go b/internal/payload/outgoing/connection.go deleted file mode 100644 index 6a4f86634..000000000 --- a/internal/payload/outgoing/connection.go +++ /dev/null @@ -1,25 +0,0 @@ -package outgoing - -import ( - "encoding/json" -) - -type ( - Connected struct { - // Who did connect? - UserID string `json:"uid"` - } - - Disconnected struct { - // Who did disconnect? - UserID string `json:"uid"` - } -) - -func (p *Connected) EncodeMessage() ([]byte, error) { - return json.Marshal(Payload{Connected: p}) -} - -func (p *Disconnected) EncodeMessage() ([]byte, error) { - return json.Marshal(Payload{Disconnected: p}) -} diff --git a/internal/payload/outgoing/payload.go b/internal/payload/outgoing/payload.go index cc7f452af..c556c6f47 100644 --- a/internal/payload/outgoing/payload.go +++ b/internal/payload/outgoing/payload.go @@ -4,9 +4,6 @@ type ( Payload struct { *Error `json:"error,omitempty"` - *Connected `json:"clientConnected,omitempty"` - *Disconnected `json:"clientDisconnected,omitempty"` - *Message `json:"message,omitempty"` *MessageSet `json:"messages,omitempty"` diff --git a/messaging/rest/activity.go b/messaging/rest/activity.go index 376719cbe..d7ed59c0c 100644 --- a/messaging/rest/activity.go +++ b/messaging/rest/activity.go @@ -29,6 +29,13 @@ func (ctrl *Activity) Send(ctx context.Context, r *request.ActivitySend) (interf return nil, errors.New("can not send activity on message without channel ID") } + switch r.Kind { + case "": + return nil, errors.New("missing value for activity kind") + case "connected", "disconnected": + return nil, errors.New("can not use reserved values for activity kind") + } + return true, ctrl.event.With(ctx).Activity(&types.Activity{ UserID: auth.GetIdentityFromContext(ctx).Identity(), ChannelID: r.ChannelID, diff --git a/messaging/websocket/session.go b/messaging/websocket/session.go index 1902fff64..7b904e8b0 100644 --- a/messaging/websocket/session.go +++ b/messaging/websocket/session.go @@ -23,7 +23,9 @@ type ( id uint64 once sync.Once conn *websocket.Conn - ctx context.Context + + ctx context.Context + ctxCancel context.CancelFunc subs *Subscriptions @@ -44,15 +46,17 @@ type ( ) func (Session) New(ctx context.Context, config *repository.Flags, conn *websocket.Conn) *Session { + s := &Session{ conn: conn, - ctx: ctx, config: config, subs: NewSubscriptions(), send: make(chan []byte, 512), stop: make(chan []byte, 1), } + s.ctx, s.ctxCancel = context.WithCancel(ctx) + s.svc.ch = service.DefaultChannel s.svc.msg = service.DefaultMessage @@ -86,16 +90,53 @@ func (sess *Session) connected() (err error) { } // Tell everyone that user has connected - if err = sess.sendToAll(&outgoing.Connected{UserID: payload.Uint64toa(sess.user.Identity())}); err != nil { + if err = sess.sendPresence("connected"); err != nil { return } + // Create a heartbeat every minute for this user + go func() { + t := time.NewTicker(time.Second * 3) + for { + select { + case <-sess.ctx.Done(): + return + case <-t.C: + sess.sendPresence("") + } + } + }() + return nil } func (sess *Session) disconnected() { // Tell everyone that user has disconnected - _ = sess.sendToAll(&outgoing.Disconnected{UserID: payload.Uint64toa(sess.user.Identity())}) + _ = sess.sendPresence("disconnected") + + // Cancel context + sess.ctxCancel() + + // Close connection + sess.conn.Close() + sess.conn = nil +} + +// Sends user presence information to all subscribers +// +// It sends "connected", "disconnected" and "" activity kinds +func (sess *Session) sendPresence(kind string) error { + connections := store.CountConnections(sess.user.Identity()) + if kind == "disconnected" { + connections-- + } + + // Tell everyone that user has disconnected + return sess.sendToAll(&outgoing.Activity{ + UserID: sess.user.Identity(), + Kind: kind, + Present: connections > 0, + }) } func (sess *Session) Handle() (err error) { @@ -111,8 +152,6 @@ func (sess *Session) Handle() (err error) { func (sess *Session) Close() { sess.once.Do(func() { sess.disconnected() - sess.conn.Close() - sess.conn = nil store.Delete(sess.id) }) } diff --git a/messaging/websocket/session_store.go b/messaging/websocket/session_store.go index 5e625bba4..d1a6389b5 100644 --- a/messaging/websocket/session_store.go +++ b/messaging/websocket/session_store.go @@ -1,8 +1,9 @@ package websocket import ( - "github.com/titpetric/factory" "sync" + + "github.com/titpetric/factory" ) type ( @@ -39,6 +40,16 @@ func (s *Store) Walk(callback func(*Session)) { } } +func (s *Store) CountConnections(userID uint64) (count uint) { + s.Walk(func(session *Session) { + if session.user.Identity() == userID { + count++ + } + }) + + return +} + func (s *Store) Get(id uint64) *Session { s.RLock() defer s.RUnlock()