upd(websocket): implement more websocket apis, structs

sam/websocket/incoming.go

@@ -1,10 +1,7 @@
 package websocket
 
 import (
-	"context"
 	"encoding/json"
-	"github.com/crusttech/crust/sam/service"
-	"github.com/crusttech/crust/sam/types"
 	"github.com/crusttech/crust/sam/websocket/incoming"
 	"github.com/pkg/errors"
 )
@@ -17,48 +14,24 @@ func (s *Session) dispatch(raw []byte) (err error) {
 
 	ctx := s.Context()
 	switch {
+
+	// message actions
 	case p.MessageCreate != nil:
 		return s.messageCreate(ctx, p)
+	case p.MessageUpdate != nil:
+		return s.messageUpdate(ctx, p)
+	case p.MessageDelete != nil:
+		return s.messageDelete(ctx, p)
+
+	// channel actions
+	case p.ChannelJoin != nil:
+		return s.channelJoin(ctx, p)
+	case p.ChannelPart != nil:
+		return s.channelPart(ctx, p)
 	case p.ChannelOpen != nil:
 		return s.channelOpen(ctx, p)
+
 	}
 
 	return nil
 }
-
-func (s *Session) messageCreate(ctx context.Context, payload *incoming.Payload) (err error) {
-	var (
-		request = payload.MessageCreate
-		msg     = &types.Message{Message: request.Message}
-	)
-
-	msg.ChannelID = parseUInt64(request.ChannelID)
-	if msg, err = service.Message().Create(ctx, msg); err != nil {
-		return
-	} else {
-		// @todo move this to outgoing.FromMessage(*types.Message) *outgoing.WsMessage
-		store.MessageFanout(payloadFromMessage(msg))
-	}
-
-	return
-}
-
-func (s *Session) channelOpen(ctx context.Context, payload *incoming.Payload) (err error) {
-	var (
-		request = payload.ChannelOpen
-		filter  = &types.MessageFilter{}
-	)
-
-	filter.ChannelID = parseUInt64(request.ChannelID)
-	filter.FromMessageID = parseUInt64(request.Since)
-
-	if messages, err := service.Message().Find(ctx, filter); err != nil {
-		return err
-	} else {
-		for _, msg := range messages {
-			s.send <- payloadFromMessage(msg)
-		}
-	}
-
-	return
-}

sam/websocket/incoming/channel.go

@@ -0,0 +1,21 @@
+package incoming
+
+type (
+	ChannelJoin struct {
+		ChannelID string `json:"cid"`
+	}
+
+	ChannelPart struct {
+		ChannelID string `json:"cid"`
+	}
+
+	ChannelPartAll struct {
+		Leave bool `json:"leave"`
+	}
+
+	ChannelOpen struct {
+		ChannelID string `json:"cid"`
+		Since     string `json:"since,omitempty"`
+		Until     string `json:"until,omitempty"`
+	}
+)

sam/websocket/incoming/messages.go

@@ -0,0 +1,17 @@
+package incoming
+
+type (
+	MessageCreate struct {
+		ChannelID string `json:"cid"`
+		Message   string `json:"msg"`
+	}
+
+	MessageUpdate struct {
+		ID      string `json:"id"`
+		Message string `json:"msg"`
+	}
+
+	MessageDelete struct {
+		ID    string `json:"id"`
+	}
+)

sam/websocket/incoming/payload.go

