From e1b7bd8252503db0fa1513164741645327040b13 Mon Sep 17 00:00:00 2001 From: claudemiro Date: Fri, 2 Jan 2015 18:20:27 -0300 Subject: [PATCH] First import --- .gitignore | 122 ++++++++++++++++++ LICENSE | 21 ++++ Makefile | 10 ++ TODO.org | 17 +++ app.go | 110 +++++++++++++++++ app_test.go | 155 +++++++++++++++++++++++ auth.go | 88 +++++++++++++ config-example.json | 23 ++++ config.go | 34 +++++ conn.go | 138 +++++++++++++++++++++ constants.go | 48 ++++++++ errors.go | 118 ++++++++++++++++++ events.go | 161 ++++++++++++++++++++++++ extra_handlers.go | 39 ++++++ main.go | 39 ++++++ rest.go | 294 ++++++++++++++++++++++++++++++++++++++++++++ rest_test.go_ | 158 ++++++++++++++++++++++++ router.go | 31 +++++ routes.go | 57 +++++++++ websockets.go | 206 +++++++++++++++++++++++++++++++ 20 files changed, 1869 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 Makefile create mode 100644 TODO.org create mode 100644 app.go create mode 100644 app_test.go create mode 100644 auth.go create mode 100644 config-example.json create mode 100644 config.go create mode 100644 conn.go create mode 100644 constants.go create mode 100644 errors.go create mode 100644 events.go create mode 100644 extra_handlers.go create mode 100644 main.go create mode 100644 rest.go create mode 100644 rest_test.go_ create mode 100644 router.go create mode 100644 routes.go create mode 100644 websockets.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..90e1e71 --- /dev/null +++ b/.gitignore @@ -0,0 +1,122 @@ +# Created by https://www.gitignore.io + +### Go ### +# Compiled Object files, Static and Dynamic libs (Shared Objects) +*.o +*.a +*.so + +# Folders +_obj +_test + +# Architecture specific extensions/prefixes +*.[568vq] +[568vq].out + +*.cgo1.go +*.cgo2.c +_cgo_defun.c +_cgo_gotypes.go +_cgo_export.* + +_testmain.go + +*.exe +*.test +*.prof + + +### SublimeText ### +# cache files for sublime text +*.tmlanguage.cache +*.tmPreferences.cache +*.stTheme.cache + +# workspace files are user-specific +*.sublime-workspace + +# project files should be checked into the repository, unless a significant +# proportion of contributors will probably not be using SublimeText +# *.sublime-project + +# sftp configuration file +sftp-config.json + + +### OSX ### +.DS_Store +.AppleDouble +.LSOverride + +# Icon must end with two \r +Icon + + +# Thumbnails +._* + +# Files that might appear on external disk +.Spotlight-V100 +.Trashes + +# Directories potentially created on remote AFP share +.AppleDB +.AppleDesktop +Network Trash Folder +Temporary Items +.apdisk + + +### Intellij ### +# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm + +*.iml + +## Directory-based project format: +.idea/ +# if you remove the above rule, at least ignore the following: + +# User-specific stuff: +# .idea/workspace.xml +# .idea/tasks.xml +# .idea/dictionaries + +# Sensitive or high-churn files: +# .idea/dataSources.ids +# .idea/dataSources.xml +# .idea/sqlDataSources.xml +# .idea/dynamic.xml +# .idea/uiDesigner.xml + +# Gradle: +# .idea/gradle.xml +# .idea/libraries + +# Mongo Explorer plugin: +# .idea/mongoSettings.xml + +## File-based project format: +*.ipr +*.iws + +## Plugin-specific files: + +# IntelliJ +out/ + +# mpeltonen/sbt-idea plugin +.idea_modules/ + +# JIRA plugin +atlassian-ide-plugin.xml + +# Crashlytics plugin (for Android Studio and IntelliJ) +com_crashlytics_export_strings.xml +crashlytics.properties +crashlytics-build.properties + +# Project Specific + +ignore_http/* + diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..f6cdb4f --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +The MIT License (MIT) + +Copyright (c) 2014-2015 Claudemiro Alves Feitosa Neto + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..1b9108a --- /dev/null +++ b/Makefile @@ -0,0 +1,10 @@ +default: install-debug + +install-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 + +test: + go test github.com/dimiro1/ipe diff --git a/TODO.org b/TODO.org new file mode 100644 index 0000000..1d8061a --- /dev/null +++ b/TODO.org @@ -0,0 +1,17 @@ +Tarefas IPE + +** 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. + +Objetivos + +** TODO Implementação Funcional. +** TODO Testes automatizados após o projeto pronto. diff --git a/app.go b/app.go new file mode 100644 index 0000000..308aa9f --- /dev/null +++ b/app.go @@ -0,0 +1,110 @@ +// 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 ( + "errors" + "strings" +) + +// Applications +type App struct { + Name string + AppID string + Key string + Secret string + OnlySSL bool + ApplicationDisabled bool + + PublicChannels []*Channel `json:"-"` + PresenceChannels []*Channel `json:"-"` + PrivateChannels []*Channel `json:"-"` + + Connections []*Connection `json:"-"` +} + +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) + } + + return channels +} + +// Create a new Connection +func (a *App) AddConnection(c *Connection) { + a.Connections = append(a.Connections, c) +} + +func (a *App) FindConnection(socketID string) (*Connection, error) { + for _, c := range a.Connections { + if c.SocketID == socketID { + return c, nil + } + } + + return nil, errors.New("Connection not found") +} + +func (a *App) AddChannel(c *Channel) { + if c.isPresence() { + a.PresenceChannels = append(a.PresenceChannels, c) + } else if c.isPrivate() { + a.PrivateChannels = append(a.PrivateChannels, c) + } else { + a.PublicChannels = append(a.PublicChannels, c) + } +} + +func (a *App) FindOrCreateChannelByChannelID(n, data string) *Channel { + channel, err := a.FindChannelByChannelID(n) + + if err != nil { + channel = NewChannel(n, data) + + a.AddChannel(channel) + } + + return channel +} + +// 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 + } + } + } + + return nil, errors.New("Channel does not exists") +} diff --git a/app_test.go b/app_test.go new file mode 100644 index 0000000..a59af53 --- /dev/null +++ b/app_test.go @@ -0,0 +1,155 @@ +// 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 newApp() App { + return App{Name: "Test", AppID: "123", Key: "123", Secret: "123", OnlySSL: false, ApplicationDisabled: false} +} + +func Test_add_channels(t *testing.T) { + + app := newApp() + + // Public + + if len(app.PublicChannels) != 0 { + t.Error("Length of public channels must be 0 before test") + } + + app.AddChannel(NewChannel("ID", "")) + + if len(app.PublicChannels) != 1 { + t.Error("Length os public channels after insert must be 1") + } + + // Presence + + if len(app.PresenceChannels) != 0 { + t.Error("Length of presence channels must be 0 before test") + } + + app.AddChannel(NewChannel("presence-test", "")) + + if len(app.PresenceChannels) != 1 { + t.Error("Length os presence channels after insert must be 1") + } + + // Private + + if len(app.PrivateChannels) != 0 { + t.Error("Length of private channels must be 0 before test") + } + + app.AddChannel(NewChannel("private-test", "")) + + if len(app.PrivateChannels) != 1 { + t.Error("Length os private channels after insert must be 1") + } + +} + +func Test_New_Connection(t *testing.T) { + app := newApp() + + if len(app.Connections) != 0 { + t.Error("Length of connections before test must be 0") + } + + app.NewConnection("1", "", nil) + + if len(app.Connections) != 1 { + t.Error("Length os connections after test must be 1") + } +} + +func Test_find_connection(t *testing.T) { + app := newApp() + app.NewConnection("1", "", nil) + + conn, err := app.FindConnection("1") + + if err != nil { + t.Error(err) + } + + if conn.SocketID != "1" { + t.Error("Wrong connection.") + } + + // Find a wrong connection + + conn, err = app.FindConnection("DoesNotExists") + + if err == nil { + t.Error("Opps, Must be nil") + } + + if conn != nil { + t.Error("Opps, Must be nil") + } +} + +func Test_find_or_create_channels(t *testing.T) { + app := newApp() + + // Public + if len(app.PublicChannels) != 0 { + t.Error("Length of public channels must be 0 before test") + } + + c := app.FindOrCreateChannelByChannelID("id", "") + + if len(app.PublicChannels) != 1 { + t.Error("Length os public channels after insert must be 1") + } + + if c.ChannelID != "id" { + t.Error("Opps wrong channel") + } + + // Presence + if len(app.PresenceChannels) != 0 { + t.Error("Length of presence channels must be 0 before test") + } + + c = app.FindOrCreateChannelByChannelID("presence-test", "") + + if len(app.PresenceChannels) != 1 { + t.Error("Length os presence channels after insert must be 1") + } + + if c.ChannelID != "presence-test" { + t.Error("Opps wrong channel") + } + + // Private + if len(app.PrivateChannels) != 0 { + t.Error("Length of private channels must be 0 before test") + } + + c = app.FindOrCreateChannelByChannelID("private-test", "") + + if len(app.PrivateChannels) != 1 { + t.Error("Length os private channels after insert must be 1") + } + + if c.ChannelID != "private-test" { + t.Error("Opps wrong channel") + } + +} + +// How to test this??? +func TestRandomID(t *testing.T) { + n := randomID() + + if n < 0 { + t.Error("Must be greater than or equal zero") + } +} diff --git a/auth.go b/auth.go new file mode 100644 index 0000000..54d7f1e --- /dev/null +++ b/auth.go @@ -0,0 +1,88 @@ +// 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 ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "github.com/gorilla/mux" + "net/http" + "net/url" + "sort" + "strings" +) + +// 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 + + for key := range params { + keys = append(keys, strings.ToLower(key)) + } + + sort.Strings(keys) + + var pieces []string + + for _, key := range keys { + pieces = append(pieces, key+"="+params.Get(key)) + } + + return strings.Join(pieces, "&") +} + +// Authenticate pusher +// see: https://gist.github.com/mloughran/376898 +// +// The signature is a HMAC SHA256 hex digest. +// This is generated by signing a string made up of the following components concatenated with newline characters \n. +// +// * The uppercase request method (e.g. POST) +// * The request path (e.g. /some/resource) +// * The query parameters sorted by key, with keys converted to lowercase, then joined as in the query string. +// Note that the string must not be url escaped (e.g. given the keys auth_key: foo, Name: Something else, you get auth_key=foo&name=Something else) +func RestAuthenticationHandler(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + appID := vars["app_id"] + + currentApp, err := Conf.GetAppByAppID(appID) + + if err != nil { + http.Error(w, "Not authorized", http.StatusUnauthorized) + return + } + + params := r.URL.Query() + + authSignature := 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)) { + h.ServeHTTP(w, r) + } else { + http.Error(w, "Not authorized", http.StatusUnauthorized) + } + }) +} diff --git a/config-example.json b/config-example.json new file mode 100644 index 0000000..457901f --- /dev/null +++ b/config-example.json @@ -0,0 +1,23 @@ +{ + "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" + }, + { + "ApplicationDisabled": false, + "OnlySSL": false, + "Secret": "2189389a989a9ab8921f", + "Key": "6a266506823423cf2a1f", + "Name": "Apresentacao", + "AppID": "2" + } + ] +} diff --git a/config.go b/config.go new file mode 100644 index 0000000..3be16d2 --- /dev/null +++ b/config.go @@ -0,0 +1,34 @@ +// 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 ( + "errors" +) + +type ConfigFile struct { + Host string + SessionName string + SessionSecret string + Apps []*App +} + +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") +} + +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") +} diff --git a/conn.go b/conn.go new file mode 100644 index 0000000..05f50df --- /dev/null +++ b/conn.go @@ -0,0 +1,138 @@ +// 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 ( + "errors" + "github.com/gorilla/websocket" + "log" + "strings" + "sync" +) + +// A subscriber +type Connection struct { + Id int + SocketID string + Data string // Extra data attached to this subscriber + Socket *websocket.Conn + Messages chan []byte +} + +// A channel +type Channel struct { + ChannelID string + Data string + Connections []*Connection + Messages chan []byte +} + +// Return true if the channel has at least one subscriber +func (c Channel) isOccupied() bool { + return c.totalConnections() > 0 +} + +// 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) totalConnections() int { + return len(c.Connections) +} + +// Get the total of users. +// For now, totalUsers is equal to totalSubscribers +func (c Channel) totalUsers() int { + return c.totalConnections() +} + +// Add a new subscriber to the channel +func (c *Channel) Subscribe(conn *Connection) { + c.Connections = append(c.Connections, conn) +} + +// Remove the subscriber from the channel +func (c *Channel) Unsubscribe(conn *Connection) error { + index := -1 + + for i, c := range c.Connections { + if c == conn { + index = i + break + } + } + + if index == -1 { + return errors.New("Connections not found") + } + + c.Connections = append(c.Connections[:index], c.Connections[index+1:]...) + + // Remove Channel if necessary + // Close sockets and Channels + + return nil +} + +// Create a new Channel +func NewChannel(channelID, data string) *Channel { + c := &Channel{ChannelID: channelID, Data: data, Messages: make(chan []byte)} + c.Listen() + + return c +} + +var mu = &sync.Mutex{} +var currentID = 0 + +// Generate a New ID +func newID() int { + mu.Lock() + defer mu.Unlock() + currentID += 1 + + return currentID +} + +// Create a new Connection +func NewConnection(socketID, data string, socket *websocket.Conn) *Connection { + id := newID() + + connection := &Connection{Id: id, SocketID: socketID, Data: data, Socket: socket, Messages: make(chan []byte)} + connection.Listen() + + return connection +} + +func (c *Channel) Listen() { + go func() { + select { + case message := <-c.Messages: + // err := c.Socket.WriteMessage(websocket.TextMessage, message) + // log.Println(err) + log.Println(message) + } + + }() +} + +// 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) + } + }() +} diff --git a/constants.go b/constants.go new file mode 100644 index 0000000..9d14228 --- /dev/null +++ b/constants.go @@ -0,0 +1,48 @@ +// 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 () + +// Error Codes +const ( + // 4000 - 4099 + // Indicates an error resulting in the connection being closed by Pusher, + // and that attempting to reconnect using the same parameters will not succeed. + APPLICATION_ONLY_ACCEPTS_SSL = 4000 + APPLICATION_DOES_NOT_EXISTS = 4001 + APPLICATION_DISABLED = 4003 + APPLICATION_IS_OVER_CONNECTION_QUOTA = 4004 // Not Implemented + PATH_NOT_FOUND = 4005 + INVALID_VERSION_STRING_FORMAT = 4006 + UNSUPPORTED_PROTOCOL_VERSION = 4007 + NO_PROTOCOL_VERSION_SUPPLIED = 4008 + + // 4100 - 4199 + // Indicates an error resulting in the connection being closed by Pusher, + // and the client may reconnect after 1s or more + OVER_CAPACITY = 4100 // Not Implemented + + // 4200 - 4299 + // Indicate an error resulting in the connection being closed by Pusher, + // and the client my reconnect immediately + GENERIC_RECONNECT_IMMEDIATELY = 4200 + PONG_REPLY_NOT_RECEIVED = 4201 // Ping was sent to the client, but no reply was received + CLOSED_AFTER_INACTIVITY = 4202 // Client has been inactive for a long time (24 hours) and client does not suppot ping. + + // 4300 - 4399 + // 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 +) + +// Only this version is supported +const SUPPORTED_PROTOCOL_VERSION = 7 + +// // Maximun event size permitted 20 kB +// See: http://blogs.gnome.org/cneumair/2008/09/30/1-kb-1024-bytes-no-1-kb-1000-bytes/ +const MAX_DATA_EVENT_SIZE = 10 * 1000 diff --git a/errors.go b/errors.go new file mode 100644 index 0000000..54d889b --- /dev/null +++ b/errors.go @@ -0,0 +1,118 @@ +// 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 + +// Base interface +type WebsocketError interface { + GetCode() int + GetMsg() string +} + +// Base struct +type BaseWebsocketError struct { + Code int + Msg string +} + +func (e BaseWebsocketError) GetCode() int { + return e.Code +} + +func (e BaseWebsocketError) GetMsg() string { + return e.Msg +} + +// Unsupprted protocol version +type UnsupportedProtocolVersionError struct { + BaseWebsocketError +} + +func NewUnsupportedProtocolVersionError() UnsupportedProtocolVersionError { + return UnsupportedProtocolVersionError{ + BaseWebsocketError{Code: UNSUPPORTED_PROTOCOL_VERSION, Msg: "Unsupported protocol version"}, + } +} + +// The application does not exists +// See the configuration file +type ApplicationDoesNotExistsError struct { + BaseWebsocketError +} + +func NewApplicationDoesNotExistsError() ApplicationDoesNotExistsError { + return ApplicationDoesNotExistsError{ + BaseWebsocketError{Code: APPLICATION_DOES_NOT_EXISTS, Msg: "Could not found an app with the given key"}, + } +} + +// The user did not send the protocol version +type NoProtocolVersionSuppliedError struct { + BaseWebsocketError +} + +func NewNoProtocolVersionSuppliedError() NoProtocolVersionSuppliedError { + return NoProtocolVersionSuppliedError{ + BaseWebsocketError{Code: NO_PROTOCOL_VERSION_SUPPLIED, Msg: "No protocol version supplied"}, + } +} + +// When the application is disabled. +// See the configuration file +type ApplicationDisabledError struct { + BaseWebsocketError +} + +func NewApplicationDisabledError() NoProtocolVersionSuppliedError { + return NoProtocolVersionSuppliedError{ + BaseWebsocketError{Code: APPLICATION_DISABLED, Msg: "Application disabled"}, + } +} + +// When the application only accepts SSL connections +type ApplicationOnlyAccepsSSLError struct { + BaseWebsocketError +} + +func NewApplicationOnlyAccepsSSLError() ApplicationOnlyAccepsSSLError { + return ApplicationOnlyAccepsSSLError{ + BaseWebsocketError{Code: APPLICATION_ONLY_ACCEPTS_SSL, Msg: "Application only accepts SSL connections, reconnect using wss://"}, + } +} + +// When the user send an invalid version +type InvalidVersionStringFormatError struct { + BaseWebsocketError +} + +func NewInvalidVersionStringFormatError() InvalidVersionStringFormatError { + return InvalidVersionStringFormatError{ + BaseWebsocketError{Code: INVALID_VERSION_STRING_FORMAT, Msg: "Invalid version string format"}, + } +} + +// Used when the error was internal +// * Decoding json +// * Writing to output +type GenericReconnectImmediatelyError struct { + BaseWebsocketError +} + +func NewGenericReconnectImmediatelyError() GenericReconnectImmediatelyError { + return GenericReconnectImmediatelyError{ + BaseWebsocketError{Code: GENERIC_RECONNECT_IMMEDIATELY, Msg: "Generic reconnect immediately"}, + } +} + +// When pusher wants to send an Generic error, it only send the message, the code become nil +// Currently I do not know how to send nil, so I send GENERIC_ERROR +type GenericError struct { + BaseWebsocketError +} + +func NewGenericError(msg string) GenericError { + return GenericError{ + BaseWebsocketError{Code: GENERIC_ERROR, Msg: msg}, + } +} diff --git a/events.go b/events.go new file mode 100644 index 0000000..77f97df --- /dev/null +++ b/events.go @@ -0,0 +1,161 @@ +// 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 + +// { +// "event": "pusher:subscribe", +// "data": { +// "channel": "the channel", +// "auth": "the auth", +// "channelData": "extra data" +// } +// } +type SubscribeEventData struct { + Channel string `json:"channel"` + Auth string `json:"auth,omitempty"` + ChannelData string `json:"channelData,omitempty"` +} + +type SubscribeEvent struct { + Event string `json:"event"` + Data SubscribeEventData `json:"data"` +} + +// Create a new subscribe event with the specified channel and data +func NewSubscribeEvent(channel, auth, channelData string) SubscribeEvent { + data := SubscribeEventData{Channel: channel, Auth: auth, ChannelData: channelData} + return SubscribeEvent{Event: "pusher:subscribe", Data: data} +} + +type UnsubscribeEventData struct { + Channel string `json:"channel"` +} + +// { +// "event": "pusher:unsubscribe", +// "data": { +// "channel": "The channel" +// } +// } +type UnsubscribeEvent struct { + Event string `json:"event"` + Data UnsubscribeEventData `json:"data"` +} + +// Create a new unsubscribe event for the specified channel +func NewUnsubscribeEvent(channel string) UnsubscribeEvent { + data := UnsubscribeEventData{Channel: channel} + return UnsubscribeEvent{Event: "pusher:unsubscribe", Data: data} +} + +// { +// "event": "pusher_internal:subscription_succeeded", +// "channel": "the channel" +// } +type SubscriptionSucceededEvent struct { + Event string `json:"event"` + Channel string `json:"channel"` +} + +// Create a new subscription succeed event for the specified channel +func NewSubscriptionSucceededEvent(channel string) SubscriptionSucceededEvent { + return SubscriptionSucceededEvent{Event: "pusher_internal:subscription_succeeded", Channel: channel} +} + +// { +// "event": "pusher:pong", +// "data": {} +// } +type PongEvent struct { + Event string `json:"event"` + Data string `json:"data"` +} + +// Create a new pong event +func NewPongEvent() PongEvent { + return PongEvent{Event: "pusher:pong", Data: "{}"} +} + +// { +// "event": "pusher:ping", +// "data": {} +// } +type PingEvent struct { + Event string `json:"event"` + Data string `json:"data"` +} + +// Create a new ping event +func NewPingEvent() PingEvent { + return PingEvent{Event: "pusher:ping", Data: "{}"} +} + +// { +// "event": "pusher:error", +// "data": { +// "message": "A Message", +// "code": 4000 +// } +// } +type ErrorEventData struct { + Message string `json:"message"` + Code int `json:"code"` +} +type ErrorEvent struct { + Event string `json:"event"` + Data ErrorEventData `json:"data"` +} + +// Create a new error event +func NewErrorEvent(code int, message string) ErrorEvent { + data := ErrorEventData{Message: message, Code: code} + return ErrorEvent{Event: "pusher:error", Data: data} +} + +// { +// "event" : "pusher:connection_established", +// "data" : { +// "socket_id" : "123456", +// "activity_timeout" : 120 +// } +// } +type ConnectionEstablishedEventData struct { + SocketId string `json:"socket_id"` + ActivityTimeout int `json:"activity_timeout"` +} + +type ConnectionEstablishedEvent struct { + Event string `json:"event"` + Data ConnectionEstablishedEventData `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} +} + +// { +// "event": "client-?", +// "channel": "The channel", +// "data": { +// "message": "A Message" +// } +// } +type ClientEventData struct { + Message string `json:"data"` +} + +type ClientEvent struct { + Event string `json:"event"` + Channel string `json:"channel"` + Data ClientEventData `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} +} diff --git a/extra_handlers.go b/extra_handlers.go new file mode 100644 index 0000000..a8592c3 --- /dev/null +++ b/extra_handlers.go @@ -0,0 +1,39 @@ +// 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 ( + "fmt" + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + "net/http" + "os" +) + +// Check if the application is disabled +func RestCheckAppDisabledHandler(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + appID := vars["app_id"] + + currentApp, err := Conf.GetAppByAppID(appID) + + if err != nil { + http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusForbidden) + return + } + + if currentApp.ApplicationDisabled { + http.Error(w, "Application disabled", http.StatusForbidden) + return + } + + h.ServeHTTP(w, r) + }) +} + +func LogHandler(h http.Handler) http.Handler { + return handlers.CombinedLoggingHandler(os.Stdout, h) +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..0661ac1 --- /dev/null +++ b/main.go @@ -0,0 +1,39 @@ +// 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" + "flag" + "io/ioutil" + "log" + "net/http" +) + +var Conf ConfigFile + +func main() { + var filename = flag.String("config", "config.json", "Config file location") + + flag.Parse() + + file, err := ioutil.ReadFile(*filename) + + if err != nil { + log.Fatal(err) + } + + if err := json.Unmarshal(file, &Conf); err != nil { + log.Fatal(err) + } + + router := NewRouter() + + log.Printf("Starting Ipê using config file: '%s'", *filename) + + if err := http.ListenAndServe(Conf.Host, router); err != nil { + log.Fatalln(err) + } +} diff --git a/rest.go b/rest.go new file mode 100644 index 0000000..02745e7 --- /dev/null +++ b/rest.go @@ -0,0 +1,294 @@ +// 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" + "fmt" + "github.com/gorilla/mux" + "log" + "net/http" + "strings" +) + +// An event consists of a name and data (typically JSON) which may be sent to all subscribers to a particular channel or channels. +// This is conventionally known as triggering an event. +// +// The body should contain a Hash of parameters encoded as JSON where data parameter itself is JSON encoded. +// +// Not Implemented: +// Note that these parameters may be provided in the query string, although this is discouraged. +// +// Example: +// +// {"name":"foo","channels":["project-3"],"data":"{\"some\":\"data\"}"} +// +// Response is an empty JSON hash. +// +// 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"` + } + + err := json.NewDecoder(r.Body).Decode(&input) + + if err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + + // The event data should not be larger than 10KB. + if len(input.Data) > MAX_DATA_EVENT_SIZE { + http.Error(w, "Request too large.", http.StatusRequestEntityTooLarge) + return + } + + // Trigger events + + w.WriteHeader(http.StatusOK) + w.Write([]byte("{}")) +} + +// Allows fetching a hash of occupied channels (optionally filtered by prefix), +// and optionally one or more attributes for each channel. +// +// Notes: +// 'user_count' is the only attribute documented on the Pusher API +// +// Example: +// { +// "channels": { +// "presence-foobar": { +// user_count: 42 +// }, +// "presence-another": { +// user_count: 123 +// } +// } +// } +// +// GET /apps/{app_id}/channels +func GetChannels(w http.ResponseWriter, r *http.Request) { + params := r.URL.Query() + vars := mux.Vars(r) + + appID := vars["app_id"] + filter := params.Get("filter_by_prefix") + info := params.Get("info") + + attributes := strings.Split(info, ",") + + requestedUserCount := false + + for _, a := range attributes { + if a == "user_count" { + requestedUserCount = true + } + } + + // 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 && filter != "presence-" { + http.Error(w, "Attribute user_count is restricted to presence channels", http.StatusBadRequest) + return + } + + 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) + } + + channels := make(map[string]interface{}) + + switch filter { + case "presence-": + for _, c := range app.PresenceChannels { + if requestedUserCount { + channels[c.ChannelID] = struct { + UserCount int `json:"user_count"` + }{ + c.totalUsers(), + } + } else { + channels[c.ChannelID] = struct{}{} + } + } + case "public-": + for _, c := range app.PublicChannels { + channels[c.ChannelID] = struct{}{} + } + case "private-": + for _, c := range app.PrivateChannels { + channels[c.ChannelID] = struct{}{} + } + default: + for _, c := range app.AllChannels() { + channels[c.ChannelID] = struct{}{} + } + } + + w.Header().Set("Content-Type", "application/json;charset=UTF-8") + + if err := json.NewEncoder(w).Encode(channels); err != nil { + http.Error(w, "Internal server error", http.StatusInternalServerError) + log.Println(err) + } +} + +// Fetch info for one channel +// +// Example: +// { +// occupied: true, +// user_count: 42, +// subscription_count: 42 +// } +// +// GET /apps/{app_id}/channels/{channel_name} +func GetChannel(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json;charset=UTF-8") + + params := r.URL.Query() + 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) + } + + channelName := vars["channel_name"] + + // Channel name could not be empty + if strings.TrimSpace(channelName) == "" { + http.Error(w, "Empty channel name", http.StatusBadRequest) + return + } + + info := params.Get("info") + attributes := strings.Split(info, ",") + + // Attributes requested + requestedUserCount := false + requestedSubscriptionCount := false + + for _, a := range attributes { + switch a { + case "subscription_count": + requestedSubscriptionCount = true + case "user_count": + requestedUserCount = true + } + } + + // Check the kind of channel + + channel, err := app.FindChannelByChannelID(channelName) + + // Channel exists? + if err != nil { + http.Error(w, fmt.Sprintf("Could not find a channel with id %s", channelName), http.StatusBadRequest) + return + } + + // 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() { + http.Error(w, "Attribute user_count is restricted to presence channels", http.StatusBadRequest) + return + } + + // Output + dtoChannel := struct { + Occupied bool `json:"occupied"` + UserCount int `json:"user_count,omitempty"` + SubscriptionCount int `json:"subscription_count,omitempty"` + }{Occupied: channel.isOccupied()} + + switch { + case requestedSubscriptionCount && requestedUserCount: + dtoChannel.UserCount = channel.totalUsers() + dtoChannel.SubscriptionCount = channel.totalConnections() + + case requestedUserCount: + dtoChannel.UserCount = channel.totalUsers() + + case requestedSubscriptionCount: + dtoChannel.SubscriptionCount = channel.totalConnections() + } + + w.Header().Set("Content-Type", "application/json;charset=UTF-8") + + if err := json.NewEncoder(w).Encode(dtoChannel); err != nil { + http.Error(w, "Internal server error", http.StatusInternalServerError) + log.Println(err) + } +} + +// Allowed only to presence-channels +// +// Example: +// { +// "users": [ +// { "id": 1 }, +// { "id": 2 } +// ] +// } +// +// GET /apps/{app_id}/channels/{channel_name}/users +func GetChannelUsers(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + + appID := vars["app_id"] + channelName := vars["channel_name"] + + isPresence := strings.HasPrefix(channelName, "presence-") + + if !isPresence { + http.Error(w, "This api endpoint is restricted to presence channels.", http.StatusBadRequest) + return + } + + 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) + } + + // Get the channel + channel, err := app.FindChannelByChannelID(channelName) + + // Channel exists? + if err != nil { + http.Error(w, fmt.Sprintf("Could not find a channel with id %s", channelName), http.StatusBadRequest) + return + } + + result := make(map[string][]interface{}) + + var users []interface{} + + for _, s := range channel.Connections { + users = append(users, struct { + Id int `json:"id"` + }{s.Id}) + } + + result["users"] = users + + w.Header().Set("Content-Type", "application/json;charset=UTF-8") + + if err := json.NewEncoder(w).Encode(result); err != nil { + http.Error(w, "Internal server error", http.StatusInternalServerError) + log.Println(err) + } +} diff --git a/rest_test.go_ b/rest_test.go_ new file mode 100644 index 0000000..5e93f8e --- /dev/null +++ b/rest_test.go_ @@ -0,0 +1,158 @@ +// 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" + "net/http" + "net/http/httptest" + "testing" +) + +func init() { + // Conf = NewConfig(":8080", "123456", "APPID", "Secret", false, false) + + channel := NewChannel("presence-c1", "") + channel.addSubscriber(Subscriber{Id: 1, SocketID: "Sock1", Data: "Data1"}) + channel.addSubscriber(Subscriber{Id: 2, SocketID: "Sock2", Data: "Data2"}) + + PresenceChannels["presence-c1"] = channel + + PrivateChannels["private-c3"] = NewChannel("private-c3", "") + PublicChannels["c2"] = NewChannel("c2", "") + +} + +// All Channels +func Test_GetChannels_all(t *testing.T) { + r, _ := http.NewRequest("GET", "/apps/APPID/channels", nil) + w := httptest.NewRecorder() + + NewRouter().ServeHTTP(w, r) + + if w.Code != http.StatusOK { + t.Errorf("Must return OK: %s returned", w.Code) + } + + channels := make(map[string]interface{}) + json.Unmarshal(w.Body.Bytes(), &channels) + + if len(channels) != 3 { + t.Error("Must return 3 channels") + } +} + +// Only presence channels +func Test_GetChannels_filter_by_presence_prefix(t *testing.T) { + r, _ := http.NewRequest("GET", "/apps/APPID/channels?filter_by_prefix=presence-", nil) + w := httptest.NewRecorder() + + NewRouter().ServeHTTP(w, r) + + if w.Code != http.StatusOK { + t.Errorf("Must return OK: %s returned", w.Code) + } + + channels := make(map[string]interface{}) + json.Unmarshal(w.Body.Bytes(), &channels) + + if len(channels) != 1 { + t.Error("Must return 1 channel") + } +} + +// Only presence channels and user_count +func Test_GetChannels_filter_by_presence_prefix_and_user_count(t *testing.T) { + r, _ := http.NewRequest("GET", "/apps/APPID/channels?filter_by_prefix=presence-&info=user_count", nil) + w := httptest.NewRecorder() + + NewRouter().ServeHTTP(w, r) + + if w.Code != http.StatusOK { + t.Errorf("Must return OK: %s returned", w.Code) + } + + channels := make(map[string]struct { + UserCount int `json:"user_count"` + }) + + json.Unmarshal(w.Body.Bytes(), &channels) + + if len(channels) != 1 { + t.Error("Must return 1 channel") + } + + channel, exists := channels["presence-c1"] + + if !exists { + t.Error("Channel must exist.") + } + + if channel.UserCount != 2 { + t.Error("Must be 2 users") + } +} + +// User count only alowed in Presence channels +func Test_GetChannels_filter_by_private_prefix_and_info_user_count(t *testing.T) { + r, _ := http.NewRequest("GET", "/apps/APPID/channels?filter_by_prefix=private-&info=user_count", nil) + w := httptest.NewRecorder() + + NewRouter().ServeHTTP(w, r) + + if w.Code != http.StatusBadRequest { + t.Errorf("Must return BadRequest: %s returned", w.Code) + } +} + +func Test_GetChannels_filter_by_public_prefix(t *testing.T) { + r, _ := http.NewRequest("GET", "/apps/APPID/channels?filter_by_prefix=public-", nil) + w := httptest.NewRecorder() + + NewRouter().ServeHTTP(w, r) + + if w.Code != http.StatusOK { + t.Errorf("Must return OK: %s returned", w.Code) + } + + channels := make(map[string]interface{}) + + json.Unmarshal(w.Body.Bytes(), &channels) + + if len(channels) != 1 { + t.Error("Must return 1 channel") + } + + _, exists := channels["c2"] + + if !exists { + t.Error("Channel must exist.") + } +} + +func Test_GetChannels_filter_by_private_prefix(t *testing.T) { + r, _ := http.NewRequest("GET", "/apps/APPID/channels?filter_by_prefix=private-", nil) + w := httptest.NewRecorder() + + NewRouter().ServeHTTP(w, r) + + if w.Code != http.StatusOK { + t.Errorf("Must return OK: %s returned", w.Code) + } + + channels := make(map[string]interface{}) + + json.Unmarshal(w.Body.Bytes(), &channels) + + if len(channels) != 1 { + t.Error("Must return 1 channel") + } + + _, exists := channels["private-c3"] + + if !exists { + t.Error("Channel must exist.") + } +} diff --git a/router.go b/router.go new file mode 100644 index 0000000..32b584c --- /dev/null +++ b/router.go @@ -0,0 +1,31 @@ +// 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 ( + "github.com/gorilla/mux" + "net/http" +) + +func NewRouter() *mux.Router { + router := mux.NewRouter().StrictSlash(true) + + for _, route := range routes { + var handler http.Handler + + handler = route.HandlerFunc + + if route.RequiresRestAuth { + handler = RestAuthenticationHandler(handler) + handler = RestCheckAppDisabledHandler(handler) + } + + handler = LogHandler(handler) + + router.Methods(route.Method).Path(route.Pattern).Name(route.Name).Handler(handler) + } + + return router +} diff --git a/routes.go b/routes.go new file mode 100644 index 0000000..feddbf4 --- /dev/null +++ b/routes.go @@ -0,0 +1,57 @@ +// 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 ( + "net/http" +) + +type Route struct { + Name string + Method string + Pattern string + HandlerFunc http.HandlerFunc + RequiresRestAuth bool +} + +type Routes []Route + +var routes = Routes{ + Route{ + "PostEvents", + "POST", + "/apps/{app_id}/events", + PostEvents, + true, + }, + Route{ + "GetChannels", + "GET", + "/apps/{app_id}/channels", + GetChannels, + true, + }, + Route{ + "GetChannel", + "GET", + "/apps/{app_id}/channels/{channel_name}", + GetChannel, + true, + }, + Route{ + "GetChannelUsers", + "GET", + "/apps/{app_id}/channels/{channel_name}/users", + GetChannelUsers, + true, + }, + Route{ + "Websocket", + "GET", + "/app/{key}", + Websocket, + false, + }, +} diff --git a/websockets.go b/websockets.go new file mode 100644 index 0000000..2bc1047 --- /dev/null +++ b/websockets.go @@ -0,0 +1,206 @@ +// 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" + "fmt" + "github.com/gorilla/mux" + "github.com/gorilla/sessions" + "github.com/gorilla/websocket" + "log" + "net/http" + "strconv" + "strings" +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, + 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 { + params := r.URL.Query() + p := params.Get("protocol") + + protocol, err := strconv.Atoi(p) + + if err != nil { + return NewInvalidVersionStringFormatError() + } + + switch { + case strings.TrimSpace(p) == "": + return NewNoProtocolVersionSuppliedError() + case protocol != SUPPORTED_PROTOCOL_VERSION: + return NewUnsupportedProtocolVersionError() + case app.ApplicationDisabled: + return NewApplicationDisabledError() + case r.TLS != nil: + if app.OnlySSL { + return NewApplicationOnlyAccepsSSLError() + } + } + + // Create the new connection + connection := NewConnection(session.ID, "", conn) + app.AddConnection(connection) + + // Everything went fine. Huhu. + if err := conn.WriteJSON(NewConnectionEstablishedEvent(connection.SocketID)); err != nil { + return NewGenericReconnectImmediatelyError() + } + + return nil +} + +// Handle messages +func onMessage(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, session *sessions.Session, app *App) WebsocketError { + var event struct { + Event string `json:"event"` + } + + for { + _, message, err := conn.ReadMessage() + + if err != nil { + return NewGenericReconnectImmediatelyError() + } + + if err := json.Unmarshal(message, &event); err != nil { + return NewGenericReconnectImmediatelyError() + } + + switch event.Event { + case "pusher:ping": + if err := conn.WriteJSON(NewPongEvent()); err != nil { + return NewGenericReconnectImmediatelyError() + } + case "pusher:subscribe": + subscribeEvent := SubscribeEvent{} + + if err := json.Unmarshal(message, &subscribeEvent); err != nil { + return NewGenericReconnectImmediatelyError() + } + + connection, err := app.FindConnection(session.ID) + + if err != nil { + return NewGenericReconnectImmediatelyError() + } + + channelName := strings.TrimSpace(subscribeEvent.Data.Channel) + + // Authentication + if strings.HasPrefix(channelName, "presence-") { + toSign := fmt.Sprintf("%s:%s:%s", connection.SocketID, channelName, subscribeEvent.Data.ChannelData) + + if subscribeEvent.Data.Auth != HashMAC([]byte(toSign), []byte(app.Secret)) { + return NewGenericError(fmt.Sprintf("Auth value for subscription to %s is invalid", channelName)) + } + } 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)) + } + } + + channel := app.FindOrCreateChannelByChannelID(channelName, subscribeEvent.Data.ChannelData) + channel.Subscribe(connection) + + if err := conn.WriteJSON(NewSubscriptionSucceededEvent(channel.ChannelID)); err != nil { + return NewGenericReconnectImmediatelyError() + } + case "pusher:unsubscribe": + unsubscribeEvent := UnsubscribeEvent{} + + if err := json.Unmarshal(message, &unsubscribeEvent); err != nil { + return NewGenericReconnectImmediatelyError() + } + + connection, err := app.FindConnection(session.ID) + + if err != nil { + return NewGenericError(fmt.Sprintf("Could not find a connection with the id %s", session.ID)) + } + + 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)) + } + + if err := channel.Unsubscribe(connection); err != nil { + return NewGenericReconnectImmediatelyError() + } + } + + return nil + // Client Events + } +} + +// Websocket GET /app/{key} +func Websocket(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + + 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) + emitWSError(NewGenericReconnectImmediatelyError(), conn) + return + } + + vars := mux.Vars(r) + appKey := vars["key"] + + app, err := Conf.GetAppByKey(appKey) + + if err != nil { + log.Println(err) + emitWSError(NewApplicationDoesNotExistsError(), conn) + return + } + + if err := onOpen(conn, w, r, session, app); err != nil { + emitWSError(err, conn) + return + } + + if err := onMessage(conn, w, r, session, app); err != nil { + emitWSError(err, conn) + + // Find the connection in app and destroy it + return + } +} + +// 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) + } + + conn.Close() +}