diff --git a/sam/repository/channel.go b/sam/repository/channel.go index 558440bd3..b5b47ac3a 100644 --- a/sam/repository/channel.go +++ b/sam/repository/channel.go @@ -157,6 +157,7 @@ func (r *channel) FindMembers(userID uint64) (types.ChannelMemberSet, error) { func (r *channel) CreateChannel(mod *types.Channel) (*types.Channel, error) { mod.ID = factory.Sonyflake.NextID() mod.CreatedAt = time.Now() + mod.UpdatedAt = nil mod.Meta = coalesceJson(mod.Meta, []byte("{}")) if mod.Type == "" { diff --git a/sam/service/events.go b/sam/service/events.go index b9a2ca152..a365700b9 100644 --- a/sam/service/events.go +++ b/sam/service/events.go @@ -35,8 +35,9 @@ func (svc *event) With(ctx context.Context) EventService { } } +// Message sends message events to subscribers func (svc *event) Message(m *types.Message) error { - return svc.push(payload.Message(m), m.ChannelID) + return svc.push(payload.Message(m), types.EventQueueItemSubTypeChannel, m.ChannelID) } // Channel notifies subscribers about channel change @@ -48,10 +49,18 @@ func (svc *event) Channel(ch *types.Channel) error { sub = 0 } - return svc.push(payload.Channel(ch), sub) + return svc.push(payload.Channel(ch), types.EventQueueItemSubTypeChannel, sub) } -func (svc *event) push(m outgoing.MessageEncoder, sub uint64) error { +// Join sends force-channel-join event to all matching subscriptions +// +// Subscription will match when session's user ID is the same as sub +// We pack channel ID (the id to subscribe to) as payload +func (svc *event) Join(userID, channelID uint64) error { + return svc.push(payload.Channel(&types.Channel{ID: channelID}), types.EventQueueItemSubTypeUser, userID) +} + +func (svc *event) push(m outgoing.MessageEncoder, subType types.EventQueueItemSubType, sub uint64) error { var enc, err = m.EncodeMessage() if err != nil { return err @@ -61,6 +70,7 @@ func (svc *event) push(m outgoing.MessageEncoder, sub uint64) error { if sub > 0 { item.Subscriber = payload.Uint64toa(sub) + item.SubType = subType } return svc.events.Push(svc.ctx, item) diff --git a/sam/types/event_queue.go b/sam/types/event_queue.go index 50daa8894..fa38a4ce9 100644 --- a/sam/types/event_queue.go +++ b/sam/types/event_queue.go @@ -6,9 +6,17 @@ import ( type ( EventQueueItem struct { - ID uint64 `db:"id"` - Origin uint64 `db:"origin"` - Subscriber string `db:"subscriber"` - Payload json.RawMessage `db:"payload"` + ID uint64 `db:"id"` + Origin uint64 `db:"origin"` + SubType EventQueueItemSubType `db:"subtype"` + Subscriber string `db:"subscriber"` + Payload json.RawMessage `db:"payload"` } + + EventQueueItemSubType string +) + +const ( + EventQueueItemSubTypeUser = "user" + EventQueueItemSubTypeChannel = "channel" ) diff --git a/sam/websocket/event_queue.go b/sam/websocket/event_queue.go index ecbf7b438..5bf7c086a 100644 --- a/sam/websocket/event_queue.go +++ b/sam/websocket/event_queue.go @@ -2,9 +2,12 @@ package websocket import ( "context" + "encoding/json" "github.com/titpetric/factory" + "github.com/crusttech/crust/internal/payload" + "github.com/crusttech/crust/internal/payload/outgoing" "github.com/crusttech/crust/sam/repository" "github.com/crusttech/crust/sam/types" ) @@ -63,12 +66,30 @@ func (eq *eventQueue) feedSessions(ctx context.Context, config *repository.Flags s.sendBytes(item.Payload) }) } else { - // Distribute payload to specific subscribers - store.Walk(func(s *Session) { - if s.subs.Get(item.Subscriber) != nil { - s.sendBytes(item.Payload) + switch item.SubType { + case types.EventQueueItemSubTypeUser: + // Holds channel info (in fact, ID only) when payload is unmarshaled + var join *outgoing.Channel + if err := json.Unmarshal(item.Payload, join); err == nil { + return err } - }) + + // This store.Walk handler does send to subscribed sessions but + // subscribes all sessions that belong to the same user + store.Walk(func(s *Session) { + if payload.Uint64toa(s.user.Identity()) == item.Subscriber { + s.subs.Add(join.ID) + } + }) + case types.EventQueueItemSubTypeChannel: + // Distribute payload to specific subscribers + store.Walk(func(s *Session) { + if s.subs.Get(item.Subscriber) != nil { + s.sendBytes(item.Payload) + } + }) + } + } } } diff --git a/sam/websocket/session_incoming_channel.go b/sam/websocket/session_incoming_channel.go index 322e0facf..2dbba1f50 100644 --- a/sam/websocket/session_incoming_channel.go +++ b/sam/websocket/session_incoming_channel.go @@ -77,21 +77,7 @@ func (s *Session) channelCreate(ctx context.Context, p *incoming.ChannelCreate) return err } - // Explicitly subscribe to newly created channel - s.subs.Add(payload.Uint64toa(ch.ID)) - - // @todo this should go over all user's sessions and subscribe there as well - - // @todo load channel member count - - pl := payload.Channel(ch) - - if ch.Type == types.ChannelTypePublic { - return s.sendToAll(pl) - } - - // By default, just send reply to user - return s.sendReply(pl) + return nil } func (s *Session) channelDelete(ctx context.Context, p *incoming.ChannelDelete) (err error) { @@ -116,12 +102,6 @@ func (s *Session) channelUpdate(ctx context.Context, p *incoming.ChannelUpdate) ch.Type = types.ChannelType(*p.Type) } - ch, err = s.svc.ch.With(ctx).Update(ch) - if err != nil { - return err - } - - // @todo load channel member count - - return s.sendToAllSubscribers(payload.Channel(ch), p.ID) + _, err = s.svc.ch.With(ctx).Update(ch) + return err }