DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Simplified forced-subscription logic

Browse Source
This commit is contained in:
Denis Arh
2018-10-16 13:48:17 +02:00
parent 2a29c50085
commit c5d70ef264
6 changed files with 112 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/payload/outgoing.go
+14
View File
@@ -95,6 +95,20 @@ func ChannelView(v *sam.ChannelView) *outgoing.ChannelView {
}
}
func ChannelJoin(channelID, userID uint64) *outgoing.ChannelJoin {
return &outgoing.ChannelJoin{
ID: Uint64toa(channelID),
UserID: Uint64toa(userID),
}
}
func ChannelPart(channelID, userID uint64) *outgoing.ChannelPart {
return &outgoing.ChannelPart{
ID: Uint64toa(channelID),
UserID: Uint64toa(userID),
}
}
func User(user *auth.User) *outgoing.User {
if user == nil {
return nil
internal/payload/outgoing/channel.go
-8
View File
@@ -22,14 +22,6 @@ type (
UserID string `json:"uid"`
}
ChannelDeleted struct {
// Channel that was deleted
ID string `json:"id"`
// Who deleted it
UserID string `json:"uid"`
}
Channel struct {
// Channel to part (nil) for ALL channels
ID string `json:"ID"`
internal/payload/outgoing/payload.go
+4 -5
View File
@@ -10,11 +10,10 @@ type (
*Message `json:"message,omitempty"`
*MessageSet `json:"messages,omitempty"`
*ChannelJoin `json:"channelJoin,omitempty"`
*ChannelPart `json:"channelPart,omitempty"`
*ChannelDeleted `json:"channelDeleted,omitempty"`
*Channel `json:"channel,omitempty"`
*ChannelSet `json:"channels,omitempty"`
*ChannelJoin `json:"channelJoin,omitempty"`
*ChannelPart `json:"channelPart,omitempty"`
*Channel `json:"channel,omitempty"`
*ChannelSet `json:"channels,omitempty"`
*ChannelMember `json:"channelMember,omitempty"`
*ChannelMemberSet `json:"channelMembers,omitempty"`
sam/service/channel.go
+15
View File
@@ -3,6 +3,7 @@ package service
import (
"context"
"fmt"
"time"
"github.com/pkg/errors"
"github.com/titpetric/factory"
@@ -264,6 +265,12 @@ func (svc *channel) Create(in *types.Channel) (out *types.Channel, err error) {
return
}
// When broadcasting, make sure we let the subscribers know
// that creator is also a member...
out.Members = append(out.Members, chCreatorID)
svc.evl.Join(chCreatorID, out.ID)
// Create the first message, doing this directly with repository to circumvent
// message service constraints
svc.scheduleSystemMessage(
@@ -364,6 +371,9 @@ func (svc *channel) Delete(id uint64) error {
if ch.DeletedAt != nil {
return errors.New("Channel already deleted")
} else {
now := time.Now()
ch.DeletedAt = &now
}
svc.scheduleSystemMessage(ch, "@%d deleted this channel", userID)
@@ -578,11 +588,14 @@ func (svc *channel) AddMember(channelID uint64, memberIDs ...uint64) (out types.
member, err = svc.cmember.Create(member)
}
svc.evl.Join(memberID, channelID)
if err != nil {
return err
}
out = append(out, member)
}
return svc.flushSystemMessages()
@@ -623,6 +636,8 @@ func (svc *channel) DeleteMember(channelID uint64, memberIDs ...uint64) (err err
if err = svc.cmember.Delete(channelID, memberID); err != nil {
return err
}
svc.evl.Part(memberID, channelID)
}
return svc.flushSystemMessages()
sam/service/events.go
+38 -4
View File
@@ -20,6 +20,8 @@ type (
With(ctx context.Context) EventService
Message(m *types.Message) error
Channel(m *types.Channel) error
Join(userID, channelID uint64) error
Part(userID, channelID uint64) error
}
)
@@ -45,6 +47,7 @@ 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
if ch.Type == types.ChannelTypePublic {
sub = 0
}
@@ -56,8 +59,40 @@ func (svc *event) Channel(ch *types.Channel) error {
//
// 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) Join(userID, channelID uint64) (err error) {
join := payload.ChannelJoin(channelID, userID)
// Subscribe user to the channel
if err = svc.push(join, types.EventQueueItemSubTypeUser, 0); err != nil {
return
}
// Let subscribers know that this user has joined the channel
if err = svc.push(join, types.EventQueueItemSubTypeChannel, channelID); err != nil {
return
}
return
}
// Part sends force-channel-part 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) Part(userID, channelID uint64) (err error) {
part := payload.ChannelPart(channelID, userID)
// Let subscribers know that this user has parted the channel
if err = svc.push(part, types.EventQueueItemSubTypeChannel, channelID); err != nil {
return
}
// Subscribe user to the channel
if err = svc.push(part, types.EventQueueItemSubTypeUser, 0); err != nil {
return
}
return
}
func (svc *event) push(m outgoing.MessageEncoder, subType types.EventQueueItemSubType, sub uint64) error {
@@ -66,11 +101,10 @@ func (svc *event) push(m outgoing.MessageEncoder, subType types.EventQueueItemSu
return err
}
item := &types.EventQueueItem{Payload: enc}
item := &types.EventQueueItem{Payload: enc, SubType: subType}
if sub > 0 {
item.Subscriber = payload.Uint64toa(sub)
item.SubType = subType
}
return svc.events.Push(svc.ctx, item)
sam/websocket/event_queue.go
+41 -24
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"github.com/davecgh/go-spew/spew"
"github.com/titpetric/factory"
"github.com/crusttech/crust/internal/payload"
@@ -60,37 +61,53 @@ func (eq *eventQueue) feedSessions(ctx context.Context, config *repository.Flags
return err
}
if item.Subscriber == "" {
if item.SubType == types.EventQueueItemSubTypeUser {
p := &outgoing.Payload{}
spew.Dump(string(item.Payload))
if err := json.Unmarshal(item.Payload, p); err != nil {
return err
}
if p.ChannelJoin != nil {
// Handle subscribing
// This store.Walk handler does not 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()) == p.ChannelJoin.UserID {
s.subs.Add(p.ChannelJoin.ID)
}
})
} else if p.ChannelPart != nil {
// Handle un-subscribing
// This store.Walk handler does not 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()) == p.ChannelPart.UserID {
s.subs.Delete(p.ChannelPart.ID)
}
})
} else {
// No other payload types we can handle at the moment.
return nil
}
} else if item.Subscriber == "" {
// Distribute payload to all connected sessions
store.Walk(func(s *Session) {
s.sendBytes(item.Payload)
})
} else {
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
// Distribute payload to specific subscribers
store.Walk(func(s *Session) {
if s.subs.Get(item.Subscriber) != nil {
s.sendBytes(item.Payload)
}
// 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)
}
})
}
})
}
}
}