ipe/channel/channel.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 channel

import (
	"encoding/json"
	"errors"
	"sync"
	"time"

	log "github.com/golang/glog"

	"ipe/connection"
	"ipe/events"
	"ipe/subscription"
	"ipe/utils"
)

// Option constructor function for Channel
type Option func(*Channel)

// ListenerFunc listener function
type ListenerFunc func(*Channel, *subscription.Subscription)

// ClientEventListenerFunc listener for client events
type ClientEventListenerFunc func(*Channel, *subscription.Subscription, string, interface{})

// Channel represents an application channel
type Channel struct {
	sync.RWMutex

	ID            string
	subscriptions map[string]*subscription.Subscription

	createdAt time.Time

	memberAddedListeners     []ListenerFunc
	memberRemovedListeners   []ListenerFunc
	channelOccupiedListeners []ListenerFunc
	channelVacatedListeners  []ListenerFunc
	clientEventListeners     []ClientEventListenerFunc
}

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

	c := &Channel{ID: channelID, createdAt: time.Now(), subscriptions: make(map[string]*subscription.Subscription)}

	for _, option := range options {
		option(c)
	}

	return c
}

// WithMemberAddedListener appends the given ListenerFunc into the memberAddedListeners list
func WithMemberAddedListener(f ListenerFunc) func(*Channel) {
	return func(c *Channel) {
		c.memberAddedListeners = append(c.memberAddedListeners, f)
	}
}

// WithMemberRemovedListener appends the given ListenerFunc into the memberRemovedListeners list
func WithMemberRemovedListener(f ListenerFunc) func(*Channel) {
	return func(c *Channel) {
		c.memberRemovedListeners = append(c.memberRemovedListeners, f)
	}
}

// WithChannelOccupiedListener appends the given ListenerFunc into the channelOccupiedListeners list
func WithChannelOccupiedListener(f ListenerFunc) func(*Channel) {
	return func(c *Channel) {
		c.channelOccupiedListeners = append(c.channelOccupiedListeners, f)
	}
}

// WithChannelVacatedListener appends the given ListenerFunc into the channelVacatedListeners list
func WithChannelVacatedListener(f ListenerFunc) func(*Channel) {
	return func(c *Channel) {
		c.channelVacatedListeners = append(c.channelVacatedListeners, f)
	}
}

// WithClientEventListener appends the given ListenerFunc into the clientEventListeners list
func WithClientEventListener(f ClientEventListenerFunc) func(*Channel) {
	return func(c *Channel) {
		c.clientEventListeners = append(c.clientEventListeners, f)
	}
}

// Subscriptions returns a slice of subscriptions
func (c *Channel) Subscriptions() []*subscription.Subscription {
	c.RLock()
	defer c.RUnlock()

	var subscriptions []*subscription.Subscription

	for _, sub := range c.subscriptions {
		subscriptions = append(subscriptions, sub)
	}

	return subscriptions
}

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

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

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

// IsPresence Check if the type of the Channel is presence
func (c *Channel) IsPresence() bool {
	return utils.IsPresenceChannel(c.ID)
}

// IsPrivate Check if the type of the Channel is private
func (c *Channel) IsPrivate() bool {
	return utils.IsPrivateChannel(c.ID)
}

// TotalSubscriptions Get the total of subscribers
func (c *Channel) TotalSubscriptions() int {
	c.RLock()
	defer c.RUnlock()

	return len(c.subscriptions)
}

// TotalUsers Get the total of users.
func (c *Channel) TotalUsers() int {
	c.RLock()
	defer c.RUnlock()

	total := make(map[string]int)

	for _, s := range c.subscriptions {
		total[s.ID]++
	}

	return len(total)
}

// Subscribe Add a new subscriber to the Channel
func (c *Channel) Subscribe(conn *connection.Connection, channelData string) error {
	log.Infof("Subscribing %s to Channel %s", conn.SocketID, c.ID)

	_subscription := subscription.New(conn, channelData)
	c.Lock()
	c.subscriptions[conn.SocketID] = _subscription
	c.Unlock()

	if c.IsPresence() {
		// 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
		}

		c.Lock()
		// Update the Subscription
		_subscription.ID = info.UserID
		_subscription.Data = string(js)
		c.Unlock()

		// Publish pusher_internal:member_added
		c.PublishMemberAddedEvent(channelData, _subscription)

		for _, hook := range c.memberAddedListeners {
			hook(c, _subscription)
		}

		// pusher_internal:subscription_succeeded
		data := make(map[string]events.SubscriptionSucceededPresenceData)
		data["presence"] = events.NewSubscriptionSucceedPresenceData(c.subscriptions)

		js, err = json.Marshal(data)

		if err != nil {
			log.Error(err)
			return err
		}

		conn.Publish(events.NewSubscriptionSucceeded(c.ID, string(js)))
	} else {
		conn.Publish(events.NewSubscriptionSucceeded(c.ID, "{}"))
	}

	if c.TotalSubscriptions() == 1 {
		for _, hook := range c.channelOccupiedListeners {
			hook(c, _subscription)
		}
	}

	return nil
}

// IsSubscribed check if the user is subscribed
func (c *Channel) IsSubscribed(conn *connection.Connection) bool {
	c.RLock()
	defer c.RUnlock()

	_, exists := c.subscriptions[conn.SocketID]
	return exists
}

// Unsubscribe Remove the subscriber from the Channel
// It destroy the Channel if the channels does not have any subscribers.
func (c *Channel) Unsubscribe(conn *connection.Connection) error {
	log.Infof("unsubscribe %s from Channel %s", conn.SocketID, c.ID)

	c.RLock()
	_subscription, exists := c.subscriptions[conn.SocketID]
	c.RUnlock()

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

	c.Lock()
	delete(c.subscriptions, conn.SocketID)
	c.Unlock()

	if c.IsPresence() {
		// Publish pusher_internal:member_removed
		c.PublishMemberRemovedEvent(_subscription)

		for _, hook := range c.memberRemovedListeners {
			hook(c, _subscription)
		}
	}

	if !c.IsOccupied() {
		for _, hook := range c.channelVacatedListeners {
			hook(c, _subscription)
		}
	}

	return nil
}

// PublishMemberAddedEvent Publish a MemberAddedEvent to all subscriptions
func (c *Channel) PublishMemberAddedEvent(data string, subscription *subscription.Subscription) {
	c.RLock()
	defer c.RUnlock()

	for _, subs := range c.subscriptions {
		if subs != subscription {
			subs.Connection.Publish(events.NewMemberAdded(c.ID, data))
		}
	}
}

// PublishMemberRemovedEvent Publish a MemberRemovedEvent to all subscriptions
func (c *Channel) PublishMemberRemovedEvent(subscription *subscription.Subscription) {
	c.RLock()
	defer c.RUnlock()

	for _, subs := range c.subscriptions {
		if subs != subscription {
			subs.Connection.Publish(events.NewMemberRemoved(c.ID, subscription.ID))
		}
	}
}

// Publish messages to all Subscribers
// skip the ignore connection
func (c *Channel) Publish(event events.Raw, ignore string) error {
	c.RLock()
	defer c.RUnlock()

	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.ID)

	for _, subs := range c.subscriptions {
		if subs.Connection.SocketID != ignore {
			subs.Connection.Publish(events.NewResponse(event.Event, event.Channel, v))
		} else {
			if utils.IsClientEvent(event.Event) {
				for _, hook := range c.clientEventListeners {
					hook(c, subs, event.Event, v)
				}
			}
		}
	}

	return nil
}