ipe/conn.go

// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package main

import (
	"encoding/json"
	"errors"
	"strings"
	"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
}

// A Channel Subscription
type Subscription struct {
	Subscriber *Subscriber
	Data       string
}

// Create a new Subscription
func NewSubscription(subscriber *Subscriber, data string) *Subscription {
	return &Subscription{Subscriber: subscriber, Data: data}
}

// A channel
type Channel struct {
	sync.Mutex

	CreatedAt     time.Time
	ChannelID     string
	Subscriptions map[string]*Subscription
}

// Return true if the channel has at least one subscriber
func (c *Channel) IsOccupied() bool {
	return c.TotalSubscriptions() > 0
}

// Check if the type of the channel is presence or is private
func (c *Channel) IsPresenceOrPrivate() bool {
	return c.IsPresence() || c.IsPrivate()
}

// Check if the type of the channel is public
func (c *Channel) IsPublic() bool {
	return !c.IsPresenceOrPrivate()
}

// Check if the type of the channel is presence
func (c *Channel) IsPresence() bool {
	return strings.HasPrefix(c.ChannelID, "presence-")
}

// Check if the type of the channel is private
func (c *Channel) IsPrivate() bool {
	return strings.HasPrefix(c.ChannelID, "private-")
}

// Get the total of subscribers
func (c *Channel) TotalSubscriptions() int {
	return len(c.Subscriptions)
}

// Get the total of users.
// For now, totalUsers is equal to totalSubscribers
func (c *Channel) TotalUsers() int {
	return c.TotalSubscriptions()
}

// Add a new subscriber to the channel
func (c *Channel) Subscribe(a *App, s *Subscriber, data string) {
	log.Infof("Subscribing %s to channel %s", s.SocketID, c.ChannelID)

	c.Lock()
	c.Subscriptions[s.SocketID] = NewSubscription(s, data)
	c.Unlock()

	if c.IsPresence() {
		// Publish pusher_internal:member_added - Para todos
		// WebHook
		a.TriggerMemberAddedHook(c, s)

		// pusher_internal:subscription_succeeded
		data := make(map[string]SubscriptionSucceeedEventPresenceData, 1)
		data["presence"] = NewSubscriptionSucceedEventPresenceData(c)

		js, err := json.Marshal(data)
		if err != nil {
			log.Error(err)
		}

		if err := s.Publish(NewSubscriptionSucceededEvent(c.ChannelID, string(js))); err != nil {
			log.Error(err)
		}
	}

	// WebHook
	if c.TotalSubscriptions() == 1 {
		a.TriggerChannelOccupiedHook(c)
	}
}

// IsSubscribed check if the user is subscribed
func (c *Channel) IsSubscribed(s *Subscriber) bool {
	_, exists := c.Subscriptions[s.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)

	c.Lock()
	defer c.Unlock()

	_, exists := c.Subscriptions[s.SocketID]

	if !exists {
		return errors.New("Subscription not found")
	}

	delete(c.Subscriptions, s.SocketID)

	if c.IsPresence() {
		// Publish pusher_internal:member_removed
		// Webhook
		a.TriggerMemberRemovedHook(c, s)
	}

	// WebHook
	if c.TotalSubscriptions() == 0 {
		a.TriggerChannelVacatedHook(c)
	}

	return nil
}

// Create a new Channel
func NewChannel(channelID string) *Channel {
	log.Infof("Creating a new channel: %s", channelID)

	return &Channel{ChannelID: channelID, CreatedAt: time.Now(), Subscriptions: make(map[string]*Subscription)}
}

// This function generate a sequencial ID
func newID() string {
	mutex.Lock()
	defer mutex.Unlock()

	currentID += 1

	return strconv.Itoa(currentID)
}

// Create a new Subscriber
func NewSubscriber(socketID string, s *websocket.Conn) *Subscriber {
	id := newID()

	log.Infof("Creating a new Subscriber %+v with id %s", socketID, id)

	return &Subscriber{Id: id, SocketID: socketID, Socket: s}
}

// Publish messages to all Subscribers
func (c *Channel) Publish(a *App, event RawEvent, ignore string) error {
	b, err := event.Data.MarshalJSON()

	if err != nil {
		return err
	}

	var v interface{}

	if err := json.Unmarshal(b, &v); err != nil {
		return err
	}

	log.Infof("Publishing message %+v to channel %s", v, c.ChannelID)

	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
			}
		} else {
			// Webhook
			if strings.HasPrefix(event.Event, "client-") {
				a.TriggerClientEventHook(c, subs, event.Event, v)
			}
		}
	}

	return nil
}

// 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
}