diff --git a/sam/repository/event_queue.go b/sam/repository/event_queue.go index 6adafc9bf..09fc8bfce 100644 --- a/sam/repository/event_queue.go +++ b/sam/repository/event_queue.go @@ -1,32 +1,20 @@ package repository import ( - "encoding/json" + "github.com/crusttech/crust/sam/types" "github.com/titpetric/factory" ) type ( EventQueue interface { - EventQueuePull(origin uint64) ([]*EventQueueItem, error) - EventQueuePush(eqi *EventQueueItem) error + EventQueuePull(origin uint64) ([]*types.EventQueueItem, error) + EventQueuePush(eqi *types.EventQueueItem) error EventQueueSync(origin uint64, ID uint64) error } - - EventQueueItem struct { - ID uint64 `db:"id"` - Origin uint64 `db:"origin"` - Subscriber string `db:"subscriber"` - Payload json.RawMessage `db:"payload"` - } - - evqs struct { - Origin uint64 `db:"origin"` - LastEvent uint64 `db:"rel_last"` - } ) -func (r *repository) EventQueuePull(origin uint64) ([]*EventQueueItem, error) { - var ee = make([]*EventQueueItem, 0) +func (r *repository) EventQueuePull(origin uint64) ([]*types.EventQueueItem, error) { + var ee = make([]*types.EventQueueItem, 0) return ee, r.db().Quiet().Select(&ee, ` SELECT * @@ -36,12 +24,17 @@ func (r *repository) EventQueuePull(origin uint64) ([]*EventQueueItem, error) { LIMIT 50`, origin, origin, origin) } -func (r *repository) EventQueuePush(eqi *EventQueueItem) error { +func (r *repository) EventQueuePush(eqi *types.EventQueueItem) error { eqi.ID = factory.Sonyflake.NextID() return r.db().Quiet().Insert("event_queue", eqi) } func (r *repository) EventQueueSync(origin uint64, ID uint64) error { + type evqs struct { + Origin uint64 `db:"origin"` + LastEvent uint64 `db:"rel_last"` + } + // @todo do we even need this? return r.db().Quiet().Replace("event_queue_synced", evqs{ Origin: origin, diff --git a/sam/types/event_queue.go b/sam/types/event_queue.go new file mode 100644 index 000000000..50daa8894 --- /dev/null +++ b/sam/types/event_queue.go @@ -0,0 +1,14 @@ +package types + +import ( + "encoding/json" +) + +type ( + EventQueueItem struct { + ID uint64 `db:"id"` + Origin uint64 `db:"origin"` + Subscriber string `db:"subscriber"` + Payload json.RawMessage `db:"payload"` + } +) diff --git a/sam/websocket/event_queue.go b/sam/websocket/event_queue.go index 5a0330b27..03efe80ab 100644 --- a/sam/websocket/event_queue.go +++ b/sam/websocket/event_queue.go @@ -2,7 +2,7 @@ package websocket import ( "context" - "github.com/crusttech/crust/sam/repository" + "github.com/crusttech/crust/sam/types" "github.com/titpetric/factory" "log" "time" @@ -10,11 +10,11 @@ import ( type ( eventQueuePuller interface { - EventQueuePull(origin uint64) ([]*repository.EventQueueItem, error) + EventQueuePull(origin uint64) ([]*types.EventQueueItem, error) EventQueueSync(origin uint64, ID uint64) error } eventQueuePusher interface { - EventQueuePush(*repository.EventQueueItem) error + EventQueuePush(*types.EventQueueItem) error } eventQueueWalker interface { @@ -23,7 +23,7 @@ type ( eventQueue struct { origin uint64 - queue chan *repository.EventQueueItem + queue chan *types.EventQueueItem } ) @@ -36,7 +36,7 @@ func init() { func EventQueue(origin uint64) *eventQueue { return &eventQueue{ origin: origin, - queue: make(chan *repository.EventQueueItem, 512), + queue: make(chan *types.EventQueueItem, 512), } } @@ -53,7 +53,7 @@ func (eq *eventQueue) store(ctx context.Context, qp eventQueuePusher) { } func (eq *eventQueue) feedSessions(ctx context.Context, qp eventQueuePuller, store eventQueueWalker) { - var items []*repository.EventQueueItem + var items []*types.EventQueueItem go func() { var err error @@ -112,7 +112,7 @@ func (eq *eventQueue) feedSessions(ctx context.Context, qp eventQueuePuller, sto } // Adds origin to the event and puts it into queue. -func (eq *eventQueue) push(ctx context.Context, eqi *repository.EventQueueItem) { +func (eq *eventQueue) push(ctx context.Context, eqi *types.EventQueueItem) { eqi.Origin = eq.origin select { diff --git a/sam/websocket/session_outgoing.go b/sam/websocket/session_outgoing.go index f53b4b8b6..f2772c12d 100644 --- a/sam/websocket/session_outgoing.go +++ b/sam/websocket/session_outgoing.go @@ -1,7 +1,7 @@ package websocket import ( - "github.com/crusttech/crust/sam/repository" + "github.com/crusttech/crust/sam/types" "github.com/crusttech/crust/sam/websocket/outgoing" "log" "time" @@ -14,7 +14,7 @@ func (s *Session) sendToAllSubscribers(p outgoing.MessageEncoder, channelID stri return err } - eq.push(s.ctx, &repository.EventQueueItem{Payload: pb, Subscriber: channelID}) + eq.push(s.ctx, &types.EventQueueItem{Payload: pb, Subscriber: channelID}) store.Walk(func(sess *Session) { // send message only to users with subscribed channels @@ -33,7 +33,7 @@ func (s *Session) sendToAll(p outgoing.MessageEncoder) error { return err } - eq.push(s.ctx, &repository.EventQueueItem{Payload: pb}) + eq.push(s.ctx, &types.EventQueueItem{Payload: pb}) store.Walk(func(sess *Session) { // send message only to users with subscribed channels