@@ -2,8 +2,9 @@ package incoming
 
 type Payload struct {
 	// Channel actions
-	*ChannelJoin `json:"chjoin"`
-	*ChannelPart `json:"chpart"`
+	*ChannelJoin    `json:"chjoin"`
+	*ChannelPart    `json:"chpart"`
+	*ChannelPartAll `json:"chpartall"`
 
 	// Get channel message history
 	*ChannelOpen `json:"chopen"`

sam/websocket/incoming/types.go

@@ -1,30 +0,0 @@
-package incoming
-
-type ChannelJoin struct {
-	ChannelID string `json:"cid"`
-}
-
-type ChannelPart struct {
-	ChannelID string `json:"cid"`
-}
-
-type ChannelOpen struct {
-	ChannelID string `json:"cid"`
-	Since     string `json:"since,omitempty"`
-	Until     string `json:"until,omitempty"`
-}
-
-type MessageCreate struct {
-	ChannelID string `json:"cid"`
-	Message   string `json:"msg"`
-}
-
-type MessageUpdate struct {
-	ID      string `json:"id"`
-	Message string `json:"msg"`
-}
-
-type MessageDelete struct {
-	ID    string `json:"id"`
-	Topic string `json:"topic"`
-}

sam/websocket/incoming_channel.go

@@ -0,0 +1,49 @@
+package websocket
+
+import (
+	"context"
+	"github.com/crusttech/crust/sam/service"
+	"github.com/crusttech/crust/sam/types"
+	"github.com/crusttech/crust/sam/websocket/incoming"
+)
+
+func (s *Session) channelJoin(ctx context.Context, payload *incoming.Payload) error {
+	var (
+		request = payload.ChannelJoin
+	)
+	// @todo: check access to channel
+	s.subs.Add(request.ChannelID, &Subscription{})
+	return nil
+}
+
+func (s *Session) channelPart(ctx context.Context, payload *incoming.Payload) error {
+	var (
+		request = payload.ChannelJoin
+	)
+	// @todo: check access to channel
+	s.subs.Delete(request.ChannelID)
+	return nil
+}
+
+func (s *Session) channelPartAll(ctx context.Context, payload *incoming.Payload) error {
+	if payload.ChannelPartAll.Leave {
+		s.subs.DeleteAll()
+	}
+	return nil
+}
+
+func (s *Session) channelOpen(ctx context.Context, payload *incoming.Payload) error {
+	var (
+		request = payload.ChannelOpen
+		filter  = &types.MessageFilter{
+			ChannelID:     parseUInt64(request.ChannelID),
+			FromMessageID: parseUInt64(request.Since),
+		}
+	)
+
+	messages, err := service.Message().Find(ctx, filter)
+	if err != nil {
+		return err
+	}
+	return s.sendMessage(payloadFromMessages(messages))
+}

sam/websocket/incoming_message.go

@@ -0,0 +1,53 @@
+package websocket
+
+import (
+	"context"
+	"github.com/crusttech/crust/sam/service"
+	"github.com/crusttech/crust/sam/types"
+	"github.com/crusttech/crust/sam/websocket/incoming"
+	"github.com/crusttech/crust/sam/websocket/outgoing"
+)
+
+func (s *Session) messageCreate(ctx context.Context, payload *incoming.Payload) error {
+	var (
+		request = payload.MessageCreate
+		msg     = &types.Message{
+			ChannelID: parseUInt64(request.ChannelID),
+			Message:   request.Message,
+		}
+	)
+
+	msg, err := service.Message().Create(ctx, msg)
+	if err != nil {
+		return err
+	}
+	return s.sendMessageChannel(uint64toa(msg.ChannelID), payloadFromMessage(msg))
+}
+
+func (s *Session) messageUpdate(ctx context.Context, payload *incoming.Payload) error {
+	var (
+		request = payload.MessageUpdate
+		msg     = &types.Message{
+			ID:      parseUInt64(request.ID),
+			Message: request.Message,
+		}
+	)
+	msg, err := service.Message().Update(ctx, msg)
+	if err != nil {
+		return err
+	}
+	return s.sendMessageChannel(uint64toa(msg.ChannelID), &outgoing.MessageUpdate{request.ID, msg.Message})
+}
+
+func (s *Session) messageDelete(ctx context.Context, payload *incoming.Payload) error {
+	var (
+		request = payload.MessageDelete
+		id      = parseUInt64(request.ID)
+	)
+
+	if err := service.Message().Delete(ctx, id); err != nil {
+		return err
+	}
+	// @todo: delete broadcast to channel
+	return s.sendMessageChannel("TODO", &outgoing.MessageDelete{request.ID})
+}

sam/websocket/outgoing.go

@@ -0,0 +1,38 @@
+package websocket
+
+import (
+	"log"
+	"encoding/json"
+	"time"
+	"github.com/crusttech/crust/sam/websocket/outgoing"
+)
+
+func (s *Session) sendMessageChannel(channelID string, message outgoing.PayloadType) error {
+	p := outgoing.Payload{}.New().Load(message)
+	// encode message once and send bytes
+	pb, err := json.Marshal(p)
+	if err != nil {
+		return err
+	}
+	store.Walk(func(sess *Session) {
+		// send message only to users with subscribed channels
+		if sess.subs.Get(channelID) != nil {
+			select {
+			case sess.send <- pb:
+			case <-time.After(2 * time.Millisecond):
+				log.Println("websocket.messageChannel send timeout")
+			}
+		}
+	})
+	return nil
+}
+
+func (s *Session) sendMessage(message outgoing.PayloadType) error {
+	p := outgoing.Payload{}.New().Load(message)
+	select {
+	case s.send <- p:
+	case <-time.After(2 * time.Millisecond):
+		log.Println("websocket.messageChannel send timeout")
+	}
+	return nil
+}

sam/websocket/outgoing/error.go

@@ -0,0 +1,9 @@
+package outgoing
+
+type (
+	Error struct {
+		Message string `json:"m"`
+	}
+)
+
+func (*Error) valid() bool { return true }

sam/websocket/outgoing/messages.go

@@ -0,0 +1,29 @@
+package outgoing
+
+type (
+	Message struct {
+		ID        string `json:"id"`
+		ChannelID string `json:"cid""`
+		Message   string `json:"m"`
+		Type      string `json:"t"`
+		ReplyTo   string `json:"rid"`
+		UserID    string `json:"uid"`
+	}
+
+	Messages []*Message
+
+	MessageUpdate struct {
+		ID      string `json:"id"`
+		Message string `json:"m"`
+	}
+
+	MessageDelete struct {
+		ID string `json:"id"`
+	}
+)
+
+func (*Message) valid() bool  { return true }
+func (*Messages) valid() bool { return true }
+
+func (*MessageUpdate) valid() bool { return true }
+func (*MessageDelete) valid() bool { return true }

sam/websocket/outgoing/payload.go

@@ -5,13 +5,21 @@ import (
 	"time"
 )
 
-type Payload struct {
-	*Error   `json:"error,omitempty"`
-	*Message `json:"m"`
+type (
+	Payload struct {
+		*Error         `json:"error,omitempty"`
+		*Message       `json:"m,omitempty"`
+		*MessageDelete `json:"md,omitempty"`
+		*MessageUpdate `json:"mu,omitempty"`
+		*Messages      `json:"ms,omitempty"`
 
-	// @todo: implement outgoing message types
-	timestamp time.Time
-}
+		// @todo: implement outgoing message types
+		timestamp time.Time
+	}
+	PayloadType interface {
+		valid() bool
+	}
+)
 
 func (p *Payload) Load(payload PayloadType) *Payload {
 	switch val := payload.(type) {
@@ -19,6 +27,12 @@ func (p *Payload) Load(payload PayloadType) *Payload {
 		p.Error = val
 	case *Message:
 		p.Message = val
+	case *Messages:
+		p.Messages = val
+	case *MessageDelete:
+		p.MessageDelete = val
+	case *MessageUpdate:
+		p.MessageUpdate = val
 	default:
 		panic(fmt.Sprintf("Unknown/unsupported Payload type: %T", val))
 	}

sam/websocket/outgoing/types.go

@@ -1,23 +0,0 @@
-package outgoing
-
-type (
-	PayloadType interface {
-		valid() bool
-	}
-
-	Error struct {
-		Message string `json:"m"`
-	}
-
-	Message struct {
-		ID        string `json:"id"`
-		ChannelID string `json:"cid""`
-		Message   string `json:"m"`
-		Type      string `json:"t"`
-		ReplyTo   string `json:"rid"`
-		UserID    string `json:"uid"`
-	}
-)
-
-func (*Error) valid() bool   { return true }
-func (*Message) valid() bool { return true }

sam/websocket/payload.go

@@ -16,3 +16,12 @@ func payloadFromMessage(msg *types.Message) *outgoing.Message {
 		ReplyTo:   strconv.FormatUint(msg.ReplyTo, 10),
 	}
 }
+
+func payloadFromMessages(msg []*types.Message) *outgoing.Messages {
+	msgs := make([]*outgoing.Message, len(msg))
+	for k, m := range msg {
+		msgs[k] = payloadFromMessage(m)
+	}
+	retval := outgoing.Messages(msgs)
+	return &retval
+}

sam/websocket/store.go

@@ -1,7 +1,6 @@
 package websocket
 
 import (
-	"github.com/crusttech/crust/sam/websocket/outgoing"
 	"github.com/titpetric/factory"
 	"sync"
 )
@@ -32,6 +31,14 @@ func (s *Store) Save(session *Session) *Session {
 	return session
 }
 
+func (s *Store) Walk(callback func(*Session)) {
+	s.RLock()
+	defer s.RUnlock()
+	for _, sess := range s.Sessions {
+		callback(sess)
+	}
+}
+
 func (s *Store) Get(id uint64) *Session {
 	s.RLock()
 	defer s.RUnlock()
@@ -43,13 +50,3 @@ func (s *Store) Delete(id uint64) {
 	defer s.Unlock()
 	delete(s.Sessions, id)
 }
-
-func (s *Store) MessageFanout(messages ...outgoing.PayloadType) {
-	// @todo this should probably implement some logic behind...
-	for _, message := range messages {
-		p := outgoing.Payload{}.New().Load(message)
-		for _, sess := range s.Sessions {
-			sess.send <- p
-		}
-	}
-}

sam/websocket/subscriptions.go

@@ -7,6 +7,8 @@ import (
 type (
 	// A subscription holds a "channel" that the user is joined to
 	Subscription struct {
+		// for tracking new messages
+		lastCommentID string
 	}
 
 	// A list of all user-joined channels

sam/websocket/util.go

@@ -8,6 +8,10 @@ import (
 
 var truthy = regexp.MustCompile("^\\s*(t(rue)?|y(es)?|1)\\s*$")
 
+func uint64toa(i uint64) string {
+	return strconv.FormatUint(i, 10)
+}
+
 // parseInt64 parses an string to int64
 func parseInt64(s string) int64 {
 	if s == "" {