From 9f013c9c9c93e9ff3ac90c58bca40d9b8ebd0cfb Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Wed, 17 Oct 2018 04:04:56 +0200 Subject: [PATCH] Direct channels & groups are now properly broadcasted to all new members --- sam/service/channel.go | 37 ++++++++++++++++++++++++++++++------ sam/service/events.go | 6 +++--- sam/types/channel_member.go | 3 +++ sam/websocket/event_queue.go | 2 -- 4 files changed, 37 insertions(+), 11 deletions(-) diff --git a/sam/service/channel.go b/sam/service/channel.go index a59ca913f..771dcc1b1 100644 --- a/sam/service/channel.go +++ b/sam/service/channel.go @@ -288,7 +288,7 @@ func (svc *channel) Create(in *types.Channel) (out *types.Channel, err error) { svc.flushSystemMessages() - return svc.evl.Channel(out) + return svc.sendChannelEvent(out) }) } @@ -353,7 +353,7 @@ func (svc *channel) Update(ch *types.Channel) (out *types.Channel, err error) { svc.flushSystemMessages() - return svc.evl.Channel(out) + return svc.sendChannelEvent(out) }) } @@ -383,7 +383,7 @@ func (svc *channel) Delete(id uint64) error { } svc.flushSystemMessages() - return svc.evl.Channel(ch) + return svc.sendChannelEvent(ch) }) } @@ -410,7 +410,7 @@ func (svc *channel) Recover(id uint64) error { } svc.flushSystemMessages() - return svc.evl.Channel(ch) + return svc.sendChannelEvent(ch) }) } @@ -438,7 +438,7 @@ func (svc *channel) Archive(id uint64) error { } svc.flushSystemMessages() - return svc.evl.Channel(ch) + return svc.sendChannelEvent(ch) }) } @@ -461,7 +461,7 @@ func (svc *channel) Unarchive(id uint64) error { svc.scheduleSystemMessage(ch, "@%d unarchived this channel", userID) svc.flushSystemMessages() - return svc.evl.Channel(ch) + return svc.sendChannelEvent(ch) }) } @@ -595,7 +595,11 @@ func (svc *channel) AddMember(channelID uint64, memberIDs ...uint64) (out types. } out = append(out, member) + } + // Push channel to all members + if err = svc.sendChannelEvent(ch); err != nil { + return } return svc.flushSystemMessages() @@ -671,7 +675,28 @@ func (svc *channel) flushSystemMessages() (err error) { return svc.evl.Message(msg) } }) +} +// Sends channel event +func (svc *channel) sendChannelEvent(ch *types.Channel) (err error) { + if ch.DeletedAt == nil && ch.ArchivedAt == nil { + // Looks like a valid channel + + // Preload members, if needed + if len(ch.Members) == 0 { + if mm, err := svc.cmember.Find(&types.ChannelMemberFilter{ChannelID: ch.ID}); err != nil { + return err + } else { + ch.Members = mm.MembersOf(ch.ID) + } + } + } + + if err = svc.evl.Channel(ch); err != nil { + return + } + + return nil } //// @todo temp location, move this somewhere else diff --git a/sam/service/events.go b/sam/service/events.go index 8934e4065..c34b426cc 100644 --- a/sam/service/events.go +++ b/sam/service/events.go @@ -46,10 +46,10 @@ func (svc *event) Message(m *types.Message) error { // // If this is a public channel we notify everyone func (svc *event) Channel(ch *types.Channel) error { - sub := ch.ID + var sub uint64 = 0 - if ch.Type == types.ChannelTypePublic { - sub = 0 + if ch.Type != types.ChannelTypePublic { + sub = ch.ID } return svc.push(payload.Channel(ch), types.EventQueueItemSubTypeChannel, sub) diff --git a/sam/types/channel_member.go b/sam/types/channel_member.go index d4e82dba9..c347227bc 100644 --- a/sam/types/channel_member.go +++ b/sam/types/channel_member.go @@ -40,6 +40,9 @@ func (mm ChannelMemberSet) Walk(w func(*ChannelMember) error) (err error) { return } +// MembersOf extracts member IDs from channel member set +// +// It filters out only members that match a particular channel func (mm ChannelMemberSet) MembersOf(channelID uint64) []uint64 { var mmof = make([]uint64, 0) diff --git a/sam/websocket/event_queue.go b/sam/websocket/event_queue.go index 99e5dc529..6a62d6ae8 100644 --- a/sam/websocket/event_queue.go +++ b/sam/websocket/event_queue.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" - "github.com/davecgh/go-spew/spew" "github.com/titpetric/factory" "github.com/crusttech/crust/internal/payload" @@ -64,7 +63,6 @@ func (eq *eventQueue) feedSessions(ctx context.Context, config *repository.Flags if item.SubType == types.EventQueueItemSubTypeUser { p := &outgoing.Payload{} - spew.Dump(string(item.Payload)) if err := json.Unmarshal(item.Payload, p); err != nil { return err }