diff --git a/app.go b/app.go index d8fa5d2..307b9b4 100644 --- a/app.go +++ b/app.go @@ -28,14 +28,14 @@ type App struct { URLWebHook string Channels map[string]*Channel `json:"-"` - Subscribers map[string]*Subscriber `json:"-"` + Connections map[string]*Connection `json:"-"` Stats *expvar.Map `json:"-"` } -// Alloc memory for Subscribers and Channels +// Alloc memory for Connections and Channels func (a *App) Init() { - a.Subscribers = make(map[string]*Subscriber) + a.Connections = make(map[string]*Connection) a.Channels = make(map[string]*Channel) a.Stats = expvar.NewMap(fmt.Sprintf("%s (%s)", a.Name, a.AppID)) } @@ -83,7 +83,7 @@ func (a *App) PublicChannels() []*Channel { func (a *App) Disconnect(socketID string) { log.Infof("Disconnecting socket %+v", socketID) - s, err := a.FindSubscriber(socketID) + conn, err := a.FindConnection(socketID) if err != nil { log.Infof("Socket not found, %+v", err) @@ -92,8 +92,8 @@ func (a *App) Disconnect(socketID string) { // Unsubscribe from channels for _, c := range a.Channels { - if c.IsSubscribed(s) { - c.Unsubscribe(a, s) + if c.IsSubscribed(conn) { + c.Unsubscribe(a, conn) } } @@ -101,38 +101,37 @@ func (a *App) Disconnect(socketID string) { a.Lock() defer a.Unlock() - _, exists := a.Subscribers[s.SocketID] + _, exists := a.Connections[conn.SocketID] if !exists { return } - delete(a.Subscribers, s.SocketID) + delete(a.Connections, conn.SocketID) - a.Stats.Add("TotalSubscribers", -1) + a.Stats.Add("TotalConnections", -1) } // Connect a new Subscriber -func (a *App) Connect(s *Subscriber) { - log.Infof("Adding a new Subscriber %s to app %s", s.SocketID, a.Name) +func (a *App) Connect(conn *Connection) { + log.Infof("Adding a new Connection %s to app %s", conn.SocketID, a.Name) a.Lock() defer a.Unlock() - a.Subscribers[s.SocketID] = s + a.Connections[conn.SocketID] = conn - a.Stats.Add("TotalSubscribers", 1) + a.Stats.Add("TotalConnections", 1) } -// Find a Subscriber on this app -func (a *App) FindSubscriber(socketID string) (*Subscriber, error) { - - s, exists := a.Subscribers[socketID] +// Find a Connection on this app +func (a *App) FindConnection(socketID string) (*Connection, error) { + conn, exists := a.Connections[socketID] if exists { - return s, nil + return conn, nil } - return nil, errors.New("Subscriber not found") + return nil, errors.New("Connection not found") } // Add a new Channel to this APP @@ -141,6 +140,7 @@ func (a *App) AddChannel(c *Channel) { a.Lock() defer a.Unlock() + a.Channels[c.ChannelID] = c if c.IsPresence() { @@ -173,7 +173,6 @@ func (a *App) FindOrCreateChannelByChannelID(n string) *Channel { // Find the channel by channel ID func (a *App) FindChannelByChannelID(n string) (*Channel, error) { - c, exists := a.Channels[n] if exists { @@ -189,10 +188,10 @@ func (a *App) Publish(c *Channel, event RawEvent, ignore string) error { return c.Publish(a, event, ignore) } -func (a *App) Unsubscribe(c *Channel, s *Subscriber) error { - return c.Unsubscribe(a, s) +func (a *App) Unsubscribe(c *Channel, conn *Connection) error { + return c.Unsubscribe(a, conn) } -func (a *App) Subscribe(c *Channel, s *Subscriber, data string) error { - return c.Subscribe(a, s, data) +func (a *App) Subscribe(c *Channel, conn *Connection, data string) error { + return c.Subscribe(a, conn, data) } diff --git a/conn.go b/conn.go index 16b64c6..4a075aa 100644 --- a/conn.go +++ b/conn.go @@ -15,25 +15,20 @@ import ( "github.com/gorilla/websocket" ) -// A subscriber -type Subscriber struct { +// An User Connection +type Connection struct { SocketID string Socket *websocket.Conn } // A Channel Subscription type Subscription struct { - Subscriber *Subscriber + Connection *Connection Id string Data string } -// Create a new Subscription -func NewSubscription(subscriber *Subscriber, data string) *Subscription { - return &Subscription{Subscriber: subscriber, Data: data} -} - -// A channel +// A Channel type Channel struct { sync.Mutex @@ -42,6 +37,11 @@ type Channel struct { Subscriptions map[string]*Subscription } +// Create a new Subscription +func NewSubscription(conn *Connection, data string) *Subscription { + return &Subscription{Connection: conn, Data: data} +} + // Return true if the channel has at least one subscriber func (c *Channel) IsOccupied() bool { return c.TotalSubscriptions() > 0 @@ -84,17 +84,16 @@ func (c *Channel) TotalUsers() int { } // Add a new subscriber to the channel -func (c *Channel) Subscribe(a *App, s *Subscriber, channelData string) error { - log.Infof("Subscribing %s to channel %s", s.SocketID, c.ChannelID) +func (c *Channel) Subscribe(a *App, conn *Connection, channelData string) error { + log.Infof("Subscribing %s to channel %s", conn.SocketID, c.ChannelID) c.Lock() defer c.Unlock() - subscription := NewSubscription(s, channelData) - c.Subscriptions[s.SocketID] = subscription + subscription := NewSubscription(conn, channelData) + c.Subscriptions[conn.SocketID] = subscription if c.IsPresence() { - // User Info Data var info struct { UserID string `json:"user_id"` @@ -121,7 +120,6 @@ func (c *Channel) Subscribe(a *App, s *Subscriber, channelData string) error { // Publish pusher_internal:member_added c.PublishMemberAddedEvent(a, channelData, subscription) - // WebHook a.TriggerMemberAddedHook(c, subscription) @@ -136,9 +134,9 @@ func (c *Channel) Subscribe(a *App, s *Subscriber, channelData string) error { return err } - s.Publish(NewSubscriptionSucceededEvent(c.ChannelID, string(js))) + conn.Publish(NewSubscriptionSucceededEvent(c.ChannelID, string(js))) } else { - s.Publish(NewSubscriptionSucceededEvent(c.ChannelID, "{}")) + conn.Publish(NewSubscriptionSucceededEvent(c.ChannelID, "{}")) } // WebHook @@ -150,31 +148,30 @@ func (c *Channel) Subscribe(a *App, s *Subscriber, channelData string) error { } // IsSubscribed check if the user is subscribed -func (c *Channel) IsSubscribed(s *Subscriber) bool { - _, exists := c.Subscriptions[s.SocketID] +func (c *Channel) IsSubscribed(conn *Connection) bool { + _, exists := c.Subscriptions[conn.SocketID] return exists } // Remove the subscriber from the channel // It destroy the channel if the channels does not have any subscribers. -func (c *Channel) Unsubscribe(a *App, s *Subscriber) error { - log.Infof("Unsubscribing %s from channel %s", s.SocketID, c.ChannelID) +func (c *Channel) Unsubscribe(a *App, conn *Connection) error { + log.Infof("Unsubscribing %s from channel %s", conn.SocketID, c.ChannelID) c.Lock() defer c.Unlock() - subscription, exists := c.Subscriptions[s.SocketID] + subscription, exists := c.Subscriptions[conn.SocketID] if !exists { return errors.New("Subscription not found") } - delete(c.Subscriptions, s.SocketID) + delete(c.Subscriptions, conn.SocketID) if c.IsPresence() { // Publish pusher_internal:member_removed c.PublishMemberRemovedEvent(a, subscription) - // Webhook a.TriggerMemberRemovedHook(c, subscription) } @@ -195,17 +192,17 @@ func NewChannel(channelID string) *Channel { } // Create a new Subscriber -func NewSubscriber(socketID string, s *websocket.Conn) *Subscriber { +func NewConnection(socketID string, s *websocket.Conn) *Connection { log.Infof("Creating a new Subscriber %+v", socketID) - return &Subscriber{SocketID: socketID, Socket: s} + return &Connection{SocketID: socketID, Socket: s} } // Publish a MemberAddedEvent to all subscriptions func (c *Channel) PublishMemberAddedEvent(a *App, data string, subscription *Subscription) { for _, subs := range c.Subscriptions { if subs != subscription { - subs.Subscriber.Publish(NewMemberAddedEvent(c.ChannelID, data)) + subs.Connection.Publish(NewMemberAddedEvent(c.ChannelID, data)) } } } @@ -214,7 +211,7 @@ func (c *Channel) PublishMemberAddedEvent(a *App, data string, subscription *Sub func (c *Channel) PublishMemberRemovedEvent(a *App, subscription *Subscription) { for _, subs := range c.Subscriptions { if subs != subscription { - subs.Subscriber.Publish(NewMemberRemovedEvent(c.ChannelID, subscription)) + subs.Connection.Publish(NewMemberRemovedEvent(c.ChannelID, subscription)) } } } @@ -236,8 +233,8 @@ func (c *Channel) Publish(a *App, event RawEvent, ignore string) error { log.Infof("Publishing message %+v to channel %s", v, c.ChannelID) for _, subs := range c.Subscriptions { - if subs.Subscriber.SocketID != ignore { - subs.Subscriber.Publish(NewResponseEvent(event.Event, event.Channel, v)) + if subs.Connection.SocketID != ignore { + subs.Connection.Publish(NewResponseEvent(event.Event, event.Channel, v)) } else { // Webhook if strings.HasPrefix(event.Event, "client-") { @@ -250,10 +247,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{}) { +func (conn *Connection) Publish(m interface{}) { go func() { - if err := s.Socket.WriteJSON(m); err != nil { - log.Errorf("Error sending message to subscriber %+v, %s", s, err) + if err := conn.Socket.WriteJSON(m); err != nil { + log.Errorf("Error publishing message to connection %+v, %s", conn, err) } }() } diff --git a/rest.go b/rest.go index 5463a7a..4b75db2 100644 --- a/rest.go +++ b/rest.go @@ -255,7 +255,7 @@ func GetChannel(w http.ResponseWriter, r *http.Request) { } } -// Allowed only to presence-channels +// Allowed only for presence-channels // // Example: // { diff --git a/webhooks.go b/webhooks.go index 71c062c..a13caa8 100644 --- a/webhooks.go +++ b/webhooks.go @@ -64,7 +64,7 @@ func NewMemberRemovedHook(channel *Channel, s *Subscription) HookEvent { } func NewClientHook(channel *Channel, s *Subscription, event string, data interface{}) HookEvent { - return HookEvent{Name: "client_event", Channel: channel.ChannelID, Event: event, Data: data, SocketID: s.Subscriber.SocketID} + return HookEvent{Name: "client_event", Channel: channel.ChannelID, Event: event, Data: data, SocketID: s.Connection.SocketID} } // channel_occupied diff --git a/websockets.go b/websockets.go index a6484a8..8653399 100644 --- a/websockets.go +++ b/websockets.go @@ -52,11 +52,11 @@ func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessio } // Create the new Subscriber - subscriber := NewSubscriber(sessionID, conn) - app.Connect(subscriber) + connection := NewConnection(sessionID, conn) + app.Connect(connection) // Everything went fine. Huhu. - if err := conn.WriteJSON(NewConnectionEstablishedEvent(subscriber.SocketID)); err != nil { + if err := conn.WriteJSON(NewConnectionEstablishedEvent(connection.SocketID)); err != nil { return NewGenericReconnectImmediatelyError() } @@ -111,7 +111,7 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses break } - subscriber, err := app.FindSubscriber(sessionID) + connection, err := app.FindConnection(sessionID) if err != nil { emitWSError(NewGenericReconnectImmediatelyError(), conn) @@ -124,7 +124,7 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses isPrivate := strings.HasPrefix(channelName, "private-") if isPresence || isPrivate { - toSign := []string{subscriber.SocketID, channelName} + toSign := []string{connection.SocketID, channelName} if isPresence { toSign = append(toSign, subscribeEvent.Data.ChannelData) @@ -140,7 +140,7 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses channel := app.FindOrCreateChannelByChannelID(channelName) log.Info(subscribeEvent.Data.ChannelData) - if err := app.Subscribe(channel, subscriber, subscribeEvent.Data.ChannelData); err != nil { + if err := app.Subscribe(channel, connection, subscribeEvent.Data.ChannelData); err != nil { emitWSError(NewGenericReconnectImmediatelyError(), conn) } case "pusher:unsubscribe": @@ -150,10 +150,10 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses emitWSError(NewGenericReconnectImmediatelyError(), conn) } - subscriber, err := app.FindSubscriber(sessionID) + connection, err := app.FindConnection(sessionID) if err != nil { - emitWSError(NewGenericError(fmt.Sprintf("Could not find a subscriber with the id %s", sessionID)), conn) + emitWSError(NewGenericError(fmt.Sprintf("Could not find a connection with the id %s", sessionID)), conn) } channel, err := app.FindChannelByChannelID(unsubscribeEvent.Data.Channel) @@ -162,7 +162,7 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses emitWSError(NewGenericError(fmt.Sprintf("Could not find a channel with the id %s", unsubscribeEvent.Data.Channel)), conn) } - if err := app.Unsubscribe(channel, subscriber); err != nil { + if err := app.Unsubscribe(channel, connection); err != nil { emitWSError(NewGenericReconnectImmediatelyError(), conn) break }