talksik/ipe
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Second Import

Browse Source
This commit is contained in:
claudemiro
2015-01-12 21:15:36 -03:00
parent ff2fcf3c54
commit e2412827cf
21 changed files with 920 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+35
View File
@@ -116,7 +116,42 @@ com_crashlytics_export_strings.xml
crashlytics.properties
crashlytics-build.properties
### Emacs ###
# -*- mode: gitignore; -*-
*~
\#*\#
/.emacs.desktop
/.emacs.desktop.lock
*.elc
auto-save-list
tramp
.\#*
# Org-mode
.org-id-locations
*_archive
# flymake-mode
*_flymake.*
# eshell files
/eshell/history
/eshell/lastdir
# elpa packages
/elpa/
# reftex files
*.rel
# AUCTeX auto folder
/auto/
# cask packages
.cask/
# Project Specific
ignore_http/*
config.json
Makefile
+16 -4
View File
@@ -1,10 +1,22 @@
default: install-debug
default: debug
install-debug:
debug:
go install -ldflags "-w" github.com/dimiro1/ipe
run-debug: install-debug
${GOPATH}/bin/ipe --config ${GOPATH}/src/github.com/dimiro1/ipe/config-example.json
run-debug: debug
${GOPATH}/bin/ipe --config ${GOPATH}/src/github.com/dimiro1/ipe/config.json -logtostderr=true -v=2
test:
go test github.com/dimiro1/ipe
macos:
GOOS=darwin GOARHC=x64 go install github.com/dimiro1/ipe
linux:
GOOS=linux GOARHC=x64 go install github.com/dimiro1/ipe
windows:
GOOS=windows GOARHC=x64 go install github.com/dimiro1/ipe
raspberry:
GOOS=linux GOARCH=arm go install github.com/dimiro1/ipe
README.md
+20
View File
@@ -0,0 +1,20 @@
This software is written in Go - the WYSIWYG lang
Why I wrote this software?
--------------------------
1. I wanted to learn Go;
2. I needed a non trivial application;
3. I use Pusher in some projects;
4. I really like Pusher;
Where this name come from?
--------------------------
Here in Brazil we have this beatiful tree called Ipê, it comes in differente colors: yellow, pink, white, purple.
What is implemented?
--------------------
1. Only websockets transport is implemented;
2. Only protocol version 7;
TODO.org
+36 -15
View File
@@ -1,17 +1,38 @@
Tarefas IPE
IPÊ
---
** DONE Autenticação API Rest
** DONE Autenticação Websockets
** TODO Console de log
** TODO Ping e Pong
** TODO Escrever testes automatizados
** TODO Refatorar partes do código, remover repetições
** TODO SSL
** TODO Otimizações
** TODO Segurança, tempo de expiração, etc
** TODO Coletor de conexões abandonadas. Tempo de vida e go routine checking from time to time.
* TODO [7/14]
- [X] Autenticação API Rest
- [X] Autenticação Websockets
- [ ] Console de log
- [X] Ping e Pong
- [ ] Escrever testes automatizados
- [ ] SSL
- [ ] Expvar - Canais, inscritos
- [ ] Otimizações [0/3]
- [ ] Refatorar partes do código, remover repetições
- [ ] Alterar tipos de dados de slices para mapas em alguns locais.
- [ ] Remover Canais vazios com uma go routine - Uma especie de coletor de lixo
- [ ] Segurança, tempo de expiração, etc
- [X] Dados extra na conexão do usuário. Ver Websockets onOpen
- [X] Webhooks [5/5]
- [X] Member added
- [X] Member removed
- [X] Channel Occupied
- [X] Channel vacated
- [X] Clients Events
- [ ] Events Presence channels [0/3]
- [ ] pusher_internal:subscription_succeeded para canais de presença
- [ ] pusher_internal:member_added
- [ ] pusher_internal:member_removed
- [X] Remover inscrições quando o web socket for fechado
- [X] Alterar os dados extras da inscrição são relacionados ao canal e não diretamente a inscrição
Objetivos
** TODO Implementação Funcional.
** TODO Testes automatizados após o projeto pronto.
* Objetivos [4/7]
- [ ] Implementação Funcional.
- [ ] WebHooks
- [ ] Presence channels
- [X] Private Channels
- [X] Public Channels
- [X] Easy Instalation
- [X] Easy configuration
app.go
+88 -51
View File
@@ -6,105 +6,142 @@ package main
import (
"errors"
"strings"
"sync"
log "github.com/golang/glog"
)
// Applications
// An App
type App struct {
sync.Mutex
Name string
AppID string
Key string
Secret string
OnlySSL bool
ApplicationDisabled bool
UserEvents bool
WebHooks bool
URLWebHook string
PublicChannels []*Channel `json:"-"`
PresenceChannels []*Channel `json:"-"`
PrivateChannels []*Channel `json:"-"`
Connections []*Connection `json:"-"`
Subscribers map[string]*Subscriber `json:"-"`
}
// Returns a list of all channels in this app
func (a *App) AllChannels() []*Channel {
var channels []*Channel
for _, c := range a.PrivateChannels {
channels = append(channels, c)
}
for _, c := range a.PublicChannels {
channels = append(channels, c)
}
for _, c := range a.PresenceChannels {
channels = append(channels, c)
}
channels = append(channels, a.PrivateChannels...)
channels = append(channels, a.PresenceChannels...)
channels = append(channels, a.PublicChannels...)
return channels
}
// Create a new Connection
func (a *App) AddConnection(c *Connection) {
a.Connections = append(a.Connections, c)
}
func (a *App) Disconnect(socketID string) {
log.Infof("Disconnecting socket %+v", socketID)
func (a *App) FindConnection(socketID string) (*Connection, error) {
for _, c := range a.Connections {
if c.SocketID == socketID {
return c, nil
s, err := a.FindSubscriber(socketID)
if err != nil {
log.Infof("Socket not found, %+v", err)
return
}
// Unsubscribe from channels
for _, c := range a.AllChannels() {
if c.IsSubscribed(s) {
c.Unsubscribe(a, s)
}
}
return nil, errors.New("Connection not found")
// Remove from app
a.Lock()
_, exists := a.Subscribers[s.SocketID]
if !exists {
return
}
delete(a.Subscribers, s.SocketID)
a.Unlock()
}
// Create a new Subscriber
func (a *App) Connect(s *Subscriber) {
log.Infof("Adding a new Subscriber %s to app %s", s.SocketID, a.Name)
a.Lock()
a.Subscribers[s.SocketID] = s
a.Unlock()
}
// Find a Subscriber on this app
func (a *App) FindSubscriber(socketID string) (*Subscriber, error) {
s, exists := a.Subscribers[socketID]
if exists {
return s, nil
}
return nil, errors.New("Subscriber not found")
}
// Add a new Channel to this APP
func (a *App) AddChannel(c *Channel) {
if c.isPresence() {
log.Infof("Adding a new channel %s to app %s", c.ChannelID, a.Name)
a.Lock()
if c.IsPresence() {
a.PresenceChannels = append(a.PresenceChannels, c)
} else if c.isPrivate() {
} else if c.IsPrivate() {
a.PrivateChannels = append(a.PrivateChannels, c)
} else {
a.PublicChannels = append(a.PublicChannels, c)
}
a.Unlock()
}
func (a *App) FindOrCreateChannelByChannelID(n, data string) *Channel {
channel, err := a.FindChannelByChannelID(n)
// Returns a Channel from this app
// If not found then the channel is created and added to this app
func (a *App) FindOrCreateChannelByChannelID(n string) *Channel {
c, err := a.FindChannelByChannelID(n)
if err != nil {
channel = NewChannel(n, data)
a.AddChannel(channel)
c = NewChannel(n)
a.AddChannel(c)
}
return channel
return c
}
// Find the channel by channel ID
func (a *App) FindChannelByChannelID(n string) (*Channel, error) {
isPrivate := strings.HasPrefix(n, "private-")
isPresence := strings.HasPrefix(n, "presence-")
// Get the channel
if isPresence {
for _, channel := range a.PresenceChannels {
if channel.ChannelID == n {
return channel, nil
}
}
} else if isPrivate {
for _, channel := range a.PrivateChannels {
if channel.ChannelID == n {
return channel, nil
}
}
} else {
for _, channel := range a.PublicChannels {
if channel.ChannelID == n {
return channel, nil
}
for _, c := range a.AllChannels() {
if c.ChannelID == n {
return c, nil
}
}
return nil, errors.New("Channel does not exists")
}
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) Subscribe(c *Channel, s *Subscriber, data string) {
c.Subscribe(a, s, data)
}
app_test.go
+15 -15
View File
@@ -9,7 +9,7 @@ import (
)
func newApp() App {
return App{Name: "Test", AppID: "123", Key: "123", Secret: "123", OnlySSL: false, ApplicationDisabled: false}
return App{Name: "Test", AppID: "123", Key: "123", Secret: "123", OnlySSL: false, ApplicationDisabled: false, UserEvents: true}
}
func Test_add_channels(t *testing.T) {
@@ -65,39 +65,39 @@ func Test_AllChannels(t *testing.T) {
}
}
func Test_New_Connection(t *testing.T) {
func Test_New_Subscriber(t *testing.T) {
app := newApp()
if len(app.Connections) != 0 {
t.Error("Length of connections before test must be 0")
if len(app.Subscribers) != 0 {
t.Error("Length of subscribers before test must be 0")
}
conn := NewConnection("1", "", nil)
app.AddConnection(conn)
conn := NewSubscriber("1", "", nil)
app.AddSubscriber(conn)
if len(app.Connections) != 1 {
t.Error("Length os connections after test must be 1")
if len(app.Subscribers) != 1 {
t.Error("Length os subscribers after test must be 1")
}
}
func Test_find_connection(t *testing.T) {
func Test_find_subscriber(t *testing.T) {
app := newApp()
conn := NewConnection("1", "", nil)
app.AddConnection(conn)
conn := NewSubscriber("1", "", nil)
app.AddSubscriber(conn)
conn, err := app.FindConnection("1")
conn, err := app.FindSubscriber("1")
if err != nil {
t.Error(err)
}
if conn.SocketID != "1" {
t.Error("Wrong connection.")
t.Error("Wrong subscriber.")
}
// Find a wrong connection
// Find a wrong subscriber
conn, err = app.FindConnection("DoesNotExists")
conn, err = app.FindSubscriber("DoesNotExists")
if err == nil {
t.Error("Opps, Must be nil")
auth.go
+7 -21
View File
@@ -5,30 +5,16 @@
package main
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"github.com/gorilla/mux"
"net/http"
"net/url"
"sort"
"strings"
"github.com/gorilla/mux"
"github.com/dimiro1/ipe/utils"
)
// Calculate the Mac
func HashMAC(message, key []byte) string {
mac := hmac.New(sha256.New, key)
mac.Write(message)
expectedMAC := mac.Sum(nil)
return hex.EncodeToString(expectedMAC)
}
// Verify the message
func checkMAC(message, messageMAC, key []byte) bool {
return string(messageMAC) == HashMAC(message, key)
}
// Prepare Querystring
func prepareQueryString(params url.Values) string {
var keys []string
@@ -63,7 +49,7 @@ func RestAuthenticationHandler(h http.Handler) http.Handler {
vars := mux.Vars(r)
appID := vars["app_id"]
currentApp, err := Conf.GetAppByAppID(appID)
app, err := Conf.GetAppByAppID(appID)
if err != nil {
http.Error(w, "Not authorized", http.StatusUnauthorized)
@@ -72,14 +58,14 @@ func RestAuthenticationHandler(h http.Handler) http.Handler {
params := r.URL.Query()
authSignature := params.Get("auth_signature")
signature := params.Get("auth_signature")
params.Del("auth_signature")
queryString := prepareQueryString(params)
toSign := strings.ToUpper(r.Method) + "\n" + r.URL.Path + "\n" + queryString
if checkMAC([]byte(toSign), []byte(authSignature), []byte(currentApp.Secret)) {
if utils.HashMAC([]byte(toSign), []byte(app.Secret)) == signature {
h.ServeHTTP(w, r)
} else {
http.Error(w, "Not authorized", http.StatusUnauthorized)
config-example.json
+12 -10
View File
@@ -1,23 +1,25 @@
{
"Host": ":8080",
"SessionSecret": "372843uhudsbdfy4ure8dbyrty73uhf%7327#vbˆB%bB66BˆVFTV#12g2",
"SessionName": "ipe",
"Apps": [
{
"ApplicationDisabled": false,
"OnlySSL": false,
"Secret": "7ad3773142a6692b25b8",
"Key": "278d425bdf160c739803",
"Name": "Teste",
"AppID": "1"
"Secret": "7ad3753142a6693b25b9",
"Key": "278d525bdf162c739803",
"Name": "App 1",
"AppID": "321",
"UserEvents": true,
"WebHooks": true
},
{
"ApplicationDisabled": false,
"OnlySSL": false,
"Secret": "2189389a989a9ab8921f",
"Key": "6a266506823423cf2a1f",
"Name": "Apresentacao",
"AppID": "2"
"Secret": "d6824d2fa32888931504",
"Key": "c8b30f611ffb13202976",
"Name": "App 2",
"AppID": "123",
"UserEvents": true,
"WebHooks": false
}
]
}
config.go
+19 -11
View File
@@ -4,31 +4,39 @@
package main
import (
"errors"
)
import "errors"
// The config file
type ConfigFile struct {
Host string
SessionName string
SessionSecret string
Apps []*App
Host string // The host, eg: :8080 will start on 0.0.0.0:8080
Apps []*App
}
func (c ConfigFile) GetAppByAppID(appID string) (*App, error) {
// Error for App not found
var AppNotFoundError = errors.New("App not found")
func (c *ConfigFile) Initialize() {
for _, app := range c.Apps {
app.Subscribers = make(map[string]*Subscriber)
}
}
// Returns an App with by appID
func (c *ConfigFile) GetAppByAppID(appID string) (*App, error) {
for _, app := range c.Apps {
if app.AppID == appID {
return app, nil
}
}
return &App{}, errors.New("App not found")
return &App{}, AppNotFoundError
}
func (c ConfigFile) GetAppByKey(key string) (*App, error) {
// Returns an App with by key
func (c *ConfigFile) GetAppByKey(key string) (*App, error) {
for _, app := range c.Apps {
if app.Key == key {
return app, nil
}
}
return &App{}, errors.New("App not found")
return &App{}, AppNotFoundError
}
conn.go
+155 -64
View File
@@ -5,134 +5,225 @@
package main
import (
"encoding/json"
"errors"
"github.com/gorilla/websocket"
"log"
"strings"
"sync"
"time"
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 Connection struct {
Id int
type Subscriber struct {
Id string
SocketID string
Data string // Extra data attached to this subscriber
Socket *websocket.Conn
Messages chan []byte
}
// 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 {
ChannelID string
Data string
Connections []*Connection
Messages chan []byte
sync.Mutex
CreatedAt time.Time
ChannelID string
Subscriptions []*Subscription
}
// Return true if the channel has at least one subscriber
func (c Channel) isOccupied() bool {
return c.totalConnections() > 0
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 presence
func (c Channel) isPresence() bool {
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 {
func (c *Channel) IsPrivate() bool {
return strings.HasPrefix(c.ChannelID, "private-")
}
// Get the total of subscribers
func (c Channel) totalConnections() int {
return len(c.Connections)
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.totalConnections()
func (c *Channel) TotalUsers() int {
return c.TotalSubscriptions()
}
// Add a new subscriber to the channel
func (c *Channel) Subscribe(conn *Connection) {
c.Connections = append(c.Connections, conn)
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 = append(c.Subscriptions, 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 {
for _, subs := range c.Subscriptions {
if subs.Subscriber == s {
return true
}
}
return false
}
// Remove the subscriber from the channel
func (c *Channel) Unsubscribe(conn *Connection) error {
index := -1
// 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)
for i, c := range c.Connections {
if c == conn {
c.Lock()
defer c.Unlock()
index := -1
for i, subs := range c.Subscriptions {
if subs.Subscriber == s {
index = i
break
}
}
if index == -1 {
return errors.New("Connections not found")
return errors.New("Subscription not found")
}
c.Subscriptions = append(c.Subscriptions[:index], c.Subscriptions[index+1:]...)
if c.IsPresence() {
// Publish pusher_internal:member_removed
// Webhook
a.TriggerMemberRemovedHook(c, s)
}
c.Connections = append(c.Connections[:index], c.Connections[index+1:]...)
// Remove Channel if necessary
// Close sockets and Channels
// WebHook
if c.TotalSubscriptions() == 0 {
a.TriggerChannelVacatedHook(c)
}
return nil
}
// Create a new Channel
func NewChannel(channelID, data string) *Channel {
c := &Channel{ChannelID: channelID, Data: data, Messages: make(chan []byte)}
c.Listen()
func NewChannel(channelID string) *Channel {
log.Infof("Creating a new channel: %s", channelID)
return c
return &Channel{ChannelID: channelID, CreatedAt: time.Now()}
}
var mu = &sync.Mutex{}
var currentID = 0
// This function generate a sequencial ID
func newID() string {
mutex.Lock()
defer mutex.Unlock()
// Generate a New ID
func newID() int {
mu.Lock()
defer mu.Unlock()
currentID += 1
return currentID
return string(currentID)
}
// Create a new Connection
func NewConnection(socketID, data string, socket *websocket.Conn) *Connection {
// Create a new Subscriber
func NewSubscriber(socketID string, s *websocket.Conn) *Subscriber {
id := newID()
connection := &Connection{Id: id, SocketID: socketID, Data: data, Socket: socket, Messages: make(chan []byte)}
connection.Listen()
log.Infof("Creating a new Subscriber %+v with id %d", socketID, id)
return connection
return &Subscriber{Id: id, SocketID: socketID, Socket: s}
}
func (c *Channel) Listen() {
go func() {
select {
case message := <-c.Messages:
// err := c.Socket.WriteMessage(websocket.TextMessage, message)
// log.Println(err)
log.Println(message)
// 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)
}
}
}
}()
return nil
}
// Listen Messages
func (c *Connection) Listen() {
go func() {
select {
case message := <-c.Messages:
// err := c.Socket.WriteMessage(websocket.TextMessage, message)
// log.Println(err)
log.Println(message)
}
}()
// 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
}
conn_test.go
+17
View File
@@ -0,0 +1,17 @@
// 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 (
"testing"
)
func Test_New_ID(t *testing.T) {
id := newID()
if newID() != id+1 {
t.Error("Every call to newID must increment the id by one")
}
}
constants.go
+4 -3
View File
@@ -15,7 +15,7 @@ const (
APPLICATION_DOES_NOT_EXISTS = 4001
APPLICATION_DISABLED = 4003
APPLICATION_IS_OVER_CONNECTION_QUOTA = 4004 // Not Implemented
PATH_NOT_FOUND = 4005
PATH_NOT_FOUND = 4005 // Not Implemented
INVALID_VERSION_STRING_FORMAT = 4006
UNSUPPORTED_PROTOCOL_VERSION = 4007
NO_PROTOCOL_VERSION_SUPPLIED = 4008
@@ -36,8 +36,9 @@ const (
// Any other type of error
CLIENT_REJECTED_DUE_TO_RATE_LIMIT = 4301 // Not Implemented
// Pusher send null, This app send -1 on the error
GENERIC_ERROR = -1
// Pusher send null, This app use this error code to send the null value
// see ErrorEvent
GENERIC_ERROR = 0
)
// Only this version is supported
events.go
+125 -26
View File
@@ -4,6 +4,8 @@
package main
import "encoding/json"
// {
// "event": "pusher:subscribe",
// "data": {
@@ -15,7 +17,7 @@ package main
type SubscribeEventData struct {
Channel string `json:"channel"`
Auth string `json:"auth,omitempty"`
ChannelData string `json:"channelData,omitempty"`
ChannelData string `json:"channel_data,omitempty"`
}
type SubscribeEvent struct {
@@ -57,11 +59,54 @@ func NewUnsubscribeEvent(channel string) UnsubscribeEvent {
type SubscriptionSucceededEvent struct {
Event string `json:"event"`
Channel string `json:"channel"`
Data string `json:"data"`
}
// Create a new subscription succeed event for the specified channel
func NewSubscriptionSucceededEvent(channel string) SubscriptionSucceededEvent {
return SubscriptionSucceededEvent{Event: "pusher_internal:subscription_succeeded", Channel: channel}
func NewSubscriptionSucceededEvent(channel, data string) SubscriptionSucceededEvent {
return SubscriptionSucceededEvent{Event: "pusher_internal:subscription_succeeded", Channel: channel, Data: data}
}
// Data Subscription Succeed
// "{
// \"presence\": {
// \"ids\": [\"11814b369700141b222a3f3791cec2d9\",\"71dd6a29da2a4833336d2a964becf820\"],
// \"hash\": {
// \"11814b369700141b222a3f3791cec2d9\": {
// \"name\":\"Phil Leggetter\",
// \"twitter\": \"@leggetter\"
// },
// \"71dd6a29da2a4833336d2a964becf820\": {
// \"name\":\"Max Williams\",
// \"twitter\": \"@maxthelion\"
// }
// },
// \"count\": 2
// }
// }"
type SubscriptionSucceeedEventPresenceData struct {
Ids []string `json:"ids"`
Hash map[string]string `json:"hash"`
count int `json:"count"`
}
func NewSubscriptionSucceedEventPresenceData(c *Channel) SubscriptionSucceeedEventPresenceData {
event := SubscriptionSucceeedEventPresenceData{}
var ids []string
hash := make(map[string]string, c.TotalSubscriptions())
for _, s := range c.Subscriptions {
ids = append(ids, s.Subscriber.SocketID)
hash[s.Subscriber.SocketID] = s.Data
}
event.Ids = ids
event.Hash = hash
event.count = c.TotalSubscriptions()
return event
}
// {
@@ -99,18 +144,36 @@ func NewPingEvent() PingEvent {
// "code": 4000
// }
// }
type ErrorEventData struct {
Message string `json:"message"`
Code int `json:"code"`
}
type ErrorEvent struct {
Event string `json:"event"`
Data ErrorEventData `json:"data"`
Event string `json:"event"`
Data interface{} `json:"data"`
}
// Create a new error event
// Pusher protocol is very strange in some parts
// It send null in some errors.
// So I created this GENERIC_ERROR thing, just to verify if the json must have null on the error code
func NewErrorEvent(code int, message string) ErrorEvent {
data := ErrorEventData{Message: message, Code: code}
var data interface{}
if code == GENERIC_ERROR {
data = struct {
Code *int `json:"code"`
Message string `json:"message"`
}{
nil,
message,
}
} else {
data = struct {
Code int `json:"code"`
Message string `json:"message"`
}{
code,
message,
}
}
return ErrorEvent{Event: "pusher:error", Data: data}
}
@@ -127,35 +190,71 @@ type ConnectionEstablishedEventData struct {
}
type ConnectionEstablishedEvent struct {
Event string `json:"event"`
Data ConnectionEstablishedEventData `json:"data"`
Event string `json:"event"`
Data string `json:"data"`
}
// Create a new connection established event using the specified socketId
func NewConnectionEstablishedEvent(socketId string) ConnectionEstablishedEvent {
data := ConnectionEstablishedEventData{SocketId: socketId, ActivityTimeout: 120}
return ConnectionEstablishedEvent{Event: "pusher:connection_established", Data: data}
b, err := json.Marshal(data)
if err != nil {
panic("events: Could not Marshal json ConnectionEstablishedEvent")
}
return ConnectionEstablishedEvent{Event: "pusher:connection_established", Data: string(b)}
}
// {
// "event": "pusher_internal:member_added",
// "channel": "presence-example-channel",
// "data": String
// }
type MemberAddedEvent struct {
Event string `json:"event"`
Channel string `json:"channel"`
Data string `json:"data"`
}
func NewMemberAddedEvent(channel, data string) MemberAddedEvent {
return MemberAddedEvent{Event: "pusher_internal:member_added", Channel: channel, Data: data}
}
// {
// "event": "pusher_internal:member_removed",
// "channel": "presence-example-channel",
// "data": String
// }
type MemberRemovedEvent struct {
Event string `json:"event"`
Channel string `json:"channel"`
Data string `json:"data"`
}
func NewMemberRemovedEvent(channel, data string) MemberRemovedEvent {
return MemberRemovedEvent{Event: "pusher_internal:member_removed", Channel: channel, Data: data}
}
// {
// "event": "client-?",
// "channel": "The channel",
// "data": {
// "message": "A Message"
// }
// "data": {}
// }
type ClientEventData struct {
Message string `json:"data"`
}
type ClientEvent struct {
type RawEvent struct {
Event string `json:"event"`
Channel string `json:"channel"`
Data ClientEventData `json:"data"`
Data json.RawMessage `json:"data"`
}
// Create a new custom client event
func NewClientEvent(name, channel, message string) ClientEvent {
data := ClientEventData{Message: message}
return ClientEvent{Event: "pusher:client-" + name, Channel: channel, Data: data}
type ResponseEvent struct {
Event string `json:"event"`
Channel string `json:"channel"`
Data interface{} `json:"data"`
}
// The response event that is broadcasted to the client sockets
func NewResponseEvent(name, channel string, data interface{}) ResponseEvent {
return ResponseEvent{Event: name, Channel: channel, Data: data}
}
extra_handlers.go
+2 -7
View File
@@ -6,10 +6,9 @@ package main
import (
"fmt"
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
"net/http"
"os"
"github.com/gorilla/mux"
)
// Check if the application is disabled
@@ -33,7 +32,3 @@ func RestCheckAppDisabledHandler(h http.Handler) http.Handler {
h.ServeHTTP(w, r)
})
}
func LogHandler(h http.Handler) http.Handler {
return handlers.CombinedLoggingHandler(os.Stdout, h)
}
main.go
+14 -3
View File
@@ -7,14 +7,18 @@ package main
import (
"encoding/json"
"flag"
"fmt"
"io/ioutil"
"log"
"net/http"
log "github.com/golang/glog"
)
var Conf ConfigFile
func main() {
printBanner()
var filename = flag.String("config", "config.json", "Config file location")
flag.Parse()
@@ -29,11 +33,18 @@ func main() {
log.Fatal(err)
}
Conf.Initialize()
router := NewRouter()
log.Printf("Starting Ipê using config file: '%s'", *filename)
log.Infof("Starting Ipê using config file: '%s'", *filename)
if err := http.ListenAndServe(Conf.Host, router); err != nil {
log.Fatalln(err)
log.Fatal(err)
}
}
func printBanner() {
fmt.Println("\033[32mWelcome to Ipê - Yet another Pusher server clone\033[0m")
fmt.Println("\033[33mBy: Claudemiro Alves Feitosa Neto <dimiro1@gmail.com>\033[0m")
}
rest.go
+49 -25
View File
@@ -7,10 +7,11 @@ package main
import (
"encoding/json"
"fmt"
"github.com/gorilla/mux"
"log"
"net/http"
"strings"
log "github.com/golang/glog"
"github.com/gorilla/mux"
)
// An event consists of a name and data (typically JSON) which may be sent to all subscribers to a particular channel or channels.
@@ -29,15 +30,24 @@ import (
//
// POST /apps/{app_id}/events
func PostEvents(w http.ResponseWriter, r *http.Request) {
var input struct {
Name string `json:"name"`
Data string `json:"data"`
Channels []string `json:"channels,omitempty"`
Channel string `json:"channel,omitempty"`
SocketID string `json:"socket_id,omitempty"`
vars := mux.Vars(r)
appID := vars["app_id"]
app, err := Conf.GetAppByAppID(appID)
if err != nil {
http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
}
err := json.NewDecoder(r.Body).Decode(&input)
var input struct {
Name string `json:"name"`
Data json.RawMessage `json:"data"`
Channels []string `json:"channels,omitempty"`
Channel string `json:"channel,omitempty"`
SocketID string `json:"socket_id,omitempty"`
}
err = json.NewDecoder(r.Body).Decode(&input)
if err != nil {
http.Error(w, "Bad request", http.StatusBadRequest)
@@ -50,8 +60,24 @@ func PostEvents(w http.ResponseWriter, r *http.Request) {
return
}
// Trigger events
log.Info(input.Channels)
if len(input.Channel) > 0 && len(input.Channels) == 0 {
input.Channels = append(input.Channels, input.Channel)
}
for _, c := range input.Channels {
channel, err := app.FindChannelByChannelID(c)
// Channel exists?
if err != nil {
http.Error(w, fmt.Sprintf("Could not find a channel with id %s", c), http.StatusBadRequest)
return
}
app.Publish(channel, RawEvent{Event: input.Name, Channel: c, Data: input.Data}, input.SocketID)
}
w.Header().Set("Content-Type", "application/json;charset=UTF-8")
w.WriteHeader(http.StatusOK)
w.Write([]byte("{}"))
}
@@ -115,7 +141,7 @@ func GetChannels(w http.ResponseWriter, r *http.Request) {
channels[c.ChannelID] = struct {
UserCount int `json:"user_count"`
}{
c.totalUsers(),
c.TotalUsers(),
}
} else {
channels[c.ChannelID] = struct{}{}
@@ -138,8 +164,8 @@ 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 {
log.Error(err)
http.Error(w, "Internal server error", http.StatusInternalServerError)
log.Println(err)
}
}
@@ -190,8 +216,6 @@ func GetChannel(w http.ResponseWriter, r *http.Request) {
}
}
// Check the kind of channel
channel, err := app.FindChannelByChannelID(channelName)
// Channel exists?
@@ -202,7 +226,7 @@ func GetChannel(w http.ResponseWriter, r *http.Request) {
// If an attribute such as user_count is requested, and the request is not limited
// to presence channels, the API will return an error (400 code)
if requestedUserCount && !channel.isPresence() {
if requestedUserCount && !channel.IsPresence() {
http.Error(w, "Attribute user_count is restricted to presence channels", http.StatusBadRequest)
return
}
@@ -212,25 +236,25 @@ func GetChannel(w http.ResponseWriter, r *http.Request) {
Occupied bool `json:"occupied"`
UserCount int `json:"user_count,omitempty"`
SubscriptionCount int `json:"subscription_count,omitempty"`
}{Occupied: channel.isOccupied()}
}{Occupied: channel.IsOccupied()}
switch {
case requestedSubscriptionCount && requestedUserCount:
dtoChannel.UserCount = channel.totalUsers()
dtoChannel.SubscriptionCount = channel.totalConnections()
dtoChannel.UserCount = channel.TotalUsers()
dtoChannel.SubscriptionCount = channel.TotalSubscriptions()
case requestedUserCount:
dtoChannel.UserCount = channel.totalUsers()
dtoChannel.UserCount = channel.TotalUsers()
case requestedSubscriptionCount:
dtoChannel.SubscriptionCount = channel.totalConnections()
dtoChannel.SubscriptionCount = channel.TotalSubscriptions()
}
w.Header().Set("Content-Type", "application/json;charset=UTF-8")
if err := json.NewEncoder(w).Encode(dtoChannel); err != nil {
log.Error(err)
http.Error(w, "Internal server error", http.StatusInternalServerError)
log.Println(err)
}
}
@@ -277,10 +301,10 @@ func GetChannelUsers(w http.ResponseWriter, r *http.Request) {
var users []interface{}
for _, s := range channel.Connections {
for _, s := range channel.Subscriptions {
users = append(users, struct {
Id int `json:"id"`
}{s.Id})
Id string `json:"id"`
}{s.Subscriber.Id})
}
result["users"] = users
@@ -289,6 +313,6 @@ func GetChannelUsers(w http.ResponseWriter, r *http.Request) {
if err := json.NewEncoder(w).Encode(result); err != nil {
http.Error(w, "Internal server error", http.StatusInternalServerError)
log.Println(err)
log.Error(err)
}
}
router.go
+7 -2
View File
@@ -5,10 +5,15 @@
package main
import (
"github.com/gorilla/mux"
"net/http"
"os"
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
)
// NewRouter is a function that returns a new configured Router
// It add the necessary middlewares
func NewRouter() *mux.Router {
router := mux.NewRouter().StrictSlash(true)
@@ -22,7 +27,7 @@ func NewRouter() *mux.Router {
handler = RestCheckAppDisabledHandler(handler)
}
handler = LogHandler(handler)
handler = handlers.CombinedLoggingHandler(os.Stdout, handler)
router.Methods(route.Method).Path(route.Pattern).Name(route.Name).Handler(handler)
}
routes.go
+1
View File
@@ -8,6 +8,7 @@ import (
"net/http"
)
// A route
type Route struct {
Name string
Method string
utils/utils.go
+20
View File
@@ -0,0 +1,20 @@
// 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 utils
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
)
// HashMAC Calculates the MAC signing with the given key and returns the hexadecimal encoded Result
func HashMAC(message, key []byte) string {
mac := hmac.New(sha256.New, key)
mac.Write(message)
expected := mac.Sum(nil)
return hex.EncodeToString(expected)
}
webhooks.go
+160
View File
@@ -0,0 +1,160 @@
// 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 (
"bytes"
"encoding/json"
"net/http"
"time"
"github.com/dimiro1/ipe/utils"
log "github.com/golang/glog"
)
// A WebHook is sent as a HTTP POST request to the url which you specify.
// The POST request payload (body) contains a JSON document, and follows the following format:
// {
// "time_ms": 1327078148132
// "events": [
// { "name": "event_name", "some": "data" }
// ]
// }
//
// Security
// Encryption
//
// You may use a HTTP or a HTTPS url for WebHooks. In most cases HTTP is sufficient, but HTTPS can be useful if your data is sensitive or if you wish to protect against replay attacks for example.
// Authentication
//
// Since anyone could in principle send WebHooks to your application, it’s important to verify that these WebHooks originated from Pusher. Valid WebHooks will therefore contain these headers which contain a HMAC signature of the WebHook payload (body):
//
// X-Pusher-Key: A Pusher app may have multiple tokens. The oldest active token will be used, identified by this key.
// X-Pusher-Signature: A HMAC SHA256 hex digest formed by signing the POST payload (body) with the token’s secret.
type WebHook struct {
TimeMs int64 `json:"time_ms"`
Events []HookEvent `json:"events"`
}
type HookEvent struct {
Name string `json:"name"`
Channel string `json:"channel"`
Event string `json:"event,omitempty"`
Data string `json:"data,omitempty"`
SocketID string `json:"socket_id,omitempty"`
UserId string `json:"user_id,omitempty"`
}
func NewChannelOcuppiedHook(channel *Channel) HookEvent {
return HookEvent{Name: "channel_occupied", Channel: channel.ChannelID}
}
func NewChannelVacatedHook(channel *Channel) HookEvent {
return HookEvent{Name: "channel_vacated", Channel: channel.ChannelID}
}
func NewMemberAddedHook(channel *Channel, s *Subscriber) HookEvent {
return HookEvent{Name: "member_added", Channel: channel.ChannelID, UserId: s.Id}
}
func NewMemberRemovedHook(channel *Channel, s *Subscriber) HookEvent {
return HookEvent{Name: "member_removed", Channel: channel.ChannelID, UserId: s.Id}
}
func NewClientHook(channel *Channel, s *Subscription, event string) HookEvent {
return HookEvent{Name: "client_event", Channel: channel.ChannelID, Event: event, Data: s.Data, SocketID: s.Subscriber.SocketID, UserId: s.Subscriber.Id}
}
// channel_occupied
// { "name": "channel_occupied", "channel": "test_channel" }
func (a *App) TriggerChannelOccupiedHook(c *Channel) {
event := NewChannelOcuppiedHook(c)
triggerHook(event.Name, a, c, event)
}
// channel_vacated
// { "name": "channel_vacated", "channel": "test_channel" }
func (a *App) TriggerChannelVacatedHook(c *Channel) {
event := NewChannelVacatedHook(c)
triggerHook(event.Name, a, c, event)
}
// {
// "name": "client_event",
// "channel": "name of the channel the event was published on",
// "event": "name of the event",
// "data": "data associated with the event",
// "socket_id": "socket_id of the sending socket",
// "user_id": "user_id associated with the sending socket" # Only for presence channels
// }
func (a *App) TriggerClientEventHook(c *Channel, s *Subscription, client_event string) {
event := NewClientHook(c, s, client_event)
triggerHook(event.Name, a, c, event)
}
// {
// "name": "member_added",
// "channel": "presence-your_channel_name",
// "user_id": "a_user_id"
// }
func (a *App) TriggerMemberAddedHook(c *Channel, s *Subscriber) {
event := NewMemberAddedHook(c, s)
triggerHook(event.Name, a, c, event)
}
// {
// "name": "member_removed",
// "channel": "presence-your_channel_name",
// "user_id": "a_user_id"
// }
func (a *App) TriggerMemberRemovedHook(c *Channel, s *Subscriber) {
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)
return
}
go func() {
log.Infof("Triggering %s event", name)
hook := WebHook{TimeMs: time.Now().Unix()}
hook.Events = append(hook.Events, event)
var js []byte
var err error
js, err = json.Marshal(hook)
if err != nil {
log.Errorf("Error decoding json: %+v", err)
return
}
var req *http.Request
req, err = http.NewRequest("POST", app.URLWebHook, bytes.NewReader(js))
if err != nil {
log.Errorf("Error creating request: %+v", err)
return
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Pusher-Key", app.Key)
req.Header.Set("X-Pusher-Signature", utils.HashMAC(js, []byte(app.Secret)))
log.V(1).Infof("%+v", req.Header)
log.V(1).Infof("%+v", string(js))
if _, err := http.DefaultClient.Do(req); err != nil {
log.Errorf("Error posting %s event: %+v", name, err)
}
}()
}
websockets.go
+118 -65
View File
@@ -5,15 +5,20 @@
package main
import (
"crypto/rand"
"encoding/base32"
"encoding/json"
"fmt"
"github.com/gorilla/mux"
"github.com/gorilla/sessions"
"github.com/gorilla/websocket"
"log"
"io"
"net/http"
"strconv"
"strings"
log "github.com/golang/glog"
"github.com/gorilla/mux"
"github.com/gorilla/websocket"
"github.com/dimiro1/ipe/utils"
)
var upgrader = websocket.Upgrader{
@@ -22,8 +27,8 @@ var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}
// Handle open connection.
func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, session *sessions.Session, app *App) WebsocketError {
// Handle open Subscriber.
func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessionID string, app *App) WebsocketError {
params := r.URL.Query()
p := params.Get("protocol")
@@ -46,20 +51,28 @@ func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessio
}
}
// Create the new connection
connection := NewConnection(session.ID, "", conn)
app.AddConnection(connection)
// Create the new Subscriber
subscriber := NewSubscriber(sessionID, conn)
app.Connect(subscriber)
// Everything went fine. Huhu.
if err := conn.WriteJSON(NewConnectionEstablishedEvent(connection.SocketID)); err != nil {
if err := conn.WriteJSON(NewConnectionEstablishedEvent(subscriber.SocketID)); err != nil {
return NewGenericReconnectImmediatelyError()
}
return nil
}
// Handle the close event
func onClose(sessionID string, app *App) {
app.Disconnect(sessionID)
}
// Handle messages
func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, session *sessions.Session, app *App) WebsocketError {
//
// If there is an unrecoverable error then break the loop,
// otherwise just keep going.
func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessionID string, app *App) {
var event struct {
Event string `json:"event"`
}
@@ -68,104 +81,137 @@ func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, ses
_, message, err := conn.ReadMessage()
if err != nil {
return NewGenericReconnectImmediatelyError()
log.Errorf("%+v", err)
switch err {
case io.EOF:
onClose(sessionID, app)
default:
emitWSError(NewGenericReconnectImmediatelyError(), conn)
}
break
}
if err := json.Unmarshal(message, &event); err != nil {
return NewGenericReconnectImmediatelyError()
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
log.Infof("websockets: Handling %s event", event.Event)
switch event.Event {
case "pusher:ping":
if err := conn.WriteJSON(NewPongEvent()); err != nil {
return NewGenericReconnectImmediatelyError()
emitWSError(NewGenericReconnectImmediatelyError(), conn)
}
case "pusher:subscribe":
subscribeEvent := SubscribeEvent{}
if err := json.Unmarshal(message, &subscribeEvent); err != nil {
return NewGenericReconnectImmediatelyError()
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
connection, err := app.FindConnection(session.ID)
subscriber, err := app.FindSubscriber(sessionID)
if err != nil {
return NewGenericReconnectImmediatelyError()
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
channelName := strings.TrimSpace(subscribeEvent.Data.Channel)
// Authentication
if strings.HasPrefix(channelName, "presence-") {
toSign := fmt.Sprintf("%s:%s:%s", connection.SocketID, channelName, subscribeEvent.Data.ChannelData)
isPresence := strings.HasPrefix(channelName, "presence-")
isPrivate := strings.HasPrefix(channelName, "private-")
if subscribeEvent.Data.Auth != HashMAC([]byte(toSign), []byte(app.Secret)) {
return NewGenericError(fmt.Sprintf("Auth value for subscription to %s is invalid", channelName))
if isPresence || isPrivate {
toSign := []string{subscriber.SocketID, channelName}
if isPresence {
toSign = append(toSign, subscribeEvent.Data.ChannelData)
}
} else if strings.HasPrefix(channelName, "private-") {
toSign := fmt.Sprintf("%s:%s", connection.SocketID, channelName)
if subscribeEvent.Data.Auth != HashMAC([]byte(toSign), []byte(app.Secret)) {
return NewGenericError(fmt.Sprintf("Auth value for subscription to %s is invalid", channelName))
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
}
}
channel := app.FindOrCreateChannelByChannelID(channelName, subscribeEvent.Data.ChannelData)
channel.Subscribe(connection)
channel := app.FindOrCreateChannelByChannelID(channelName)
app.Subscribe(channel, subscriber, subscribeEvent.Data.ChannelData)
if err := conn.WriteJSON(NewSubscriptionSucceededEvent(channel.ChannelID)); err != nil {
return NewGenericReconnectImmediatelyError()
if err := conn.WriteJSON(NewSubscriptionSucceededEvent(channel.ChannelID, "{}")); err != nil {
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
case "pusher:unsubscribe":
unsubscribeEvent := UnsubscribeEvent{}
if err := json.Unmarshal(message, &unsubscribeEvent); err != nil {
return NewGenericReconnectImmediatelyError()
emitWSError(NewGenericReconnectImmediatelyError(), conn)
}
connection, err := app.FindConnection(session.ID)
subscriber, err := app.FindSubscriber(sessionID)
if err != nil {
return NewGenericError(fmt.Sprintf("Could not find a connection with the id %s", session.ID))
emitWSError(NewGenericError(fmt.Sprintf("Could not find a subscriber with the id %s", sessionID)), conn)
}
channel, err := app.FindChannelByChannelID(unsubscribeEvent.Data.Channel)
if err != nil {
return NewGenericError(fmt.Sprintf("Could not find a channel with the id %s", unsubscribeEvent.Data.Channel))
emitWSError(NewGenericError(fmt.Sprintf("Could not find a channel with the id %s", unsubscribeEvent.Data.Channel)), conn)
}
if err := channel.Unsubscribe(connection); err != nil {
return NewGenericReconnectImmediatelyError()
if err := app.Unsubscribe(channel, subscriber); err != nil {
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
}
default: // CLient Events ??
// see http://pusher.com/docs/client_api_guide/client_events#trigger-events
if strings.HasPrefix(event.Event, "client-") {
if !app.UserEvents {
emitWSError(NewGenericError("To send client events, you must enable this feature in the Settings."), conn)
}
return nil
// Client Events
}
clientEvent := RawEvent{}
if err := json.Unmarshal(message, &clientEvent); err != nil {
log.Error(err)
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
channel, err := app.FindChannelByChannelID(clientEvent.Channel)
if !channel.IsPresenceOrPrivate() {
emitWSError(NewGenericError("Client event rejected - only supported on private and presence channels"), conn)
break
}
if err != nil {
emitWSError(NewGenericError(fmt.Sprintf("Could not find a channel with the id %s", clientEvent.Channel)), conn)
}
if err := app.Publish(channel, clientEvent, sessionID); err != nil {
log.Error(err)
emitWSError(NewGenericReconnectImmediatelyError(), conn)
break
}
}
} // switch
} // For
}
// Websocket GET /app/{key}
func Websocket(w http.ResponseWriter, r *http.Request) {
conn, err := upgrader.Upgrade(w, r, nil)
defer conn.Close()
if err != nil {
log.Println(err)
emitWSError(NewGenericReconnectImmediatelyError(), conn)
return
}
var store = sessions.NewFilesystemStore("", []byte(Conf.SessionSecret))
session, err := store.Get(r, Conf.SessionName)
if err != nil {
log.Println(err)
emitWSError(NewGenericReconnectImmediatelyError(), conn)
return
}
if err := session.Save(r, w); err != nil {
log.Println(err)
log.Error(err)
emitWSError(NewGenericReconnectImmediatelyError(), conn)
return
}
@@ -176,31 +222,38 @@ func Websocket(w http.ResponseWriter, r *http.Request) {
app, err := Conf.GetAppByKey(appKey)
if err != nil {
log.Println(err)
log.Error(err)
emitWSError(NewApplicationDoesNotExistsError(), conn)
return
}
if err := onOpen(conn, w, r, session, app); err != nil {
sessionID := randomSessionID()
if err := onOpen(conn, w, r, sessionID, app); err != nil {
emitWSError(err, conn)
return
}
if err := onMessage(conn, w, r, session, app); err != nil {
emitWSError(err, conn)
onMessage(conn, w, r, sessionID, app)
}
// Find the connection in app and destroy it
return
// Generate a new random sessionID
func randomSessionID() string {
b := make([]byte, 20)
if _, err := rand.Read(b); err != nil {
panic("websockets: Could not generate a random session ID")
}
return base32.StdEncoding.EncodeToString(b)
}
// Emit an Websocket ErrorEvent
func emitWSError(err WebsocketError, conn *websocket.Conn) {
event := NewErrorEvent(err.GetCode(), err.GetMsg())
if err := conn.WriteJSON(event); err != nil {
log.Println(err)
log.Error(err)
}
conn.Close()
}