From 5043b05c7ba3a7b9a82e07048daa3bc3e1ec58df Mon Sep 17 00:00:00 2001 From: claudemiro Date: Wed, 14 Jan 2015 13:19:11 -0300 Subject: [PATCH] Triggering Presence channels events. --- app.go | 5 +- auth.go | 3 ++ conn.go | 123 +++++++++++++++++++++++++++++++------------------- events.go | 38 ++++++++++++---- rest.go | 11 +++-- webhooks.go | 12 ++--- websockets.go | 6 +-- 7 files changed, 126 insertions(+), 72 deletions(-) diff --git a/app.go b/app.go index 6fe97f5..d8fa5d2 100644 --- a/app.go +++ b/app.go @@ -102,6 +102,7 @@ func (a *App) Disconnect(socketID string) { defer a.Unlock() _, exists := a.Subscribers[s.SocketID] + if !exists { return } @@ -192,6 +193,6 @@ func (a *App) Unsubscribe(c *Channel, s *Subscriber) error { return c.Unsubscribe(a, s) } -func (a *App) Subscribe(c *Channel, s *Subscriber, data string) { - c.Subscribe(a, s, data) +func (a *App) Subscribe(c *Channel, s *Subscriber, data string) error { + return c.Subscribe(a, s, data) } diff --git a/auth.go b/auth.go index 220015c..fe280d0 100644 --- a/auth.go +++ b/auth.go @@ -10,6 +10,7 @@ import ( "sort" "strings" + log "github.com/golang/glog" "github.com/gorilla/mux" "github.com/dimiro1/ipe/utils" @@ -52,6 +53,7 @@ func RestAuthenticationHandler(h http.Handler) http.Handler { app, err := Conf.GetAppByAppID(appID) if err != nil { + log.Error(err) http.Error(w, "Not authorized", http.StatusUnauthorized) return } @@ -68,6 +70,7 @@ func RestAuthenticationHandler(h http.Handler) http.Handler { if utils.HashMAC([]byte(toSign), []byte(app.Secret)) == signature { h.ServeHTTP(w, r) } else { + log.Error("Not authorized") http.Error(w, "Not authorized", http.StatusUnauthorized) } }) diff --git a/conn.go b/conn.go index 7f8cd74..a67ea0b 100644 --- a/conn.go +++ b/conn.go @@ -11,22 +11,12 @@ import ( "sync" "time" - "strconv" - log "github.com/golang/glog" "github.com/gorilla/websocket" ) -// This mutex is used to sync the generation of the new ID -var mutex = &sync.Mutex{} - -// This variable store the current user id -// Every call to newID this variable is incremented -var currentID = 0 - // A subscriber type Subscriber struct { - Id string SocketID string Socket *websocket.Conn } @@ -34,6 +24,7 @@ type Subscriber struct { // A Channel Subscription type Subscription struct { Subscriber *Subscriber + Id string Data string } @@ -88,36 +79,69 @@ func (c *Channel) TotalUsers() int { } // Add a new subscriber to the channel -func (c *Channel) Subscribe(a *App, s *Subscriber, data string) { +func (c *Channel) Subscribe(a *App, s *Subscriber, channelData string) error { log.Infof("Subscribing %s to channel %s", s.SocketID, c.ChannelID) c.Lock() - c.Subscriptions[s.SocketID] = NewSubscription(s, data) - c.Unlock() + defer c.Unlock() + + subscription := NewSubscription(s, channelData) + c.Subscriptions[s.SocketID] = subscription if c.IsPresence() { - // Publish pusher_internal:member_added - Para todos + + // User Info Data + var info struct { + UserID string `json:"user_id"` + UserInfo json.RawMessage `json:"user_info"` + } + + log.Infof("%+v", channelData) + + if err := json.Unmarshal([]byte(channelData), &info); err != nil { + log.Error(err) + return err + } + + js, err := info.UserInfo.MarshalJSON() + + if err != nil { + log.Error(err) + return err + } + + // Update the Subscription + subscription.Id = info.UserID + subscription.Data = string(js) + + // Publish pusher_internal:member_added + c.PublishMemberAddedEvent(a, channelData, subscription) + // WebHook - a.TriggerMemberAddedHook(c, s) + a.TriggerMemberAddedHook(c, subscription) // pusher_internal:subscription_succeeded data := make(map[string]SubscriptionSucceeedEventPresenceData, 1) data["presence"] = NewSubscriptionSucceedEventPresenceData(c) - js, err := json.Marshal(data) + js, err = json.Marshal(data) + if err != nil { log.Error(err) + return err } - if err := s.Publish(NewSubscriptionSucceededEvent(c.ChannelID, string(js))); err != nil { - log.Error(err) - } + s.Publish(NewSubscriptionSucceededEvent(c.ChannelID, string(js))) + } else { + s.Publish(NewSubscriptionSucceededEvent(c.ChannelID, "{}")) } // WebHook if c.TotalSubscriptions() == 1 { a.TriggerChannelOccupiedHook(c) } + + return nil } // IsSubscribed check if the user is subscribed @@ -134,7 +158,7 @@ func (c *Channel) Unsubscribe(a *App, s *Subscriber) error { c.Lock() defer c.Unlock() - _, exists := c.Subscriptions[s.SocketID] + subscription, exists := c.Subscriptions[s.SocketID] if !exists { return errors.New("Subscription not found") @@ -144,8 +168,10 @@ func (c *Channel) Unsubscribe(a *App, s *Subscriber) error { if c.IsPresence() { // Publish pusher_internal:member_removed + c.PublishMemberRemovedEvent(a, subscription) + // Webhook - a.TriggerMemberRemovedHook(c, s) + a.TriggerMemberRemovedHook(c, subscription) } // WebHook @@ -163,23 +189,33 @@ func NewChannel(channelID string) *Channel { return &Channel{ChannelID: channelID, CreatedAt: time.Now(), Subscriptions: make(map[string]*Subscription)} } -// This function generate a sequencial ID -func newID() (string, int) { - mutex.Lock() - defer mutex.Unlock() - - currentID += 1 - - return strconv.Itoa(currentID), currentID -} - // Create a new Subscriber func NewSubscriber(socketID string, s *websocket.Conn) *Subscriber { - id, _ := newID() + log.Infof("Creating a new Subscriber %+v", socketID) - log.Infof("Creating a new Subscriber %+v with id %s", socketID, id) + return &Subscriber{SocketID: socketID, Socket: s} +} - return &Subscriber{Id: id, SocketID: socketID, Socket: s} +// Publish a MemberAddedEvent to all subscriptions +func (c *Channel) PublishMemberAddedEvent(a *App, data string, subscription *Subscription) error { + for _, subs := range c.Subscriptions { + if subs != subscription { + subs.Subscriber.Publish(NewMemberAddedEvent(c.ChannelID, data)) + } + } + + return nil +} + +// Publish a MemberRemovedEvent to all subscriptions +func (c *Channel) PublishMemberRemovedEvent(a *App, subscription *Subscription) error { + for _, subs := range c.Subscriptions { + if subs != subscription { + subs.Subscriber.Publish(NewMemberRemovedEvent(c.ChannelID, subscription)) + } + } + + return nil } // Publish messages to all Subscribers @@ -200,11 +236,7 @@ func (c *Channel) Publish(a *App, event RawEvent, ignore string) error { for _, subs := range c.Subscriptions { if subs.Subscriber.SocketID != ignore { - js := NewResponseEvent(event.Event, event.Channel, v) - - if err := subs.Subscriber.Publish(js); err != nil { - continue - } + subs.Subscriber.Publish(NewResponseEvent(event.Event, event.Channel, v)) } else { // Webhook if strings.HasPrefix(event.Event, "client-") { @@ -217,11 +249,10 @@ func (c *Channel) Publish(a *App, event RawEvent, ignore string) error { } // Publish the message to websocket atached to this client -func (s *Subscriber) Publish(m interface{}) error { - if err := s.Socket.WriteJSON(m); err != nil { - log.Errorf("Error sending message to subscriber %+v, %s", s, err) - return err - } - - return nil +func (s *Subscriber) Publish(m interface{}) { + go func() { + if err := s.Socket.WriteJSON(m); err != nil { + log.Errorf("Error sending message to subscriber %+v, %s", s, err) + } + }() } diff --git a/events.go b/events.go index fa5b6b2..6e8449d 100644 --- a/events.go +++ b/events.go @@ -4,7 +4,11 @@ package main -import "encoding/json" +import ( + "encoding/json" + + log "github.com/golang/glog" +) // { // "event": "pusher:subscribe", @@ -86,25 +90,29 @@ func NewSubscriptionSucceededEvent(channel, data string) SubscriptionSucceededEv // } // }" type SubscriptionSucceeedEventPresenceData struct { - Ids []string `json:"ids"` - Hash map[string]string `json:"hash"` - count int `json:"count"` + Ids []string `json:"ids"` + Hash map[string]interface{} `json:"hash"` + Count int `json:"count"` } func NewSubscriptionSucceedEventPresenceData(c *Channel) SubscriptionSucceeedEventPresenceData { event := SubscriptionSucceeedEventPresenceData{} var ids []string - hash := make(map[string]string, c.TotalSubscriptions()) + hash := make(map[string]interface{}, c.TotalSubscriptions()) for _, s := range c.Subscriptions { - ids = append(ids, s.Subscriber.SocketID) - hash[s.Subscriber.SocketID] = s.Data + // Do you have any other idea? + var js interface{} + json.Unmarshal([]byte(s.Data), &js) + + hash[s.Id] = js + ids = append(ids, s.Id) } event.Ids = ids event.Hash = hash - event.count = c.TotalSubscriptions() + event.Count = c.TotalSubscriptions() return event } @@ -233,8 +241,18 @@ type MemberRemovedEvent struct { Data string `json:"data"` } -func NewMemberRemovedEvent(channel, data string) MemberRemovedEvent { - return MemberRemovedEvent{Event: "pusher_internal:member_removed", Channel: channel, Data: data} +func NewMemberRemovedEvent(channel string, s *Subscription) MemberRemovedEvent { + data, err := json.Marshal(struct { + UserID string `json:"user_id"` + }{ + UserID: s.Id, + }) + + if err != nil { + log.Error(err) + } + + return MemberRemovedEvent{Event: "pusher_internal:member_removed", Channel: channel, Data: string(data)} } // { diff --git a/rest.go b/rest.go index 246979a..5463a7a 100644 --- a/rest.go +++ b/rest.go @@ -157,7 +157,10 @@ func GetChannels(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json;charset=UTF-8") - if err := json.NewEncoder(w).Encode(channels); err != nil { + js := make(map[string]interface{}, 1) + js["channels"] = channels + + if err := json.NewEncoder(w).Encode(js); err != nil { log.Error(err) http.Error(w, "Internal server error", http.StatusInternalServerError) } @@ -257,8 +260,8 @@ func GetChannel(w http.ResponseWriter, r *http.Request) { // Example: // { // "users": [ -// { "id": 1 }, -// { "id": 2 } +// { "id": "1" }, +// { "id": "2" } // ] // } // @@ -298,7 +301,7 @@ func GetChannelUsers(w http.ResponseWriter, r *http.Request) { for _, s := range channel.Subscriptions { users = append(users, struct { Id string `json:"id"` - }{s.Subscriber.Id}) + }{s.Id}) } result["users"] = users diff --git a/webhooks.go b/webhooks.go index 60728ad..bed9931 100644 --- a/webhooks.go +++ b/webhooks.go @@ -55,11 +55,11 @@ func NewChannelVacatedHook(channel *Channel) HookEvent { return HookEvent{Name: "channel_vacated", Channel: channel.ChannelID} } -func NewMemberAddedHook(channel *Channel, s *Subscriber) HookEvent { +func NewMemberAddedHook(channel *Channel, s *Subscription) HookEvent { return HookEvent{Name: "member_added", Channel: channel.ChannelID, UserId: s.Id} } -func NewMemberRemovedHook(channel *Channel, s *Subscriber) HookEvent { +func NewMemberRemovedHook(channel *Channel, s *Subscription) HookEvent { return HookEvent{Name: "member_removed", Channel: channel.ChannelID, UserId: s.Id} } @@ -93,7 +93,7 @@ func (a *App) TriggerClientEventHook(c *Channel, s *Subscription, client_event s event := NewClientHook(c, s, client_event, data) if c.IsPresence() { - event.UserId = s.Subscriber.Id + event.UserId = s.Id } triggerHook(event.Name, a, c, event) @@ -104,7 +104,7 @@ func (a *App) TriggerClientEventHook(c *Channel, s *Subscription, client_event s // "channel": "presence-your_channel_name", // "user_id": "a_user_id" // } -func (a *App) TriggerMemberAddedHook(c *Channel, s *Subscriber) { +func (a *App) TriggerMemberAddedHook(c *Channel, s *Subscription) { event := NewMemberAddedHook(c, s) triggerHook(event.Name, a, c, event) } @@ -114,14 +114,14 @@ func (a *App) TriggerMemberAddedHook(c *Channel, s *Subscriber) { // "channel": "presence-your_channel_name", // "user_id": "a_user_id" // } -func (a *App) TriggerMemberRemovedHook(c *Channel, s *Subscriber) { +func (a *App) TriggerMemberRemovedHook(c *Channel, s *Subscription) { event := NewMemberRemovedHook(c, s) triggerHook(event.Name, a, c, event) } func triggerHook(name string, app *App, c *Channel, event HookEvent) { if !app.WebHooks { - log.Infof("Checking webhooks enabled for app: %+v", app) + log.Infof("Checking webhooks enabled for app: %s", app.Name) return } diff --git a/websockets.go b/websockets.go index 75237da..a6484a8 100644 --- a/websockets.go +++ b/websockets.go @@ -131,7 +131,6 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses } expectedAuthKey := fmt.Sprintf("%s:%s", app.Key, utils.HashMAC([]byte(strings.Join(toSign, ":")), []byte(app.Secret))) - if subscribeEvent.Data.Auth != expectedAuthKey { emitWSError(NewGenericError(fmt.Sprintf("Auth value for subscription to %s is invalid", channelName)), conn) continue @@ -139,11 +138,10 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses } channel := app.FindOrCreateChannelByChannelID(channelName) - app.Subscribe(channel, subscriber, subscribeEvent.Data.ChannelData) + log.Info(subscribeEvent.Data.ChannelData) - if err := conn.WriteJSON(NewSubscriptionSucceededEvent(channel.ChannelID, "{}")); err != nil { + if err := app.Subscribe(channel, subscriber, subscribeEvent.Data.ChannelData); err != nil { emitWSError(NewGenericReconnectImmediatelyError(), conn) - break } case "pusher:unsubscribe": unsubscribeEvent := UnsubscribeEvent{}