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

Compare commits

base: talksik/ipe:v1.2.0
Branches Tags
talksik/ipe:master talksik/ipe:better-errors talksik/ipe:gh-pages talksik/ipe:black-box-test
talksik/ipe:v1.2.0 talksik/ipe:v1.1.0 talksik/ipe:1.0.0
...
compare: talksik/ipe:better-errors
Branches Tags
talksik/ipe:master talksik/ipe:better-errors talksik/ipe:gh-pages talksik/ipe:black-box-test
talksik/ipe:v1.2.0 talksik/ipe:v1.1.0 talksik/ipe:1.0.0

13 Commits
v1.2.0 ... better-errors

Author SHA1 Message Date
claudemiro 722e0ad3e3 WIP: Cleaning the error handling... 2016-10-02 23:01:12 -03:00
claudemiro 065d6ab286 Removing unused paremeters 2016-09-29 23:30:28 -03:00
somebody 1d1ffa00b5 correcting typo in README.md: instalation ->installation (#33) 2016-09-27 21:00:29 -03:00
claudemiro 983f490737 Params not necessary 2016-09-01 23:43:09 -03:00
claudemiro 5c9528bce4 Reverting to slicebased memdb after checking benchmark. 2016-09-01 23:42:55 -03:00
claudemiro 8fb96f3cbf Simpler logic 2016-09-01 23:42:18 -03:00
claudemiro 9d732fb01a Added option to enable pprof. 2016-09-01 21:03:29 -03:00
claudemiro 0bd4fa02e9 Using method references for Mux 2016-08-22 23:20:30 -03:00
claudemiro 4a305b32a6 Removed GoJi in favor of Pressly Chi 2016-08-22 23:06:15 -03:00
Claudemiro 4c5d5302ec Basic context implementation in webhooks. 2016-08-21 10:22:35 -03:00
Claudemiro 994e8e00f1 Show only a log message when could not read conf 2016-08-14 21:00:47 -03:00
Claudemiro 9ad14daeeb Simplified the websockets handler code (#30) 
* Simplified the websockets handler code

* created function validateAuthKey

* Using go default error interface.
 2016-08-13 12:25:38 -03:00
claudemiro 0ac6c7a55c Started working on v1.3.0 2016-08-11 22:00:19 -03:00
22 changed files with 532 additions and 617 deletions
Expand all files Collapse all files
README.md
+3 -2
View File
@@ -18,7 +18,7 @@ This software is written in Go - the WYSIWYG lang
* Web Hooks; * Web Hooks;
* Client events; * Client events;
* Complete REST API; * Complete REST API;
* Easy instalation; * Easy installation;
* A single binary without dependencies; * A single binary without dependencies;
* Easy configuration; * Easy configuration;
* Protocol version 7; * Protocol version 7;
@@ -50,7 +50,8 @@ $ go install github.com/dimiro1/ipe
```javascript ```javascript
{ {
"Host": ":8080", // Required "Host": ":8080", // Required
"SSL": false, // Required but can be false "SSL": false, // Not Required, default is false
"Profiling": false, // Mount pprof at /debug. Not Required, default is false
"SSLHost": ":4433", // Required if SSL is true "SSLHost": ":4433", // Required if SSL is true
"SSLKeyFile": "A key.pem file", // Required if SSL is true "SSLKeyFile": "A key.pem file", // Required if SSL is true
"SSLCertFile": "A cert.pem file", // Required if SSL is true "SSLCertFile": "A cert.pem file", // Required if SSL is true
Rakefile
+1 -1
View File
@@ -4,7 +4,7 @@
require 'rake/clean' require 'rake/clean'
VERSION = 'v1.2.0' VERSION = 'v1.3.0-SNAPSHOT'
GITHASH = `git rev-parse --short HEAD` GITHASH = `git rev-parse --short HEAD`
DATE = Time.now.strftime '%Y%m%d%H%M%S' DATE = Time.now.strftime '%Y%m%d%H%M%S'
TODO.org
-41
View File
@@ -1,41 +0,0 @@
IPÊ
---
* TODO [12/14]
* [X] Autenticação API Rest
* [X] Autenticação Websockets
* [X] Ping e Pong
* [ ] Escrever testes automatizados
* [X] SSL
* [X] Expvar - Canais, inscritos
* [X] Otimizações [3/3]
* [X] Refatorar partes do código, remover repetições
* [X] Alterar tipos de dados de slices para mapas em alguns locais.
* [X] Remover Canais vazios.
* [ ] 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
* [X] Events Presence channels [3/3]
* [X] pusher_internal:subscription_succeeded para canais de presença
* [X] pusher_internal:member_added
* [X] 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
* [X] Escrever um README
* Dúvidas
* [X] Atualmente o ID do usuário em canais de presença está como string, o protocolo não deixa claro, já ví exemplos onde a chave é um inteiro. Nesse caso essa chave terá que ser um RawMessage, O campo Data inteiro??
* Objetivos [7/7]
* [X] Implementação Funcional.
* [X] WebHooks
* [X] Presence channels
* [X] Private Channels
* [X] Public Channels
* [X] Easy Instalation
* [X] Easy configuration
functional/SpecRunner.html
+12 -12
View File
@@ -1,24 +1,24 @@
<html> <html>
<head> <head>
<meta charset="utf-8"> <meta charset="utf-8">
<title>Pusher Spec</title> <title>Pusher Spec</title>
<link href="https://cdn.rawgit.com/mochajs/mocha/2.2.5/mocha.css" rel="stylesheet" /> <link href="https://cdn.rawgit.com/mochajs/mocha/2.2.5/mocha.css" rel="stylesheet"/>
</head> </head>
<body> <body>
<div id="mocha"></div> <div id="mocha"></div>
<script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/2.2.0/jquery.min.js"></script> <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/2.2.0/jquery.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/mocha/2.3.4/mocha.min.js"></script> <script src="https://cdnjs.cloudflare.com/ajax/libs/mocha/2.3.4/mocha.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/chai/3.4.1/chai.min.js"></script> <script src="https://cdnjs.cloudflare.com/ajax/libs/chai/3.4.1/chai.min.js"></script>
<script src="//js.pusher.com/3.2/pusher.min.js"></script> <script src="//js.pusher.com/3.2/pusher.min.js"></script>
<script>mocha.setup('bdd')</script> <script>mocha.setup('bdd')</script>
<script src="test.pusher.js"></script> <script src="test.pusher.js"></script>
<script> <script>
mocha.checkLeaks(); mocha.checkLeaks();
mocha.globals(['jQuery', 'Pusher']); mocha.globals(['jQuery', 'Pusher']);
mocha.run(); mocha.run();
</script> </script>
</body> </body>
</html> </html>
functional/client.go
+4 -4
View File
@@ -34,7 +34,7 @@ func pusherPresenceAuth(res http.ResponseWriter, req *http.Request) {
panic(err) panic(err)
} }
fmt.Fprintf(res, string(response)) fmt.Fprint(res, string(response))
} }
func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) { func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) {
@@ -48,13 +48,13 @@ func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) {
panic(err) panic(err)
} }
fmt.Fprintf(res, string(response)) fmt.Fprint(res, string(response))
} }
func triggerMessage(res http.ResponseWriter, req *http.Request) { func triggerMessage(res http.ResponseWriter, _ *http.Request) {
client.Trigger("private-messages", "messages", "The message from server") client.Trigger("private-messages", "messages", "The message from server")
fmt.Fprintf(res, "OK") fmt.Fprint(res, "OK")
} }
func main() { func main() {
functional/functional-config.json
+19 -18
View File
@@ -1,20 +1,21 @@
{ {
"Host": ":8080", "Host": ":8080",
"Encrypted": false, "SSL": false,
"SSLHost": ":8090", "Profiling": true,
"SSLKeyFile": "key.pem", "SSLHost": ":8090",
"SSLCertFile": "cert.pem", "SSLKeyFile": "key.pem",
"Apps": [ "SSLCertFile": "cert.pem",
{ "Apps": [
"ApplicationDisabled": false, {
"OnlySSL": false, "ApplicationDisabled": false,
"Secret": "7ad3753142a6693b25b9", "OnlySSL": false,
"Key": "278d525bdf162c739803", "Secret": "7ad3753142a6693b25b9",
"Name": "App for Functional Test", "Key": "278d525bdf162c739803",
"AppID": "1", "Name": "App for Functional Test",
"UserEvents": true, "AppID": "1",
"WebHooks": false, "UserEvents": true,
"URLWebHook": "http://127.0.0.1:4567/php/hook.php" "WebHooks": false,
} "URLWebHook": "http://127.0.0.1:4567/php/hook.php"
] }
]
} }
glide.lock
Generated
+5 -13
View File
@@ -1,20 +1,12 @@
hash: 0a451841a1f9493d419d658fab8da6863ef5df0645ddea506a225865078c9dcb hash: ee2de935e70f2a39cbfd2dd16add0736c698b32f5c02409d0c25844d883c91a1
updated: 2016-08-10T22:23:10.989237578-03:00 updated: 2016-08-22T22:55:07.279113421-03:00
imports: imports:
- name: github.com/golang/glog - name: github.com/golang/glog
version: 23def4e6c14b4da8ac2ed8007337bc5eb5007998 version: 23def4e6c14b4da8ac2ed8007337bc5eb5007998
- name: github.com/gorilla/websocket - name: github.com/gorilla/websocket
version: a69d25be2fe2923a97c2af6849b2f52426f68fc0 version: a69d25be2fe2923a97c2af6849b2f52426f68fc0
- name: github.com/pusher/pusher-http-go - name: github.com/pressly/chi
version: 2bba5f217f6f0f4f0c0a9bb11b945b206b32bec5 version: 12aad88c7d86de2affe686f855b6ed94a07cba9c
- name: goji.io
version: e355964ac565b94cf0fc7f218346626529125086
subpackages: subpackages:
- pat - middleware
- pattern
- internal
- name: golang.org/x/net
version: 075e191f18186a8ff2becaf64478e30f4545cdad
subpackages:
- context
testImports: [] testImports: []
glide.yaml
+3 -7
View File
@@ -2,10 +2,6 @@ package: github.com/dimiro1/ipe
import: import:
- package: github.com/golang/glog - package: github.com/golang/glog
- package: github.com/gorilla/websocket - package: github.com/gorilla/websocket
- package: github.com/pusher/pusher-http-go - package: github.com/pressly/chi
- package: goji.io excludeDirs:
subpackages: - functional
- pat
- package: golang.org/x/net
subpackages:
- context
ipe/channel.go
+43 -43
View File
@@ -74,51 +74,51 @@ func (c *channel) Subscribe(a *app, conn *connection, channelData string) error
subscription := newSubscription(conn, channelData) subscription := newSubscription(conn, channelData)
c.Subscriptions[conn.SocketID] = subscription c.Subscriptions[conn.SocketID] = subscription
if c.IsPresence() { if !c.IsPresence() {
// User Info Data
var info struct {
UserID string `json:"user_id"`
UserInfo json.RawMessage `json:"user_info"`
}
log.Infof("%+v", channelData)
if err := json.Unmarshal([]byte(channelData), &info); err != nil {
log.Error(err)
return err
}
js, err := info.UserInfo.MarshalJSON()
if err != nil {
log.Error(err)
return err
}
// Update the Subscription
subscription.ID = info.UserID
subscription.Data = string(js)
// Publish pusher_internal:member_added
c.PublishMemberAddedEvent(a, channelData, subscription)
// WebHook
a.TriggerMemberAddedHook(c, subscription)
// pusher_internal:subscription_succeeded
data := make(map[string]subscriptionSucceeedEventPresenceData)
data["presence"] = newSubscriptionSucceedEventPresenceData(c)
js, err = json.Marshal(data)
if err != nil {
log.Error(err)
return err
}
conn.Publish(newSubscriptionSucceededEvent(c.ChannelID, string(js)))
} else {
conn.Publish(newSubscriptionSucceededEvent(c.ChannelID, "{}")) conn.Publish(newSubscriptionSucceededEvent(c.ChannelID, "{}"))
return nil
} }
// User Info Data
var info struct {
UserID string `json:"user_id"`
UserInfo json.RawMessage `json:"user_info"`
}
log.Infof("%+v", channelData)
if err := json.Unmarshal([]byte(channelData), &info); err != nil {
log.Error(err)
return err
}
js, err := info.UserInfo.MarshalJSON()
if err != nil {
log.Error(err)
return err
}
// Update the Subscription
subscription.ID = info.UserID
subscription.Data = string(js)
// Publish pusher_internal:member_added
c.PublishMemberAddedEvent(a, channelData, subscription)
// WebHook
a.TriggerMemberAddedHook(c, subscription)
// pusher_internal:subscription_succeeded
data := make(map[string]subscriptionSucceeedEventPresenceData)
data["presence"] = newSubscriptionSucceedEventPresenceData(c)
js, err = json.Marshal(data)
if err != nil {
log.Error(err)
return err
}
conn.Publish(newSubscriptionSucceededEvent(c.ChannelID, string(js)))
// WebHook // WebHook
if c.TotalSubscriptions() == 1 { if c.TotalSubscriptions() == 1 {
ipe/config-example.json
+19 -18
View File
@@ -1,20 +1,21 @@
{ {
"Host": ":8080", "Host": ":8080",
"SSL": false, "SSL": false,
"SSLHost": ":4433", "Profiling": false,
"SSLKeyFile": "A key.pem file", "SSLHost": ":4433",
"SSLCertFile": "A cert.pem file", "SSLKeyFile": "A key.pem file",
"Apps": [ "SSLCertFile": "A cert.pem file",
{ "Apps": [
"ApplicationDisabled": false, {
"Secret": "A really secret random string", "ApplicationDisabled": false,
"Key": "A random Key string", "Secret": "A really secret random string",
"OnlySSL": false, "Key": "A random Key string",
"Name": "The app name", "OnlySSL": false,
"AppID": "The app ID", "Name": "The app name",
"UserEvents": true, "AppID": "The app ID",
"WebHooks": true, "UserEvents": true,
"URLWebHook": "Some URL to send webhooks" "WebHooks": true,
} "URLWebHook": "Some URL to send webhooks"
] }
]
} }
ipe/config.go
+1
View File
@@ -9,6 +9,7 @@ type configFile struct {
Host string // The host, eg: :8080 will start on 0.0.0.0:8080 Host string // The host, eg: :8080 will start on 0.0.0.0:8080
User string User string
SSL bool SSL bool
Profiling bool
SSLHost string SSLHost string
SSLKeyFile string SSLKeyFile string
SSLCertFile string SSLCertFile string
ipe/constants.go
-35
View File
@@ -4,41 +4,6 @@
package ipe package ipe
// 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.
applicationOnlyAcceptsSSL = 4000
applicationDoesNotExists = 4001
applicationDisabled = 4003
applicationIsOverConnectionQuota = 4004 // Not Implemented
pathNotFound = 4005 // Not Implemented
invalidVersionStringFormat = 4006
unsupportedProtocolVersion = 4007
noProtocolVersionSupplied = 4008
// 4100 - 4199
// Indicates an error resulting in the connection being closed by Pusher,
// and the client may reconnect after 1s or more
overCapacity = 4100 // Not Implemented
// 4200 - 4299
// Indicate an error resulting in the connection being closed by Pusher,
// and the client my reconnect immediately
genericReconnectImmediately = 4200
pongReplyNotReceived = 4201 // Ping was sent to the client, but no reply was received; Not Implemented
closedAfterInactivity = 4202 // Client has been inactive for a long time (24 hours) and client does not suppot ping.; Not Implemented
// 4300 - 4399
// Any other type of error
clientRejectedDueToRateLimit = 4301 // Not Implemented
// Pusher send null, This app use this error code to send the null value
// see ErrorEvent
otherError = 0
)
// Only this version is supported // Only this version is supported
const supportedProtocolVersion = 7 const supportedProtocolVersion = 7
ipe/db.go
+15 -27
View File
@@ -18,50 +18,38 @@ type db interface {
AddApp(*app) error AddApp(*app) error
} }
// memdb is an in memory implementation of db interface
type memdb struct { type memdb struct {
IDMutex sync.Mutex sync.Mutex
KeyMutex sync.Mutex Apps []*app
AppsByAppID map[string]*app
AppsByKey map[string]*app
} }
func newMemdb() *memdb { func newMemdb() db {
return &memdb{ return &memdb{}
AppsByAppID: make(map[string]*app),
AppsByKey: make(map[string]*app),
}
} }
func (db *memdb) AddApp(a *app) error { func (db *memdb) AddApp(a *app) error {
db.IDMutex.Lock() db.Lock()
db.AppsByAppID[a.AppID] = a db.Apps = append(db.Apps, a)
db.IDMutex.Unlock() db.Unlock()
db.KeyMutex.Lock()
db.AppsByKey[a.Key] = a
db.KeyMutex.Unlock()
return nil return nil
} }
// GetAppByAppID returns an App with by appID // GetAppByAppID returns an App with by appID
func (db *memdb) GetAppByAppID(appID string) (*app, error) { func (db *memdb) GetAppByAppID(appID string) (*app, error) {
db.IDMutex.Lock() for _, a := range db.Apps {
a, ok := db.AppsByAppID[appID] if a.AppID == appID {
db.IDMutex.Unlock() return a, nil
if ok { }
return a, nil
} }
return nil, errors.New("App not found") return nil, errors.New("App not found")
} }
// GetAppByKey returns an App with by key // GetAppByKey returns an App with by key
func (db *memdb) GetAppByKey(key string) (*app, error) { func (db *memdb) GetAppByKey(key string) (*app, error) {
db.KeyMutex.Lock() for _, a := range db.Apps {
a, ok := db.AppsByKey[key] if a.Key == key {
db.KeyMutex.Unlock() return a, nil
if ok { }
return a, nil
} }
return nil, errors.New("App not found") return nil, errors.New("App not found")
} }
ipe/db_test.go
+13
View File
@@ -6,6 +6,19 @@ package ipe
import "testing" import "testing"
func Benchmark_memdb_GetAppByAppID(b *testing.B) {
db := newMemdb()
db.AddApp(&app{AppID: "123456", Name: "Example"})
db.AddApp(&app{AppID: "654321", Name: "Example2"})
db.AddApp(&app{AppID: "678901", Name: "Example3"})
b.ResetTimer()
for i := 0; i < b.N; i++ {
db.GetAppByAppID("123456")
}
}
func Test_db_GetAppByAppID(t *testing.T) { func Test_db_GetAppByAppID(t *testing.T) {
app := &app{AppID: "123456", Name: "Example"} app := &app{AppID: "123456", Name: "Example"}
ipe/errors.go
+34 -88
View File
@@ -4,115 +4,61 @@
package ipe package ipe
// Base interface import "fmt"
type websocketError interface {
GetCode() int
GetMsg() string
}
// Base struct // Base struct
type baseWebsocketError struct { type websocketError struct {
Code int Code *int
Msg string Msg string
} }
func (e baseWebsocketError) GetCode() int { func (e websocketError) GetCode() *int {
return e.Code return e.Code
} }
func (e baseWebsocketError) GetMsg() string { func (e websocketError) GetMsg() string {
return e.Msg return e.Msg
} }
// Unsupprted protocol version func (e websocketError) Error() string {
type unsupportedProtocolVersionError struct { return fmt.Sprintf("%d: %s", e.Code, e.Msg)
baseWebsocketError
} }
func newUnsupportedProtocolVersionError() unsupportedProtocolVersionError { func newWebsocketError(code int, msg string) websocketError {
return unsupportedProtocolVersionError{ return websocketError{Code: &code, Msg: msg}
baseWebsocketError{Code: unsupportedProtocolVersion, Msg: "Unsupported protocol version"},
}
} }
// The application does not exists var (
// See the configuration file // Unsupprted protocol version
type applicationDoesNotExistsError struct { unsupportedProtocolVersionError = newWebsocketError(4007, "Unsupported protocol version")
baseWebsocketError
}
func newApplicationDoesNotExistsError() applicationDoesNotExistsError { // The application does not exists
return applicationDoesNotExistsError{ // See the configuration file
baseWebsocketError{Code: applicationDoesNotExists, Msg: "Could not found an app with the given key"}, applicationDoesNotExistsError = newWebsocketError(4001, "Could not found an app with the given key")
}
}
// The user did not send the protocol version // The user did not send the protocol version
type noProtocolVersionSuppliedError struct { noProtocolVersionSuppliedError = newWebsocketError(4008, "No protocol version supplied")
baseWebsocketError
}
func newNoProtocolVersionSuppliedError() noProtocolVersionSuppliedError { // When the application is disabled.
return noProtocolVersionSuppliedError{ // See the configuration file
baseWebsocketError{Code: noProtocolVersionSupplied, Msg: "No protocol version supplied"}, applicationDisabledError = newWebsocketError(4003, "Application disabled")
}
}
// When the application is disabled. // When the application only accepts SSL connections
// See the configuration file applicationOnlyAccepsSSLError = newWebsocketError(4000, "Application only accepts SSL connections, reconnect using wss://")
type applicationDisabledError struct {
baseWebsocketError
}
func newApplicationDisabledError() noProtocolVersionSuppliedError { // When the user send an invalid version
return noProtocolVersionSuppliedError{ invalidVersionStringFormatError = newWebsocketError(4006, "Invalid version string format")
baseWebsocketError{Code: applicationDisabled, Msg: "Application disabled"},
}
}
// When the application only accepts SSL connections // Used when the error was internal
type applicationOnlyAccepsSSLError struct { // * Decoding json
baseWebsocketError // * Writing to output
} genericReconnectImmediatelyError = newWebsocketError(4200, "Generic reconnect immediately")
func newApplicationOnlyAccepsSSLError() applicationOnlyAccepsSSLError { // When pusher wants to send an Generic error, it only send the message, the code become nil
return applicationOnlyAccepsSSLError{ // Currently I do not know how to send nil, so I send GENERIC_ERROR
baseWebsocketError{Code: applicationOnlyAcceptsSSL, Msg: "Application only accepts SSL connections, reconnect using wss://"}, genericError = newWebsocketError(0, "Generic Error")
}
}
// When the user send an invalid version disabledClientEventsError = websocketError{Msg: "To send client events, you must enable this feature in the Settings."}
type invalidVersionStringFormatError struct {
baseWebsocketError
}
func newInvalidVersionStringFormatError() invalidVersionStringFormatError { couldNotFoundChannelError = websocketError{Msg: "Could not find a channel with the given id"}
return invalidVersionStringFormatError{ )
baseWebsocketError{Code: invalidVersionStringFormat, 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: genericReconnectImmediately, 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: otherError, Msg: msg},
}
}
ipe/events.go
+8 -20
View File
@@ -160,29 +160,17 @@ type errorEvent struct {
// Create a new error event // Create a new error event
// Pusher protocol is very strange in some parts // Pusher protocol is very strange in some parts
// It send null in some errors. // 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 {
func newErrorEvent(code int, message string) errorEvent {
var data interface{}
if code == otherError { type dataErrorEvent struct {
data = struct { Code *int `json:"code"`
Code *int `json:"code"` Message string `json:"message"`
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} return errorEvent{Event: "pusher:error", Data: dataErrorEvent{
Code: code,
Message: message,
}}
} }
// { // {
ipe/events_test.go
+29
View File
@@ -0,0 +1,29 @@
package ipe
import (
"bytes"
"encoding/json"
"testing"
)
func Test_newErrorEvent_with_invalid_code(t *testing.T) {
event := newErrorEvent(0, "The error message")
data, _ := json.Marshal(event)
expected := `{"event":"pusher:error","data":{"code":null,"message":"The error message"}}`
if bytes.Compare(data, []byte(expected)) != 0 {
t.Errorf("%s != %s", string(data), expected)
}
}
func Test_newErrorEvent_with_valid_code(t *testing.T) {
event := newErrorEvent(4007, "Unsupported protocol version")
data, _ := json.Marshal(event)
expected := `{"event":"pusher:error","data":{"code":4007,"message":"Unsupported protocol version"}}`
if bytes.Compare(data, []byte(expected)) != 0 {
t.Errorf("%s != %s", string(data), expected)
}
}
ipe/handlers.go
+54 -85
View File
@@ -12,12 +12,8 @@ import (
"sort" "sort"
"strings" "strings"
goji "goji.io"
"goji.io/pat"
log "github.com/golang/glog" log "github.com/golang/glog"
"golang.org/x/net/context" "github.com/pressly/chi"
"github.com/dimiro1/ipe/utils" "github.com/dimiro1/ipe/utils"
) )
@@ -51,79 +47,64 @@ func prepareQueryString(params url.Values) string {
// * The request path (e.g. /some/resource) // * 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. // * 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) // 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(DB db, next goji.Handler) goji.HandlerFunc { func authenticationHandler(DB db) func(http.Handler) http.Handler {
return func(ctx context.Context, w http.ResponseWriter, r *http.Request) { return func(next http.Handler) http.Handler {
appID := pat.Param(ctx, "app_id") fn := func(w http.ResponseWriter, r *http.Request) {
appID := chi.URLParam(r, "app_id")
app, err := DB.GetAppByAppID(appID) app, err := DB.GetAppByAppID(appID)
if err != nil { if err != nil {
log.Error(err) log.Error(err)
http.Error(w, "Not authorized", http.StatusUnauthorized) http.Error(w, "Not authorized", http.StatusUnauthorized)
return return
}
query := r.URL.Query()
signature := query.Get("auth_signature")
query.Del("auth_signature")
queryString := prepareQueryString(query)
toSign := strings.ToUpper(r.Method) + "\n" + r.URL.Path + "\n" + queryString
if utils.HashMAC([]byte(toSign), []byte(app.Secret)) == signature {
next.ServeHTTP(w, r)
} else {
log.Error("Not authorized")
http.Error(w, "Not authorized", http.StatusUnauthorized)
}
} }
query := r.URL.Query() return http.HandlerFunc(fn)
signature := query.Get("auth_signature")
query.Del("auth_signature")
queryString := prepareQueryString(query)
toSign := strings.ToUpper(r.Method) + "\n" + r.URL.Path + "\n" + queryString
if utils.HashMAC([]byte(toSign), []byte(app.Secret)) == signature {
next.ServeHTTPC(ctx, w, r)
} else {
log.Error("Not authorized")
http.Error(w, "Not authorized", http.StatusUnauthorized)
}
} }
} }
// Check if the application is disabled // Check if the application is disabled
func restCheckAppDisabledHandler(DB db, next goji.Handler) goji.HandlerFunc { func checkAppDisabled(DB db) func(http.Handler) http.Handler {
return func(ctx context.Context, w http.ResponseWriter, r *http.Request) { return func(next http.Handler) http.Handler {
appID := pat.Param(ctx, "app_id") fn := func(w http.ResponseWriter, r *http.Request) {
appID := chi.URLParam(r, "app_id")
currentApp, err := DB.GetAppByAppID(appID) currentApp, err := DB.GetAppByAppID(appID)
if err != nil { if err != nil {
http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusForbidden) 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
}
next.ServeHTTPC(ctx, w, r)
}
}
func recoverHandler(next goji.Handler) goji.HandlerFunc {
return func(ctx context.Context, w http.ResponseWriter, r *http.Request) {
defer func() {
if r := recover(); r != nil {
log.Errorf("Please verify the url parameters error was: %s", r)
http.Error(w, "Not authorized", http.StatusUnauthorized)
return return
} }
}()
next.ServeHTTPC(ctx, w, r) if currentApp.ApplicationDisabled {
http.Error(w, "Application disabled", http.StatusForbidden)
return
}
next.ServeHTTP(w, r)
}
return http.HandlerFunc(fn)
} }
} }
// commonHandlers combine restCheckAppDisabledHandler and restAuthenticationHandler handlers
func commonHandlers(DB db, next goji.Handler) goji.HandlerFunc {
return recoverHandler(restCheckAppDisabledHandler(DB, restAuthenticationHandler(DB, next)))
}
func newPostEventsHandler(DB db) goji.HandlerFunc {
return commonHandlers(DB, &postEventsHandler{DB})
}
type postEventsHandler struct{ DB db } type postEventsHandler struct{ DB db }
// ServeHTTPC An event consists of a name and data (typically JSON) which may be sent to all subscribers to a particular channel or channels. // ServeHTTPC An event consists of a name and data (typically JSON) which may be sent to all subscribers to a particular channel or channels.
@@ -141,8 +122,8 @@ type postEventsHandler struct{ DB db }
// Response is an empty JSON hash. // Response is an empty JSON hash.
// //
// POST /apps/{app_id}/events // POST /apps/{app_id}/events
func (h *postEventsHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter, r *http.Request) { func (h *postEventsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
appID := pat.Param(ctx, "app_id") appID := chi.URLParam(r, "app_id")
app, err := h.DB.GetAppByAppID(appID) app, err := h.DB.GetAppByAppID(appID)
@@ -187,10 +168,6 @@ func (h *postEventsHandler) ServeHTTPC(ctx context.Context, w http.ResponseWrite
w.Write([]byte("{}")) w.Write([]byte("{}"))
} }
func newGetChannelsHandler(DB db) goji.HandlerFunc {
return commonHandlers(DB, &getChannelsHandler{DB})
}
type getChannelsHandler struct{ DB db } type getChannelsHandler struct{ DB db }
// Allows fetching a hash of occupied channels (optionally filtered by prefix), // Allows fetching a hash of occupied channels (optionally filtered by prefix),
@@ -212,10 +189,10 @@ type getChannelsHandler struct{ DB db }
// } // }
// //
// GET /apps/{app_id}/channels // GET /apps/{app_id}/channels
func (h *getChannelsHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter, r *http.Request) { func (h *getChannelsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
query := r.URL.Query() query := r.URL.Query()
appID := pat.Param(ctx, "app_id") appID := chi.URLParam(r, "app_id")
filter := query.Get("filter_by_prefix") filter := query.Get("filter_by_prefix")
info := query.Get("info") info := query.Get("info")
@@ -282,10 +259,6 @@ func (h *getChannelsHandler) ServeHTTPC(ctx context.Context, w http.ResponseWrit
} }
} }
func newGetChannelHandler(DB db) goji.HandlerFunc {
return commonHandlers(DB, &getChannelHandler{DB})
}
type getChannelHandler struct{ DB db } type getChannelHandler struct{ DB db }
// Fetch info for one channel // Fetch info for one channel
@@ -298,19 +271,19 @@ type getChannelHandler struct{ DB db }
// } // }
// //
// GET /apps/{app_id}/channels/{channel_name} // GET /apps/{app_id}/channels/{channel_name}
func (h *getChannelHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter, r *http.Request) { func (h *getChannelHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json;charset=UTF-8") w.Header().Set("Content-Type", "application/json;charset=UTF-8")
query := r.URL.Query() query := r.URL.Query()
appID := pat.Param(ctx, "app_id") appID := chi.URLParam(r, "app_id")
app, err := h.DB.GetAppByAppID(appID) app, err := h.DB.GetAppByAppID(appID)
if err != nil { if err != nil {
http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest) http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
} }
channelName := pat.Param(ctx, "channel_name") channelName := chi.URLParam(r, "channel_name")
// Channel name could not be empty // Channel name could not be empty
if strings.TrimSpace(channelName) == "" { if strings.TrimSpace(channelName) == "" {
@@ -376,10 +349,6 @@ func (h *getChannelHandler) ServeHTTPC(ctx context.Context, w http.ResponseWrite
} }
} }
func newGetChannelUsersHandler(DB db) goji.HandlerFunc {
return commonHandlers(DB, &getChannelUsersHandler{DB})
}
type getChannelUsersHandler struct{ DB db } type getChannelUsersHandler struct{ DB db }
// Allowed only for presence-channels // Allowed only for presence-channels
@@ -393,9 +362,9 @@ type getChannelUsersHandler struct{ DB db }
// } // }
// //
// GET /apps/{app_id}/channels/{channel_name}/users // GET /apps/{app_id}/channels/{channel_name}/users
func (h *getChannelUsersHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter, r *http.Request) { func (h *getChannelUsersHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
appID := pat.Param(ctx, "app_id") appID := chi.URLParam(r, "app_id")
channelName := pat.Param(ctx, "channel_name") channelName := chi.URLParam(r, "channel_name")
isPresence := utils.IsPresenceChannel(channelName) isPresence := utils.IsPresenceChannel(channelName)
ipe/handlers_test.go
+26 -21
View File
@@ -1,15 +1,14 @@
package ipe package ipe
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"testing" "testing"
"goji.io/pattern" "github.com/pressly/chi"
"golang.org/x/net/context"
) )
var ( var (
@@ -41,14 +40,15 @@ func init() {
func Test_getChannels_all(t *testing.T) { func Test_getChannels_all(t *testing.T) {
appID := testApp.AppID appID := testApp.AppID
ctx := context.Background() rctx := chi.NewRouteContext()
ctx = context.WithValue(ctx, pattern.Variable("app_id"), appID) rctx.URLParams.Add("app_id", appID)
r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels", appID), nil) r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels", appID), nil)
r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx))
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler := &getChannelsHandler{database} handler := &getChannelsHandler{database}
handler.ServeHTTPC(ctx, w, r) handler.ServeHTTP(w, r)
if w.Code != http.StatusOK { if w.Code != http.StatusOK {
t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK) t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
@@ -67,14 +67,15 @@ func Test_getChannels_all(t *testing.T) {
func Test_getChannels_filter_by_presence_prefix(t *testing.T) { func Test_getChannels_filter_by_presence_prefix(t *testing.T) {
appID := testApp.AppID appID := testApp.AppID
ctx := context.Background() rctx := chi.NewRouteContext()
ctx = context.WithValue(ctx, pattern.Variable("app_id"), appID) rctx.URLParams.Add("app_id", appID)
r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=presence-", appID), nil) r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=presence-", appID), nil)
r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx))
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler := &getChannelsHandler{database} handler := &getChannelsHandler{database}
handler.ServeHTTPC(ctx, w, r) handler.ServeHTTP(w, r)
if w.Code != http.StatusOK { if w.Code != http.StatusOK {
t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK) t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
@@ -94,14 +95,15 @@ func Test_getChannels_filter_by_presence_prefix(t *testing.T) {
func Test_getChannels_filter_by_presence_prefix_and_user_count(t *testing.T) { func Test_getChannels_filter_by_presence_prefix_and_user_count(t *testing.T) {
appID := testApp.AppID appID := testApp.AppID
ctx := context.Background() rctx := chi.NewRouteContext()
ctx = context.WithValue(ctx, pattern.Variable("app_id"), appID) rctx.URLParams.Add("app_id", appID)
r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=presence-&info=user_count", appID), nil) r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=presence-&info=user_count", appID), nil)
r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx))
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler := &getChannelsHandler{database} handler := &getChannelsHandler{database}
handler.ServeHTTPC(ctx, w, r) handler.ServeHTTP(w, r)
if w.Code != http.StatusOK { if w.Code != http.StatusOK {
t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK) t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
@@ -133,14 +135,15 @@ func Test_getChannels_filter_by_presence_prefix_and_user_count(t *testing.T) {
func Test_getChannels_filter_by_private_prefix_and_info_user_count(t *testing.T) { func Test_getChannels_filter_by_private_prefix_and_info_user_count(t *testing.T) {
appID := testApp.AppID appID := testApp.AppID
ctx := context.Background() rctx := chi.NewRouteContext()
ctx = context.WithValue(ctx, pattern.Variable("app_id"), appID) rctx.URLParams.Add("app_id", appID)
r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=private-&info=user_count", appID), nil) r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=private-&info=user_count", appID), nil)
r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx))
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler := &getChannelsHandler{database} handler := &getChannelsHandler{database}
handler.ServeHTTPC(ctx, w, r) handler.ServeHTTP(w, r)
if w.Code != http.StatusBadRequest { if w.Code != http.StatusBadRequest {
t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusBadRequest) t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusBadRequest)
@@ -150,14 +153,15 @@ func Test_getChannels_filter_by_private_prefix_and_info_user_count(t *testing.T)
func Test_getChannels_filter_by_public_prefix(t *testing.T) { func Test_getChannels_filter_by_public_prefix(t *testing.T) {
appID := testApp.AppID appID := testApp.AppID
ctx := context.Background() rctx := chi.NewRouteContext()
ctx = context.WithValue(ctx, pattern.Variable("app_id"), appID) rctx.URLParams.Add("app_id", appID)
r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=public-", appID), nil) r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=public-", appID), nil)
r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx))
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler := &getChannelsHandler{database} handler := &getChannelsHandler{database}
handler.ServeHTTPC(ctx, w, r) handler.ServeHTTP(w, r)
if w.Code != http.StatusOK { if w.Code != http.StatusOK {
t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK) t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
@@ -183,14 +187,15 @@ func Test_getChannels_filter_by_public_prefix(t *testing.T) {
func Test_getChannels_filter_by_private_prefix(t *testing.T) { func Test_getChannels_filter_by_private_prefix(t *testing.T) {
appID := testApp.AppID appID := testApp.AppID
ctx := context.Background() rctx := chi.NewRouteContext()
ctx = context.WithValue(ctx, pattern.Variable("app_id"), appID) rctx.URLParams.Add("app_id", appID)
r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=private-", appID), nil) r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=private-", appID), nil)
r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx))
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler := &getChannelsHandler{database} handler := &getChannelsHandler{database}
handler.ServeHTTPC(ctx, w, r) handler.ServeHTTP(w, r)
if w.Code != http.StatusOK { if w.Code != http.StatusOK {
t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK) t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
ipe/ipe.go
+25 -14
View File
@@ -11,11 +11,9 @@ import (
"os" "os"
"time" "time"
"goji.io/pat"
goji "goji.io"
log "github.com/golang/glog" log "github.com/golang/glog"
"github.com/pressly/chi"
"github.com/pressly/chi/middleware"
) )
// Start Parse the configuration file and starts the ipe server // Start Parse the configuration file and starts the ipe server
@@ -27,14 +25,16 @@ func Start(filename string) {
file, err := os.Open(filename) file, err := os.Open(filename)
if err != nil { if err != nil {
log.Fatal(err) log.Error(err)
return
} }
defer file.Close() defer file.Close()
// Reading config // Reading config
if err := json.NewDecoder(file).Decode(&conf); err != nil { if err := json.NewDecoder(file).Decode(&conf); err != nil {
log.Fatal(err) log.Error(err)
return
} }
// Using a in memory database // Using a in memory database
@@ -45,20 +45,31 @@ func Start(filename string) {
db.AddApp(newAppFromConfig(a)) db.AddApp(newAppFromConfig(a))
} }
router := goji.NewMux() r := chi.NewRouter()
router.HandleFuncC(pat.Post("/apps/:app_id/events"), newPostEventsHandler(db)) r.Use(middleware.Recoverer)
router.HandleFuncC(pat.Get("/apps/:app_id/channels"), newGetChannelsHandler(db))
router.HandleFuncC(pat.Get("/apps/:app_id/channels/:channel_name"), newGetChannelHandler(db)) r.Get("/app/:key", (&websocketHandler{db}).ServeHTTP)
router.HandleFuncC(pat.Get("/apps/:app_id/channels/:channel_name/users"), newGetChannelUsersHandler(db)) r.Group(func(r chi.Router) {
router.HandleC(pat.Get("/app/:key"), newWebsocketHandler(db)) r.Use(checkAppDisabled(db))
r.Use(authenticationHandler(db))
r.Post("/apps/:app_id/events", (&postEventsHandler{db}).ServeHTTP)
r.Get("/apps/:app_id/channels", (&getChannelsHandler{db}).ServeHTTP)
r.Get("/apps/:app_id/channels/:channel_name", (&getChannelHandler{db}).ServeHTTP)
r.Get("/apps/:app_id/channels/:channel_name/users", (&getChannelUsersHandler{db}).ServeHTTP)
})
if conf.Profiling {
r.Mount("/debug", middleware.Profiler())
}
if conf.SSL { if conf.SSL {
go func() { go func() {
log.Infof("Starting HTTPS service on %s ...", conf.SSLHost) log.Infof("Starting HTTPS service on %s ...", conf.SSLHost)
log.Fatal(http.ListenAndServeTLS(conf.SSLHost, conf.SSLCertFile, conf.SSLKeyFile, router)) log.Fatal(http.ListenAndServeTLS(conf.SSLHost, conf.SSLCertFile, conf.SSLKeyFile, r))
}() }()
} }
log.Infof("Starting HTTP service on %s ...", conf.Host) log.Infof("Starting HTTP service on %s ...", conf.Host)
log.Fatal(http.ListenAndServe(conf.Host, router)) log.Fatal(http.ListenAndServe(conf.Host, r))
} }
ipe/webhooks.go
+41 -9
View File
@@ -10,10 +10,15 @@ import (
"net/http" "net/http"
"time" "time"
"context"
"fmt"
"github.com/dimiro1/ipe/utils" "github.com/dimiro1/ipe/utils"
log "github.com/golang/glog" log "github.com/golang/glog"
) )
const maxTimeout = 3 * time.Second
// A WebHook is sent as a HTTP POST request to the url which you specify. // 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: // The POST request payload (body) contains a JSON document, and follows the following format:
// { // {
@@ -71,14 +76,19 @@ func newClientHook(channel *channel, s *subscription, event string, data interfa
// { "name": "channel_occupied", "channel": "test_channel" } // { "name": "channel_occupied", "channel": "test_channel" }
func (a *app) TriggerChannelOccupiedHook(c *channel) { func (a *app) TriggerChannelOccupiedHook(c *channel) {
event := newChannelOcuppiedHook(c) event := newChannelOcuppiedHook(c)
triggerHook(event.Name, a, c, event) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
triggerHook(ctx, a, event)
} }
// channel_vacated // channel_vacated
// { "name": "channel_vacated", "channel": "test_channel" } // { "name": "channel_vacated", "channel": "test_channel" }
func (a *app) TriggerChannelVacatedHook(c *channel) { func (a *app) TriggerChannelVacatedHook(c *channel) {
event := newChannelVacatedHook(c) event := newChannelVacatedHook(c)
triggerHook(event.Name, a, c, event) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
triggerHook(ctx, a, event)
} }
// { // {
@@ -96,7 +106,9 @@ func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent st
event.UserID = s.ID event.UserID = s.ID
} }
triggerHook(event.Name, a, c, event) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
triggerHook(ctx, a, event)
} }
// { // {
@@ -106,7 +118,9 @@ func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent st
// } // }
func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) { func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) {
event := newMemberAddedHook(c, s) event := newMemberAddedHook(c, s)
triggerHook(event.Name, a, c, event) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
triggerHook(ctx, a, event)
} }
// { // {
@@ -116,17 +130,22 @@ func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) {
// } // }
func (a *app) TriggerMemberRemovedHook(c *channel, s *subscription) { func (a *app) TriggerMemberRemovedHook(c *channel, s *subscription) {
event := newMemberRemovedHook(c, s) event := newMemberRemovedHook(c, s)
triggerHook(event.Name, a, c, event) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
defer cancel()
triggerHook(ctx, a, event)
} }
func triggerHook(name string, a *app, c *channel, event hookEvent) { func triggerHook(ctx context.Context, a *app, event hookEvent) error {
if !a.WebHooks { if !a.WebHooks {
log.Infof("Webhooks are not enabled for app: %s", a.Name) log.Infof("Webhooks are not enabled for app: %s", a.Name)
return return fmt.Errorf("Webhooks are not enabled for app: %s", a.Name)
} }
var done chan bool
defer close(done)
go func() { go func() {
log.Infof("Triggering %s event", name) log.Infof("Triggering %s event", event.Name)
hook := webHook{TimeMs: time.Now().Unix()} hook := webHook{TimeMs: time.Now().Unix()}
@@ -145,11 +164,14 @@ func triggerHook(name string, a *app, c *channel, event hookEvent) {
var req *http.Request var req *http.Request
req, err = http.NewRequest("POST", a.URLWebHook, bytes.NewReader(js)) req, err = http.NewRequest("POST", a.URLWebHook, bytes.NewReader(js))
if err != nil { if err != nil {
log.Errorf("Error creating request: %+v", err) log.Errorf("Error creating request: %+v", err)
return return
} }
req.WithContext(ctx)
req.Header.Set("User-Agent", "Ipe UA; (+https://github.com/dimiro1/ipe)") req.Header.Set("User-Agent", "Ipe UA; (+https://github.com/dimiro1/ipe)")
req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Pusher-Key", a.Key) req.Header.Set("X-Pusher-Key", a.Key)
@@ -166,7 +188,17 @@ func triggerHook(name string, a *app, c *channel, event hookEvent) {
} }
if err != nil { if err != nil {
log.Errorf("Error posting %s event: %+v", name, err) log.Errorf("Error posting %s event: %+v", event.Name, err)
} }
// Successfully terminated
done <- true
}() }()
select {
case <-ctx.Done():
return ctx.Err()
case <-done:
return nil
}
} }
ipe/websocket.go
+177 -159
View File
@@ -12,13 +12,9 @@ import (
"strconv" "strconv"
"strings" "strings"
goji "goji.io"
"goji.io/pat"
log "github.com/golang/glog" log "github.com/golang/glog"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
"golang.org/x/net/context" "github.com/pressly/chi"
"github.com/dimiro1/ipe/utils" "github.com/dimiro1/ipe/utils"
) )
@@ -26,11 +22,58 @@ import (
var upgrader = websocket.Upgrader{ var upgrader = websocket.Upgrader{
ReadBufferSize: 1024, ReadBufferSize: 1024,
WriteBufferSize: 1024, WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool { return true }, CheckOrigin: func(_ *http.Request) bool {
return true
},
} }
// Handle open Subscriber. func handleMessages(conn *websocket.Conn, sessionID string, app *app) {
func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessionID string, app *app) websocketError { var event struct {
Event string `json:"event"`
}
for {
_, message, err := conn.ReadMessage()
if err != nil {
handleError(conn, sessionID, app, err)
return
}
if err := json.Unmarshal(message, &event); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
return
}
log.Infof("websockets: Handling %s event", event.Event)
switch event.Event {
case "pusher:ping":
onPing(conn)
case "pusher:subscribe":
onSubscribe(conn, sessionID, app, message)
case "pusher:unsubscribe":
onUnsubscribe(conn, sessionID, app, message)
default:
if utils.IsClientEvent(event.Event) {
onClientEvent(conn, sessionID, app, message)
}
}
}
}
func handleError(conn *websocket.Conn, sessionID string, app *app, err error) {
log.Errorf("%+v", err)
if err == io.EOF {
onClose(sessionID, app)
} else if _, ok := err.(*websocket.CloseError); ok {
onClose(sessionID, app)
} else {
emitWSError(newGenericReconnectImmediatelyError(), conn)
}
}
func onOpen(conn *websocket.Conn, r *http.Request, sessionID string, app *app) error {
params := r.URL.Query() params := r.URL.Query()
p := params.Get("protocol") p := params.Get("protocol")
@@ -44,7 +87,7 @@ func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessio
case strings.TrimSpace(p) == "": case strings.TrimSpace(p) == "":
return newNoProtocolVersionSuppliedError() return newNoProtocolVersionSuppliedError()
case protocol != supportedProtocolVersion: case protocol != supportedProtocolVersion:
return newUnsupportedProtocolVersionError() return unsupportedProtocolVersionError
case app.ApplicationDisabled: case app.ApplicationDisabled:
return newApplicationDisabledError() return newApplicationDisabledError()
case app.OnlySSL: case app.OnlySSL:
@@ -65,160 +108,145 @@ func onOpen(conn *websocket.Conn, w http.ResponseWriter, r *http.Request, sessio
return nil return nil
} }
// Handle the close event
func onClose(sessionID string, app *app) { func onClose(sessionID string, app *app) {
app.Disconnect(sessionID) app.Disconnect(sessionID)
} }
// Handle messages func onPing(conn *websocket.Conn) {
// if err := conn.WriteJSON(newPongEvent()); err != nil {
// If there is an unrecoverable error then break the loop, emitWSError(newGenericReconnectImmediatelyError(), conn)
// 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"` func onClientEvent(conn *websocket.Conn, sessionID string, app *app, message []byte) {
if !app.UserEvents {
emitWSError(newGenericError("To send client events, you must enable this feature in the Settings."), conn)
} }
for { clientEvent := rawEvent{}
_, message, err := conn.ReadMessage()
if err != nil { if err := json.Unmarshal(message, &clientEvent); err != nil {
log.Errorf("%+v", err) log.Error(err)
if err == io.EOF { emitWSError(newGenericReconnectImmediatelyError(), conn)
onClose(sessionID, app) return
} else if _, ok := err.(*websocket.CloseError); ok { }
onClose(sessionID, app)
} else {
emitWSError(newGenericReconnectImmediatelyError(), conn)
}
break
}
if err := json.Unmarshal(message, &event); err != nil { channel, err := app.FindChannelByChannelID(clientEvent.Channel)
emitWSError(newGenericReconnectImmediatelyError(), conn)
break
}
log.Infof("websockets: Handling %s event", event.Event) if err != nil {
emitWSError(newGenericError(fmt.Sprintf("Could not find a channel with the id %s", clientEvent.Channel)), conn)
}
switch event.Event { if !channel.IsPresenceOrPrivate() {
case "pusher:ping": emitWSError(newGenericError("Client event rejected - only supported on private and presence channels"), conn)
if err := conn.WriteJSON(newPongEvent()); err != nil { return
emitWSError(newGenericReconnectImmediatelyError(), conn) }
}
case "pusher:subscribe":
subscribeEvent := subscribeEvent{}
if err := json.Unmarshal(message, &subscribeEvent); err != nil { if err := app.Publish(channel, clientEvent, sessionID); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn) log.Error(err)
break emitWSError(newGenericReconnectImmediatelyError(), conn)
} return
}
connection, err := app.FindConnection(sessionID)
if err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
break
}
channelName := strings.TrimSpace(subscribeEvent.Data.Channel)
if !utils.IsChannelNameValid(channelName) {
emitWSError(newGenericError(fmt.Sprintf("This channel name is not valid")), conn)
break
}
isPresence := utils.IsPresenceChannel(channelName)
isPrivate := utils.IsPrivateChannel(channelName)
if isPresence || isPrivate {
toSign := []string{connection.SocketID, channelName}
if isPresence || len(subscribeEvent.Data.ChannelData) > 0 {
toSign = append(toSign, subscribeEvent.Data.ChannelData)
}
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)
log.Info(subscribeEvent.Data.ChannelData)
if err := app.Subscribe(channel, connection, subscribeEvent.Data.ChannelData); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
}
case "pusher:unsubscribe":
unsubscribeEvent := unsubscribeEvent{}
if err := json.Unmarshal(message, &unsubscribeEvent); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
}
connection, err := app.FindConnection(sessionID)
if err != nil {
emitWSError(newGenericError(fmt.Sprintf("Could not find a connection with the id %s", sessionID)), conn)
}
channel, err := app.FindChannelByChannelID(unsubscribeEvent.Data.Channel)
if err != nil {
emitWSError(newGenericError(fmt.Sprintf("Could not find a channel with the id %s", unsubscribeEvent.Data.Channel)), conn)
}
if err := app.Unsubscribe(channel, connection); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
break
}
default: // CLient Events ??
// see http://pusher.com/docs/client_api_guide/client_events#trigger-events
if utils.IsClientEvent(event.Event) {
if !app.UserEvents {
emitWSError(newGenericError("To send client events, you must enable this feature in the Settings."), conn)
}
clientEvent := rawEvent{}
if err := json.Unmarshal(message, &clientEvent); err != nil {
log.Error(err)
emitWSError(newGenericReconnectImmediatelyError(), conn)
break
}
channel, err := app.FindChannelByChannelID(clientEvent.Channel)
if err != nil {
emitWSError(newGenericError(fmt.Sprintf("Could not find a channel with the id %s", clientEvent.Channel)), conn)
}
if !channel.IsPresenceOrPrivate() {
emitWSError(newGenericError("Client event rejected - only supported on private and presence channels"), conn)
break
}
if err := app.Publish(channel, clientEvent, sessionID); err != nil {
log.Error(err)
emitWSError(newGenericReconnectImmediatelyError(), conn)
break
}
}
} // switch
} // For
} }
func newWebsocketHandler(DB db) goji.Handler { func onUnsubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) {
return &websocketHandler{DB} unsubscribeEvent := unsubscribeEvent{}
if err := json.Unmarshal(message, &unsubscribeEvent); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
}
connection, err := app.FindConnection(sessionID)
if err != nil {
emitWSError(newGenericError(fmt.Sprintf("Could not find a connection with the id %s", sessionID)), conn)
}
channel, err := app.FindChannelByChannelID(unsubscribeEvent.Data.Channel)
if err != nil {
emitWSError(newGenericError(fmt.Sprintf("Could not find a channel with the id %s", unsubscribeEvent.Data.Channel)), conn)
}
if err := app.Unsubscribe(channel, connection); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
return
}
} }
type websocketHandler struct{ DB db } func onSubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) {
subscribeEvent := subscribeEvent{}
if err := json.Unmarshal(message, &subscribeEvent); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
return
}
connection, err := app.FindConnection(sessionID)
if err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
return
}
channelName := strings.TrimSpace(subscribeEvent.Data.Channel)
if !utils.IsChannelNameValid(channelName) {
emitWSError(newGenericError("This channel name is not valid"), conn)
return
}
isPresence := utils.IsPresenceChannel(channelName)
isPrivate := utils.IsPrivateChannel(channelName)
if isPresence || isPrivate {
toSign := []string{connection.SocketID, channelName}
if isPresence || len(subscribeEvent.Data.ChannelData) > 0 {
toSign = append(toSign, subscribeEvent.Data.ChannelData)
}
if !validateAuthKey(subscribeEvent.Data.Auth, toSign, app) {
emitWSError(newGenericError(fmt.Sprintf("Auth value for subscription to %s is invalid", channelName)), conn)
return
}
}
channel := app.FindOrCreateChannelByChannelID(channelName)
log.Info(subscribeEvent.Data.ChannelData)
if err := app.Subscribe(channel, connection, subscribeEvent.Data.ChannelData); err != nil {
emitWSError(newGenericReconnectImmediatelyError(), conn)
}
}
func validateAuthKey(givenAuthKey string, toSign []string, app *app) bool {
expectedAuthKey := fmt.Sprintf("%s:%s", app.Key, utils.HashMAC([]byte(strings.Join(toSign, ":")), []byte(app.Secret)))
return givenAuthKey == expectedAuthKey
}
// Emit an Websocket ErrorEvent
func emitWSError(err error, conn *websocket.Conn) {
e, ok := err.(websocketError)
if !ok {
log.Error(err)
return
}
event := newErrorEvent(e.GetCode(), e.GetMsg())
if err := conn.WriteJSON(event); err != nil {
log.Error(err)
}
}
type websocketHandler struct {
DB db
}
// Websocket GET /app/{key} // Websocket GET /app/{key}
func (h *websocketHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter, r *http.Request) { func (h *websocketHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
conn, err := upgrader.Upgrade(w, r, nil) conn, err := upgrader.Upgrade(w, r, nil)
defer func() { defer func() {
if conn != nil { if conn != nil {
@@ -231,7 +259,7 @@ func (h *websocketHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter
return return
} }
appKey := pat.Param(ctx, "key") appKey := chi.URLParam(r, "key")
app, err := h.DB.GetAppByKey(appKey) app, err := h.DB.GetAppByKey(appKey)
@@ -243,20 +271,10 @@ func (h *websocketHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter
sessionID := utils.GenerateSessionID() sessionID := utils.GenerateSessionID()
if err := onOpen(conn, w, r, sessionID, app); err != nil { if err := onOpen(conn, r, sessionID, app); err != nil {
emitWSError(err, conn) emitWSError(err, conn)
return return
} }
onMessage(conn, w, r, sessionID, app) handleMessages(conn, sessionID, app)
}
// 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.Error(err)
}
} }