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

import (
	"errors"
	"expvar"
	"fmt"
	"sync"

	log "github.com/golang/glog"

	"ipe/channel"
	"ipe/connection"
	"ipe/events"
	"ipe/subscription"
)

// Application represents a Pusher application
type Application struct {
	sync.RWMutex

	Name       string
	AppID      string
	Key        string
	Secret     string
	OnlySSL    bool
	Enabled    bool
	UserEvents bool
	WebHooks   bool
	URLWebHook string

	channels    map[string]*channel.Channel
	connections map[string]*connection.Connection

	Stats *expvar.Map `json:"-"`
}

// NewApplication returns a new Application
func NewApplication(
	name,
	appID,
	key,
	secret string,
	onlySSL,
	enabled,
	userEvents,
	webHooks bool,
	webHookURL string,
) *Application {

	a := &Application{
		Name:       name,
		AppID:      appID,
		Key:        key,
		Secret:     secret,
		OnlySSL:    onlySSL,
		Enabled:    enabled,
		UserEvents: userEvents,
		WebHooks:   webHooks,
		URLWebHook: webHookURL,
	}

	a.connections = make(map[string]*connection.Connection)
	a.channels = make(map[string]*channel.Channel)
	a.Stats = expvar.NewMap(fmt.Sprintf("%s (%s)", a.Name, a.AppID))

	return a
}

// Channels returns the full list of channels
func (a *Application) Channels() []*channel.Channel {
	a.RLock()
	defer a.RUnlock()

	var channels []*channel.Channel

	for _, c := range a.channels {
		channels = append(channels, c)
	}

	return channels
}

// PresenceChannels Only Presence channels
func (a *Application) PresenceChannels() []*channel.Channel {
	a.RLock()
	defer a.RUnlock()

	var channels []*channel.Channel

	for _, c := range a.channels {
		if c.IsPresence() {
			channels = append(channels, c)
		}
	}

	return channels
}

// PrivateChannels Only Private channels
func (a *Application) PrivateChannels() []*channel.Channel {
	a.RLock()
	defer a.RUnlock()

	var channels []*channel.Channel

	for _, c := range a.channels {
		if c.IsPrivate() {
			channels = append(channels, c)
		}
	}

	return channels
}

// PublicChannels Only Public channels
func (a *Application) PublicChannels() []*channel.Channel {
	a.RLock()
	defer a.RUnlock()

	var channels []*channel.Channel

	for _, c := range a.channels {
		if c.IsPublic() {
			channels = append(channels, c)
		}
	}

	return channels
}

// Disconnect Socket
func (a *Application) Disconnect(socketID string) {
	log.Infof("disconnecting socket %+v", socketID)

	conn, err := a.FindConnection(socketID)

	if err != nil {
		log.Infof("socket not found, %+v", err)
		return
	}

	// Unsubscribe from channels
	for _, c := range a.channels {
		if c.IsSubscribed(conn) {
			if err := c.Unsubscribe(conn); err != nil {
				log.Errorf("error while calling Channel.Unsubscribe, %+v", err)
				continue
			}
		}
	}

	// Remove from Application
	a.Lock()
	_, exists := a.connections[conn.SocketID]
	a.Unlock()

	if !exists {
		return
	}

	a.Lock()
	delete(a.connections, conn.SocketID)
	a.Unlock()

	a.Stats.Add("TotalConnections", -1)
}

// Connect a new Subscriber
func (a *Application) Connect(conn *connection.Connection) {
	log.Infof("adding a new Connection %s to Application %s", conn.SocketID, a.Name)
	a.Lock()
	defer a.Unlock()

	a.connections[conn.SocketID] = conn

	a.Stats.Add("TotalConnections", 1)
}

// FindConnection Find a Connection on this Application
func (a *Application) FindConnection(socketID string) (*connection.Connection, error) {
	a.RLock()
	defer a.RUnlock()

	conn, exists := a.connections[socketID]

	if exists {
		return conn, nil
	}

	return nil, errors.New("connection not found")
}

// RemoveChannel removes the Channel from Application
func (a *Application) RemoveChannel(c *channel.Channel) {
	log.Infof("remove the Channel %s from Application %s", c.ID, a.Name)
	a.Lock()
	defer a.Unlock()

	delete(a.channels, c.ID)

	if c.IsPresence() {
		a.Stats.Add("TotalPresenceChannels", -1)
	}

	if c.IsPrivate() {
		a.Stats.Add("TotalPrivateChannels", -1)
	}

	if c.IsPublic() {
		a.Stats.Add("TotalPublicChannels", -1)
	}

	a.Stats.Add("TotalChannels", -1)
}

// AddChannel Add a new Channel to this APP
func (a *Application) AddChannel(c *channel.Channel) {
	log.Infof("adding a new Channel %s to Application %s", c.ID, a.Name)

	a.Lock()
	defer a.Unlock()

	a.channels[c.ID] = c

	if c.IsPresence() {
		a.Stats.Add("TotalPresenceChannels", 1)
	}

	if c.IsPrivate() {
		a.Stats.Add("TotalPrivateChannels", 1)
	}

	if c.IsPublic() {
		a.Stats.Add("TotalPublicChannels", 1)
	}

	a.Stats.Add("TotalChannels", 1)
}

// FindOrCreateChannelByChannelID Returns a Channel from this Application
// If not found then the Channel is created and added to this Application
func (a *Application) FindOrCreateChannelByChannelID(n string) *channel.Channel {
	c, err := a.FindChannelByChannelID(n)

	if err != nil {
		c = channel.New(
			n,
			channel.WithChannelOccupiedListener(func(c *channel.Channel, s *subscription.Subscription) {
				a.TriggerChannelOccupiedHook(c)
			}),
			channel.WithChannelVacatedListener(func(c *channel.Channel, s *subscription.Subscription) {
				a.TriggerChannelVacatedHook(c)
			}),
			channel.WithMemberAddedListener(func(c *channel.Channel, s *subscription.Subscription) {
				a.TriggerMemberAddedHook(c, s)
			}),
			channel.WithMemberRemovedListener(func(c *channel.Channel, s *subscription.Subscription) {
				a.TriggerMemberRemovedHook(c, s)
			}),
			channel.WithClientEventListener(func(c *channel.Channel, s *subscription.Subscription, event string, data interface{}) {
				a.TriggerClientEventHook(c, s, event, data)
			}),
		)
		a.AddChannel(c)
	}

	return c
}

// FindChannelByChannelID Find the Channel by Channel ID
func (a *Application) FindChannelByChannelID(n string) (*channel.Channel, error) {
	a.RLock()
	defer a.RUnlock()

	c, exists := a.channels[n]

	if exists {
		return c, nil
	}

	return nil, errors.New("channel does not exists")
}

// Publish an event into the channel
// skip the ignore connection
func (a *Application) Publish(c *channel.Channel, event events.Raw, ignore string) error {
	a.Stats.Add("TotalUniqueMessages", 1)

	return c.Publish(event, ignore)
}

// Unsubscribe unsubscribe the given connection from the channel
// remove the channel from the application if it is empty
func (a *Application) Unsubscribe(c *channel.Channel, conn *connection.Connection) error {
	err := c.Unsubscribe(conn)
	if err != nil {
		return err
	}

	if !c.IsOccupied() {
		a.RemoveChannel(c)
	}

	return nil
}

// Subscribe the connection into the given channel
func (a *Application) Subscribe(c *channel.Channel, conn *connection.Connection, data string) error {
	return c.Subscribe(conn, data)
}