making it work and build

Added instructions on how to build ipe

.gitignore

@@ -38,7 +38,7 @@ _testmain.go
 
 # project files should be checked into the repository, unless a significant
 # proportion of contributors will probably not be using SublimeText
-# *.sublime-project
+*.sublime-project
 
 # sftp configuration file
 sftp-config.json
@@ -155,7 +155,10 @@ flymake*
 
 ignore_http/*
 config.json
+config.yml
 *.pem
 build
 .vscode/*
 debug
+
+vendor/*

Dockerfile

@@ -0,0 +1,20 @@
+FROM golang:alpine as builder
+RUN mkdir /build 
+ADD . /build/
+WORKDIR /build 
+RUN apk add git gcc musl-dev
+RUN go build -o ipe ./cmd
+FROM alpine
+USER root
+RUN  mkdir -p /config
+RUN adduser -S -D -H -h /app appuser
+COPY ./entrypoint.sh /app/entrypoint.sh
+RUN chmod +x /app/entrypoint.sh
+USER appuser
+WORKDIR /app
+COPY --from=builder /build/ipe /app/
+COPY --from=builder /build/config-example.yml /app/config-example.yml
+VOLUME /config
+CMD ["/bin/sh", "/app/entrypoint.sh"]
+EXPOSE 4343
+EXPOSE 8080

README.md

@@ -1,14 +1,17 @@
-[![Quality code report](http://goreportcard.com/badge/dimiro1/ipe)](http://goreportcard.com/report/dimiro1/ipe)
+[![Go Report Card](https://goreportcard.com/badge/github.com/dimiro1/ipe)](https://goreportcard.com/report/github.com/dimiro1/ipe)
+
+Try browsing [the code on Sourcegraph](https://sourcegraph.com/github.com/dimiro1/ipe)!
 
 # IPÊ
 
-This software is written in Go - the WYSIWYG lang
+An open source Pusher server implementation compatible with Pusher client libraries written in Go.
 
 # Why I wrote this software?
 
 1. I wanted to learn Go and I needed a non trivial application;
 2. I use Pusher in some projects;
 3. I really like Pusher;
+4. I was using Pusher on some projects behind a firewall;
 
 # Features
 
@@ -18,7 +21,7 @@ This software is written in Go - the WYSIWYG lang
 * Web Hooks;
 * Client events;
 * Complete REST API;
-* Easy instalation;
+* Easy installation;
 * A single binary without dependencies;
 * Easy configuration;
 * Protocol version 7;
@@ -29,8 +32,6 @@ This software is written in Go - the WYSIWYG lang
 
 You can download pre built binaries from the [releases tab](https://github.com/dimiro1/ipe/releases).
 
-I do not have a Windows machine, so I can only distribute binaries for amd64 linux and amd64 darwin.
-
 # Building
 
 ```console
@@ -43,31 +44,39 @@ or simply
 $ go install github.com/dimiro1/ipe
 ```
 
+Building from a local checkout
+
+```console
+$ git clone https://github.com/dimiro1/ipe.git
+$ cd ipe/cmd
+$ go build -o ipe
+```
+
 # How to configure?
 
 ## The server
 
-```javascript
+```yaml
-{
+
-	"Host": ":8080",                    // Required
+---
-	"SSL": false,                       // Required but can be false
+host: ":8080"
-	"SSLHost": ":4433",                 // Required if SSL is true
+profiling: false
-	"SSLKeyFile": "A key.pem file",     // Required if SSL is true
+ssl:
-	"SSLCertFile": "A cert.pem file",   // Required if SSL is true
+  enabled: false
-	"Apps": [                           // Required, A Json arrays with multiple apps
+  host: ":4343"
-		{
+  key_file: "key.pem"
-			"ApplicationDisabled": false,               // Required but can be false
+  cert_file: "cert.pem"
-			"Secret": "A really secret random string",  // Required
+apps:
-			"Key": "A random Key string",               // Required
+  - name: "Sample Application"
-			"OnlySSL": false,                           // Required but can be false
+    enabled: true
-			"Name": "The app name",                     // Required
+    only_ssl: false
-			"AppID": "The app ID",                      // Required
+    key: "278d525bdf162c739803"
-			"UserEvents": true,                         // Required but can be false
+    secret: "${APP_SECRET}" # Expand env vars
-			"WebHooks": true,                           // Required but can be false
+    app_id: "1"
-			"URLWebHook": "Some URL to send webhooks"   // Required if WebHooks is true
+    user_events: true
-		}
+    webhooks:
-	]
+      enabled: true
-}
+      url: "http://127.0.0.1:5000/hook"
 
 ```
 
@@ -76,7 +85,7 @@ $ go install github.com/dimiro1/ipe
 ### Client javascript library
 
 ```javascript
-var pusher = new Pusher(APP_KEY, {
+let pusher = new Pusher(APP_KEY, {
   wsHost: 'localhost',
   wsPort: 8080,
   wssPort: 4433,    // Required if encrypted is true
@@ -101,10 +110,29 @@ PHP
 $pusher = new Pusher(APP_KEY, APP_SECRET, APP_ID, DEBUG, "http://localhost", "8080");
 ```
 
+Go
+
+```go
+package main
+
+import "github.com/pusher/pusher-http-go"
+
+func main() {
+	client := pusher.Client{
+        AppId:  "APP_ID",
+        Key:    "APP_KEY",
+        Secret: "APP_SECRET",
+        Host:   ":8080",
+    }
+	
+	// use the client
+}
+```
+
 NodeJS
 
 ```javascript
-var pusher = new Pusher({
+let pusher = new Pusher({
   appId: APP_ID,
   key: APP_KEY,
   secret: APP_SECRET
@@ -137,7 +165,7 @@ Feel free to fork this repo.
 
 # Pusher
 
-Pusher is an excelent service, their service is very reliable. I recomend for everyone.
+Pusher is an excellent service, their service is very reliable. I recommend for everyone.
 
 # Where this name came from?
 
@@ -151,7 +179,7 @@ Claudemiro Alves Feitosa Neto
 
 # LICENSE
 
-Copyright 2014, 2015, 2016 Claudemiro Alves Feitosa Neto. All rights reserved.
+Copyright 2014, 2018 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.
 

Rakefile

@@ -1,58 +0,0 @@
-# Copyright 2016 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.
-
-require 'rake/clean'
-
-VERSION = 'v1.1.0'
-GITHASH = `git rev-parse --short HEAD`
-DATE = Time.now.strftime '%Y%m%d%H%M%S'
-
-CLOBBER.include 'build'
-
-task :default => [:'run-debug']
-
-desc 'Build a debug version'
-task :debug do
-	sh "GO15VENDOREXPERIMENT=1 go install -ldflags '-w -X main.version=DEBUG -X main.buildstamp=DEBUG -X main.githash=DEBUG' github.com/dimiro1/ipe"
-end
-
-desc 'Build and run debug version'
-task :'run-debug' => :debug do
-	sh '$GOPATH/bin/ipe --config $GOPATH/src/github.com/dimiro1/ipe/config.json -logtostderr=true -v=2'
-end
-
-desc 'Run test suite'
-task :test do
-	sh 'GO15VENDOREXPERIMENT=1 go test . `glide nv`'
-end
-
-desc 'Download the development dependencies'
-task :'dev-deps' do
-	sh 'go get github.com/pusher/pusher-http-go'
-end
-
-desc 'Generate distributions'
-task :distribute => [:linux, :darwin]
-
-desc 'Generate a linux distribution'
-task :linux do
-	Rake::Task['build'].invoke 'linux'
-end
-
-desc 'Generate a darwin distribution'
-task :darwin do
-	Rake::Task['build'].invoke 'darwin'
-end
-
-task :build, [:os] do |t, args|
-	t.reenable
-	os = args[:os]
-
-	sh "mkdir -p build/#{os}"
-	sh "GO15VENDOREXPERIMENT=1 GOOS=#{os} GOARCH=amd64 go build -ldflags '-X main.version=#{VERSION} -X main.buildstamp=#{DATE} -X main.githash=#{GITHASH}' -o build/#{os}/ipe github.com/dimiro1/ipe"
-	sh "cp ipe/config-example.json build/#{os}/config.json"
-	sh "cp LICENSE build/#{os}/"
-	sh "cp README.md build/#{os}/"
-	sh "tar -C build/#{os} -czf build/ipe_#{VERSION}_#{os}_amd64.tar.gz ."
-end

TODO.org

@@ -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

api/handlers.go

@@ -2,7 +2,7 @@
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
 
-package ipe
+package api
 
 import (
 	"encoding/json"
@@ -13,11 +13,18 @@ import (
 	"strings"
 
 	log "github.com/golang/glog"
+	"github.com/gorilla/mux"
 
-	"github.com/dimiro1/ipe/utils"
+	"ipe/events"
+	"ipe/storage"
+	"ipe/utils"
 )
 
-// Prepare Querystring
+// // Maximum event size permitted 10 kB
+// See: http://blogs.gnome.org/cneumair/2008/09/30/1-kb-1024-bytes-no-1-kb-1000-bytes/
+const maxDataEventSize = 10 * 1000
+
+// Prepare QueryString
 func prepareQueryString(params url.Values) string {
 	var keys []string
 
@@ -36,7 +43,7 @@ func prepareQueryString(params url.Values) string {
 	return strings.Join(pieces, "&")
 }
 
-// Authenticate pusher
+// Authentication Authenticate pusher
 // see: https://gist.github.com/mloughran/376898
 //
 // The signature is a HMAC SHA256 hex digest.
@@ -46,63 +53,79 @@ func prepareQueryString(params url.Values) string {
 //  * 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(ctx *applicationContext, h contextHandler) contextHandler {
+func Authentication(storage storage.Storage) func(http.Handler) http.Handler {
-	return contextHandlerFunc(func(ctx *applicationContext, p params, w http.ResponseWriter, r *http.Request) {
+	return func(next http.Handler) http.Handler {
-		appID := p.Get("app_id")
+		fn := func(w http.ResponseWriter, r *http.Request) {
+			var (
+				pathVars = mux.Vars(r)
+				appID    = pathVars["app_id"]
+			)
 
-		app, err := ctx.DB.GetAppByAppID(appID)
+			app, err := storage.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 {
-			h.ServeWithContext(ctx, p, w, r)
-		} else {
-			log.Error("Not authorized")
-			http.Error(w, "Not authorized", http.StatusUnauthorized)
-		}
-	})
 }
 
-// Check if the application is disabled
+// CheckAppDisabled Check if the application is disabled
-func restCheckAppDisabledHandler(ctx *applicationContext, h contextHandler) contextHandler {
+func CheckAppDisabled(storage storage.Storage) func(http.Handler) http.Handler {
-	return contextHandlerFunc(func(ctx *applicationContext, p params, w http.ResponseWriter, r *http.Request) {
+	return func(next http.Handler) http.Handler {
-		appID := p.Get("app_id")
+		fn := func(w http.ResponseWriter, r *http.Request) {
+			var (
+				pathVars = mux.Vars(r)
+				appID    = pathVars["app_id"]
+			)
 
-		currentApp, err := ctx.DB.GetAppByAppID(appID)
+			currentApp, err := storage.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
+				return
+			}
+
+			if !currentApp.Enabled {
+				http.Error(w, "Application disabled", http.StatusForbidden)
+				return
+			}
+
+			next.ServeHTTP(w, r)
 		}
-
+		return http.HandlerFunc(fn)
-		if currentApp.ApplicationDisabled {
+	}
-			http.Error(w, "Application disabled", http.StatusForbidden)
-			return
-		}
-
-		h.ServeWithContext(ctx, p, w, r)
-	})
 }
 
-// commonHandlers combine restCheckAppDisabledHandler and restAuthenticationHandler handlers
+// PostEvents handle post events
-func commonHandlers(ctx *applicationContext, h contextHandlerFunc) contextHandler {
+type PostEvents struct{ storage storage.Storage }
-	return restCheckAppDisabledHandler(ctx, restAuthenticationHandler(ctx, h))
+
+// NewPostEvents return a new PostEvents handler
+func NewPostEvents(storage storage.Storage) *PostEvents {
+	return &PostEvents{storage: storage}
 }
 
-// An event consists of a name and data (typically JSON) which may be sent to all subscribers to a particular channel or channels.
+// ServeHTTP 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.
@@ -117,10 +140,13 @@ func commonHandlers(ctx *applicationContext, h contextHandlerFunc) contextHandle
 // Response is an empty JSON hash.
 //
 // POST /apps/{app_id}/events
-func postEvents(ctx *applicationContext, p params, w http.ResponseWriter, r *http.Request) {
+func (h *PostEvents) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	appID := p.Get("app_id")
+	var (
+		pathVars = mux.Vars(r)
+		appID    = pathVars["app_id"]
+	)
 
-	app, err := ctx.DB.GetAppByAppID(appID)
+	app, err := h.storage.GetAppByAppID(appID)
 
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
@@ -155,15 +181,28 @@ func postEvents(ctx *applicationContext, p params, w http.ResponseWriter, r *htt
 	for _, c := range input.Channels {
 		channel := app.FindOrCreateChannelByChannelID(c)
 
-		app.Publish(channel, rawEvent{Event: input.Name, Channel: c, Data: input.Data}, input.SocketID)
+		if err := app.Publish(channel, events.Raw{Event: input.Name, Channel: c, Data: input.Data}, input.SocketID); err != nil {
+			log.Errorf("error publishing event %+v", err)
+			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+		}
 	}
 
-	w.Header().Set("Content-Type", "application/json;charset=UTF-8")
+	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
-	w.Write([]byte("{}"))
+	if _, err := w.Write([]byte("{}")); err != nil {
+		log.Errorf("unexpected error while writing into response %+v", err)
+	}
 }
 
-// Allows fetching a hash of occupied channels (optionally filtered by prefix),
+// GetChannels handle get channels
+type GetChannels struct{ storage storage.Storage }
+
+// NewGetChannels return a new GetChannels handler
+func NewGetChannels(storage storage.Storage) *GetChannels {
+	return &GetChannels{storage: storage}
+}
+
+// ServeHTTP Allows fetching a hash of occupied channels (optionally filtered by prefix),
 // and optionally one or more attributes for each channel.
 //
 // Notes:
@@ -182,14 +221,15 @@ func postEvents(ctx *applicationContext, p params, w http.ResponseWriter, r *htt
 // }
 //
 // GET /apps/{app_id}/channels
-func getChannels(ctx *applicationContext, p params, w http.ResponseWriter, r *http.Request) {
+func (h *GetChannels) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	query := r.URL.Query()
+	var (
-
+		pathVars   = mux.Vars(r)
-	appID := p.Get("app_id")
+		queryVars  = r.URL.Query()
-	filter := query.Get("filter_by_prefix")
+		appID      = pathVars["app_id"]
-	info := query.Get("info")
+		filter     = queryVars.Get("filter_by_prefix")
-
+		info       = queryVars.Get("info")
-	attributes := strings.Split(info, ",")
+		attributes = strings.Split(info, ",")
+	)
 
 	requestedUserCount := false
 
@@ -206,7 +246,7 @@ func getChannels(ctx *applicationContext, p params, w http.ResponseWriter, r *ht
 		return
 	}
 
-	app, err := ctx.DB.GetAppByAppID(appID)
+	app, err := h.storage.GetAppByAppID(appID)
 
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
@@ -218,30 +258,30 @@ func getChannels(ctx *applicationContext, p params, w http.ResponseWriter, r *ht
 	case "presence-":
 		for _, c := range app.PresenceChannels() {
 			if requestedUserCount {
-				channels[c.ChannelID] = struct {
+				channels[c.ID] = struct {
 					UserCount int `json:"user_count"`
 				}{
 					c.TotalUsers(),
 				}
 			} else {
-				channels[c.ChannelID] = struct{}{}
+				channels[c.ID] = struct{}{}
 			}
 		}
 	case "public-":
 		for _, c := range app.PublicChannels() {
-			channels[c.ChannelID] = struct{}{}
+			channels[c.ID] = struct{}{}
 		}
 	case "private-":
 		for _, c := range app.PrivateChannels() {
-			channels[c.ChannelID] = struct{}{}
+			channels[c.ID] = struct{}{}
 		}
 	default:
-		for _, c := range app.Channels {
+		for _, c := range app.Channels() {
-			channels[c.ChannelID] = struct{}{}
+			channels[c.ID] = struct{}{}
 		}
 	}
 
-	w.Header().Set("Content-Type", "application/json;charset=UTF-8")
+	w.Header().Set("Content-Type", "application/json")
 
 	js := make(map[string]interface{}, 1)
 	js["channels"] = channels
@@ -252,7 +292,15 @@ func getChannels(ctx *applicationContext, p params, w http.ResponseWriter, r *ht
 	}
 }
 
-// Fetch info for one channel
+// GetChannel handle get channel
+type GetChannel struct{ storage storage.Storage }
+
+// NewGetChannel return a new GetChannel handler
+func NewGetChannel(storage storage.Storage) *GetChannel {
+	return &GetChannel{storage: storage}
+}
+
+// ServeHTTP Fetch info for one channel
 //
 // Example:
 // {
@@ -262,29 +310,28 @@ func getChannels(ctx *applicationContext, p params, w http.ResponseWriter, r *ht
 // }
 //
 // GET /apps/{app_id}/channels/{channel_name}
-func getChannel(ctx *applicationContext, p params, w http.ResponseWriter, r *http.Request) {
+func (h *GetChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	w.Header().Set("Content-Type", "application/json;charset=UTF-8")
+	var (
+		pathVars    = mux.Vars(r)
+		queryVars   = r.URL.Query()
+		appID       = pathVars["app_id"]
+		channelName = pathVars["channel_name"]
+		info        = queryVars.Get("info")
+		attributes  = strings.Split(info, ",")
+	)
 
-	query := r.URL.Query()
+	app, err := h.storage.GetAppByAppID(appID)
-
-	appID := p.Get("app_id")
-	app, err := ctx.DB.GetAppByAppID(appID)
 
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
 	}
 
-	channelName := p.Get("channel_name")
-
 	// Channel name could not be empty
 	if strings.TrimSpace(channelName) == "" {
 		http.Error(w, "Empty channel name", http.StatusBadRequest)
 		return
 	}
 
-	info := query.Get("info")
-	attributes := strings.Split(info, ",")
-
 	// Attributes requested
 	requestedUserCount := false
 	requestedSubscriptionCount := false
@@ -332,15 +379,22 @@ func getChannel(ctx *applicationContext, p params, w http.ResponseWriter, r *htt
 		dtoChannel.SubscriptionCount = channel.TotalSubscriptions()
 	}
 
-	w.Header().Set("Content-Type", "application/json;charset=UTF-8")
+	w.Header().Set("Content-Type", "application/json")
-
 	if err := json.NewEncoder(w).Encode(dtoChannel); err != nil {
 		log.Error(err)
 		http.Error(w, "Internal server error", http.StatusInternalServerError)
 	}
 }
 
-// Allowed only for presence-channels
+// GetChannelUsers handle get users from a channel
+type GetChannelUsers struct{ storage storage.Storage }
+
+// NewGetChannelUsers return a new GetChannelUsers handler
+func NewGetChannelUsers(storage storage.Storage) *GetChannelUsers {
+	return &GetChannelUsers{storage: storage}
+}
+
+// ServeHTTP Allowed only for presence-channels
 //
 // Example:
 // {
@@ -351,9 +405,12 @@ func getChannel(ctx *applicationContext, p params, w http.ResponseWriter, r *htt
 // }
 //
 // GET /apps/{app_id}/channels/{channel_name}/users
-func getChannelUsers(ctx *applicationContext, p params, w http.ResponseWriter, r *http.Request) {
+func (h *GetChannelUsers) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	appID := p.Get("app_id")
+	var (
-	channelName := p.Get("channel_name")
+		pathVars    = mux.Vars(r)
+		appID       = pathVars["app_id"]
+		channelName = pathVars["channel_name"]
+	)
 
 	isPresence := utils.IsPresenceChannel(channelName)
 
@@ -362,7 +419,7 @@ func getChannelUsers(ctx *applicationContext, p params, w http.ResponseWriter, r
 		return
 	}
 
-	app, err := ctx.DB.GetAppByAppID(appID)
+	app, err := h.storage.GetAppByAppID(appID)
 
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusBadRequest)
@@ -381,7 +438,7 @@ func getChannelUsers(ctx *applicationContext, p params, w http.ResponseWriter, r
 
 	var users []interface{}
 
-	for _, s := range channel.Subscriptions {
+	for _, s := range channel.Subscriptions() {
 		users = append(users, struct {
 			ID string `json:"id"`
 		}{s.ID})
@@ -389,8 +446,7 @@ func getChannelUsers(ctx *applicationContext, p params, w http.ResponseWriter, r
 
 	result["users"] = users
 
-	w.Header().Set("Content-Type", "application/json;charset=UTF-8")
+	w.Header().Set("Content-Type", "application/json")
-
 	if err := json.NewEncoder(w).Encode(result); err != nil {
 		http.Error(w, "Internal server error", http.StatusInternalServerError)
 		log.Error(err)

api/handlers_test.go

@@ -1,57 +1,74 @@
-package ipe
+package api
 
 import (
 	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"testing"
+
+	"github.com/gorilla/mux"
+
+	"ipe/app"
+	channel2 "ipe/channel"
+	"ipe/connection"
+	"ipe/mocks"
+	"ipe/storage"
 )
 
 var (
-	testApp *app
+	testApp  *app.Application
-	ctx     *applicationContext
+	database storage.Storage
+	id       = 0
 )
 
+func newTestApp() *app.Application {
+	a := app.NewApplication("Test", strconv.Itoa(id), "123", "123", false, false, true, false, "")
+	id++
+
+	return a
+}
+
 func init() {
 	testApp = newTestApp()
 
-	channel := newChannel("presence-c1")
+	channel := channel2.New("presence-c1")
 	testApp.AddChannel(channel)
-	testApp.AddChannel(newChannel("c2"))
+	testApp.AddChannel(channel2.New("c2"))
-	testApp.AddChannel(newChannel("private-c3"))
+	testApp.AddChannel(channel2.New("private-c3"))
 
-	conn := newConnection("123.456", mockSocket{})
+	conn := connection.New("123.456", mocks.MockSocket{})
-	testApp.Subscribe(channel, conn, "{}")
+	_ = testApp.Subscribe(channel, conn, "{}")
 
-	conn = newConnection("321.654", mockSocket{})
+	conn = connection.New("321.654", mocks.MockSocket{})
-	testApp.Subscribe(channel, conn, "{}")
+	_ = testApp.Subscribe(channel, conn, "{}")
 
-	db := newMemdb()
+	_storage := storage.NewInMemory()
-	db.AddApp(testApp)
+	_ = _storage.AddApp(testApp)
 
-	ctx = &applicationContext{DB: db}
+	database = _storage
 }
 
-// All Channels
+// All channels
 func Test_getChannels_all(t *testing.T) {
-
 	appID := testApp.AppID
 
-	p := map[string]string{}
-	p["app_id"] = appID
-
 	r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels", appID), nil)
+	r = mux.SetURLVars(r, map[string]string{
+		"app_id": appID,
+	})
 	w := httptest.NewRecorder()
 
-	getChannels(ctx, params(p), w, r)
+	handler := &GetChannels{database}
+	handler.ServeHTTP(w, r)
 
 	if w.Code != http.StatusOK {
 		t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
 	}
 
 	data := make(map[string]interface{})
-	json.Unmarshal(w.Body.Bytes(), &data)
+	_ = json.Unmarshal(w.Body.Bytes(), &data)
 
 	channels := data["channels"].(map[string]interface{})
 
@@ -63,20 +80,21 @@ func Test_getChannels_all(t *testing.T) {
 func Test_getChannels_filter_by_presence_prefix(t *testing.T) {
 	appID := testApp.AppID
 
-	p := map[string]string{}
-	p["app_id"] = appID
-
 	r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=presence-", appID), nil)
+	r = mux.SetURLVars(r, map[string]string{
+		"app_id": appID,
+	})
 	w := httptest.NewRecorder()
 
-	getChannels(ctx, params(p), w, r)
+	handler := &GetChannels{database}
+	handler.ServeHTTP(w, r)
 
 	if w.Code != http.StatusOK {
 		t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
 	}
 
 	data := make(map[string]interface{})
-	json.Unmarshal(w.Body.Bytes(), &data)
+	_ = json.Unmarshal(w.Body.Bytes(), &data)
 
 	channels := data["channels"].(map[string]interface{})
 
@@ -87,23 +105,23 @@ func Test_getChannels_filter_by_presence_prefix(t *testing.T) {
 
 // Only presence channels and user_count
 func Test_getChannels_filter_by_presence_prefix_and_user_count(t *testing.T) {
-
 	appID := testApp.AppID
 
-	p := map[string]string{}
-	p["app_id"] = appID
-
 	r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=presence-&info=user_count", appID), nil)
+	r = mux.SetURLVars(r, map[string]string{
+		"app_id": appID,
+	})
 	w := httptest.NewRecorder()
 
-	getChannels(ctx, params(p), w, r)
+	handler := &GetChannels{database}
+	handler.ServeHTTP(w, r)
 
 	if w.Code != http.StatusOK {
 		t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
 	}
 
 	data := make(map[string]interface{})
-	json.Unmarshal(w.Body.Bytes(), &data)
+	_ = json.Unmarshal(w.Body.Bytes(), &data)
 
 	channels := data["channels"].(map[string]interface{})
 
@@ -124,17 +142,18 @@ func Test_getChannels_filter_by_presence_prefix_and_user_count(t *testing.T) {
 	}
 }
 
-// User count only alowed in Presence channels
+// User count only allowed in Presence channels
 func Test_getChannels_filter_by_private_prefix_and_info_user_count(t *testing.T) {
 	appID := testApp.AppID
 
-	p := map[string]string{}
-	p["app_id"] = appID
-
 	r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=private-&info=user_count", appID), nil)
+	r = mux.SetURLVars(r, map[string]string{
+		"app_id": appID,
+	})
 	w := httptest.NewRecorder()
 
-	getChannels(ctx, params(p), w, r)
+	handler := &GetChannels{database}
+	handler.ServeHTTP(w, r)
 
 	if w.Code != http.StatusBadRequest {
 		t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusBadRequest)
@@ -144,13 +163,14 @@ func Test_getChannels_filter_by_private_prefix_and_info_user_count(t *testing.T)
 func Test_getChannels_filter_by_public_prefix(t *testing.T) {
 	appID := testApp.AppID
 
-	p := map[string]string{}
-	p["app_id"] = appID
-
 	r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=public-", appID), nil)
+	r = mux.SetURLVars(r, map[string]string{
+		"app_id": appID,
+	})
 	w := httptest.NewRecorder()
 
-	getChannels(ctx, params(p), w, r)
+	handler := &GetChannels{database}
+	handler.ServeHTTP(w, r)
 
 	if w.Code != http.StatusOK {
 		t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
@@ -158,7 +178,7 @@ func Test_getChannels_filter_by_public_prefix(t *testing.T) {
 
 	data := make(map[string]interface{})
 
-	json.Unmarshal(w.Body.Bytes(), &data)
+	_ = json.Unmarshal(w.Body.Bytes(), &data)
 
 	channels := data["channels"].(map[string]interface{})
 
@@ -174,16 +194,16 @@ func Test_getChannels_filter_by_public_prefix(t *testing.T) {
 }
 
 func Test_getChannels_filter_by_private_prefix(t *testing.T) {
-
 	appID := testApp.AppID
 
-	p := map[string]string{}
-	p["app_id"] = appID
-
 	r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels?filter_by_prefix=private-", appID), nil)
+	r = mux.SetURLVars(r, map[string]string{
+		"app_id": appID,
+	})
 	w := httptest.NewRecorder()
 
-	getChannels(ctx, params(p), w, r)
+	handler := &GetChannels{database}
+	handler.ServeHTTP(w, r)
 
 	if w.Code != http.StatusOK {
 		t.Errorf("w.Code == %d, wants %d", w.Code, http.StatusOK)
@@ -191,7 +211,7 @@ func Test_getChannels_filter_by_private_prefix(t *testing.T) {
 
 	data := make(map[string]interface{})
 
-	json.Unmarshal(w.Body.Bytes(), &data)
+	_ = json.Unmarshal(w.Body.Bytes(), &data)
 
 	channels := data["channels"].(map[string]interface{})
 

app/app.go

@@ -0,0 +1,314 @@
+// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package app
+
+import (
+	"errors"
+	"expvar"
+	"fmt"
+	"sync"
+
+	log "github.com/golang/glog"
+
+	"ipe/channel"
+	"ipe/connection"
+	"ipe/events"
+	"ipe/subscription"
+)
+
+// Application represents a Pusher application
+type Application struct {
+	sync.RWMutex
+
+	Name       string
+	AppID      string
+	Key        string
+	Secret     string
+	OnlySSL    bool
+	Enabled    bool
+	UserEvents bool
+	WebHooks   bool
+	URLWebHook string
+
+	channels    map[string]*channel.Channel
+	connections map[string]*connection.Connection
+
+	Stats *expvar.Map `json:"-"`
+}
+
+// NewApplication returns a new Application
+func NewApplication(
+	name,
+	appID,
+	key,
+	secret string,
+	onlySSL,
+	enabled,
+	userEvents,
+	webHooks bool,
+	webHookURL string,
+) *Application {
+
+	a := &Application{
+		Name:       name,
+		AppID:      appID,
+		Key:        key,
+		Secret:     secret,
+		OnlySSL:    onlySSL,
+		Enabled:    enabled,
+		UserEvents: userEvents,
+		WebHooks:   webHooks,
+		URLWebHook: webHookURL,
+	}
+
+	a.connections = make(map[string]*connection.Connection)
+	a.channels = make(map[string]*channel.Channel)
+	a.Stats = expvar.NewMap(fmt.Sprintf("%s (%s)", a.Name, a.AppID))
+
+	return a
+}
+
+// Channels returns the full list of channels
+func (a *Application) Channels() []*channel.Channel {
+	a.RLock()
+	defer a.RUnlock()
+
+	var channels []*channel.Channel
+
+	for _, c := range a.channels {
+		channels = append(channels, c)
+	}
+
+	return channels
+}
+
+// PresenceChannels Only Presence channels
+func (a *Application) PresenceChannels() []*channel.Channel {
+	a.RLock()
+	defer a.RUnlock()
+
+	var channels []*channel.Channel
+
+	for _, c := range a.channels {
+		if c.IsPresence() {
+			channels = append(channels, c)
+		}
+	}
+
+	return channels
+}
+
+// PrivateChannels Only Private channels
+func (a *Application) PrivateChannels() []*channel.Channel {
+	a.RLock()
+	defer a.RUnlock()
+
+	var channels []*channel.Channel
+
+	for _, c := range a.channels {
+		if c.IsPrivate() {
+			channels = append(channels, c)
+		}
+	}
+
+	return channels
+}
+
+// PublicChannels Only Public channels
+func (a *Application) PublicChannels() []*channel.Channel {
+	a.RLock()
+	defer a.RUnlock()
+
+	var channels []*channel.Channel
+
+	for _, c := range a.channels {
+		if c.IsPublic() {
+			channels = append(channels, c)
+		}
+	}
+
+	return channels
+}
+
+// Disconnect Socket
+func (a *Application) Disconnect(socketID string) {
+	log.Infof("disconnecting socket %+v", socketID)
+
+	conn, err := a.FindConnection(socketID)
+
+	if err != nil {
+		log.Infof("socket not found, %+v", err)
+		return
+	}
+
+	// Unsubscribe from channels
+	for _, c := range a.channels {
+		if c.IsSubscribed(conn) {
+			if err := c.Unsubscribe(conn); err != nil {
+				log.Errorf("error while calling Channel.Unsubscribe, %+v", err)
+				continue
+			}
+		}
+	}
+
+	// Remove from Application
+	a.Lock()
+	_, exists := a.connections[conn.SocketID]
+	a.Unlock()
+
+	if !exists {
+		return
+	}
+
+	a.Lock()
+	delete(a.connections, conn.SocketID)
+	a.Unlock()
+
+	a.Stats.Add("TotalConnections", -1)
+}
+
+// Connect a new Subscriber
+func (a *Application) Connect(conn *connection.Connection) {
+	log.Infof("adding a new Connection %s to Application %s", conn.SocketID, a.Name)
+	a.Lock()
+	defer a.Unlock()
+
+	a.connections[conn.SocketID] = conn
+
+	a.Stats.Add("TotalConnections", 1)
+}
+
+// FindConnection Find a Connection on this Application
+func (a *Application) FindConnection(socketID string) (*connection.Connection, error) {
+	a.RLock()
+	defer a.RUnlock()
+
+	conn, exists := a.connections[socketID]
+
+	if exists {
+		return conn, nil
+	}
+
+	return nil, errors.New("connection not found")
+}
+
+// RemoveChannel removes the Channel from Application
+func (a *Application) RemoveChannel(c *channel.Channel) {
+	log.Infof("remove the Channel %s from Application %s", c.ID, a.Name)
+	a.Lock()
+	defer a.Unlock()
+
+	delete(a.channels, c.ID)
+
+	if c.IsPresence() {
+		a.Stats.Add("TotalPresenceChannels", -1)
+	}
+
+	if c.IsPrivate() {
+		a.Stats.Add("TotalPrivateChannels", -1)
+	}
+
+	if c.IsPublic() {
+		a.Stats.Add("TotalPublicChannels", -1)
+	}
+
+	a.Stats.Add("TotalChannels", -1)
+}
+
+// AddChannel Add a new Channel to this APP
+func (a *Application) AddChannel(c *channel.Channel) {
+	log.Infof("adding a new Channel %s to Application %s", c.ID, a.Name)
+
+	a.Lock()
+	defer a.Unlock()
+
+	a.channels[c.ID] = c
+
+	if c.IsPresence() {
+		a.Stats.Add("TotalPresenceChannels", 1)
+	}
+
+	if c.IsPrivate() {
+		a.Stats.Add("TotalPrivateChannels", 1)
+	}
+
+	if c.IsPublic() {
+		a.Stats.Add("TotalPublicChannels", 1)
+	}
+
+	a.Stats.Add("TotalChannels", 1)
+}
+
+// FindOrCreateChannelByChannelID Returns a Channel from this Application
+// If not found then the Channel is created and added to this Application
+func (a *Application) FindOrCreateChannelByChannelID(n string) *channel.Channel {
+	c, err := a.FindChannelByChannelID(n)
+
+	if err != nil {
+		c = channel.New(
+			n,
+			channel.WithChannelOccupiedListener(func(c *channel.Channel, s *subscription.Subscription) {
+				a.TriggerChannelOccupiedHook(c)
+			}),
+			channel.WithChannelVacatedListener(func(c *channel.Channel, s *subscription.Subscription) {
+				a.TriggerChannelVacatedHook(c)
+			}),
+			channel.WithMemberAddedListener(func(c *channel.Channel, s *subscription.Subscription) {
+				a.TriggerMemberAddedHook(c, s)
+			}),
+			channel.WithMemberRemovedListener(func(c *channel.Channel, s *subscription.Subscription) {
+				a.TriggerMemberRemovedHook(c, s)
+			}),
+			channel.WithClientEventListener(func(c *channel.Channel, s *subscription.Subscription, event string, data interface{}) {
+				a.TriggerClientEventHook(c, s, event, data)
+			}),
+		)
+		a.AddChannel(c)
+	}
+
+	return c
+}
+
+// FindChannelByChannelID Find the Channel by Channel ID
+func (a *Application) FindChannelByChannelID(n string) (*channel.Channel, error) {
+	a.RLock()
+	defer a.RUnlock()
+
+	c, exists := a.channels[n]
+
+	if exists {
+		return c, nil
+	}
+
+	return nil, errors.New("channel does not exists")
+}
+
+// Publish an event into the channel
+// skip the ignore connection
+func (a *Application) Publish(c *channel.Channel, event events.Raw, ignore string) error {
+	a.Stats.Add("TotalUniqueMessages", 1)
+
+	return c.Publish(event, ignore)
+}
+
+// Unsubscribe unsubscribe the given connection from the channel
+// remove the channel from the application if it is empty
+func (a *Application) Unsubscribe(c *channel.Channel, conn *connection.Connection) error {
+	err := c.Unsubscribe(conn)
+	if err != nil {
+		return err
+	}
+
+	if !c.IsOccupied() {
+		a.RemoveChannel(c)
+	}
+
+	return nil
+}
+
+// Subscribe the connection into the given channel
+func (a *Application) Subscribe(c *channel.Channel, conn *connection.Connection, data string) error {
+	return c.Subscribe(conn, data)
+}

app/app_test.go

@@ -0,0 +1,255 @@
+// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package app
+
+import (
+	"strconv"
+	"testing"
+
+	channel2 "ipe/channel"
+	"ipe/connection"
+	"ipe/mocks"
+)
+
+var id = 0
+
+func newTestApp() *Application {
+	a := NewApplication("Test", strconv.Itoa(id), "123", "123", false, false, true, false, "")
+	id++
+
+	return a
+}
+
+func TestConnect(t *testing.T) {
+	app := newTestApp()
+
+	app.Connect(connection.New("socketID", mocks.MockSocket{}))
+
+	if len(app.connections) != 1 {
+		t.Errorf("len(Application.connections) == %d, wants %d", len(app.connections), 1)
+	}
+
+}
+
+func TestDisconnect(t *testing.T) {
+	app := newTestApp()
+
+	app.Connect(connection.New("socketID", mocks.MockSocket{}))
+	app.Disconnect("socketID")
+
+	if len(app.connections) != 0 {
+		t.Errorf("len(Application.connections) == %d, wants %d", len(app.connections), 0)
+	}
+
+}
+
+func TestFindConnection(t *testing.T) {
+	app := newTestApp()
+
+	app.Connect(connection.New("socketID", mocks.MockSocket{}))
+
+	if _, err := app.FindConnection("socketID"); err != nil {
+		t.Errorf("Application.FindConnection('socketID') == _, %q, wants %v", err, nil)
+	}
+
+	if _, err := app.FindConnection("NotFound"); err == nil {
+		t.Errorf("Application.FindConnection('socketID') == _, %q, wants !nil", err)
+	}
+
+}
+
+func TestFindChannelByChannelID(t *testing.T) {
+	app := newTestApp()
+
+	channel := channel2.New("ID")
+	app.AddChannel(channel)
+
+	if _, err := app.FindChannelByChannelID("ID"); err != nil {
+		t.Errorf("Application.FindChannelByChannelID('ID') == _, %q, wants %v", err, nil)
+	}
+}
+
+func TestFindOrCreateChannelByChannelID(t *testing.T) {
+	app := newTestApp()
+
+	if len(app.channels) != 0 {
+		t.Errorf("len(Application.channels) == %d, wants %d", len(app.channels), 0)
+	}
+
+	app.FindOrCreateChannelByChannelID("ID")
+
+	if len(app.channels) != 1 {
+		t.Errorf("len(Application.channels) == %d, wants %d", len(app.channels), 1)
+	}
+
+}
+
+func TestRemoveChannel(t *testing.T) {
+	app := newTestApp()
+
+	if len(app.channels) != 0 {
+		t.Errorf("len(Application.channels) == %d, wants %d", len(app.channels), 0)
+	}
+
+	channel := channel2.New("ID")
+	app.AddChannel(channel)
+
+	if len(app.channels) != 1 {
+		t.Errorf("len(Application.channels) == %d, wants %d", len(app.channels), 1)
+	}
+
+	app.RemoveChannel(channel)
+
+	if len(app.channels) != 0 {
+		t.Errorf("len(Application.channels) == %d, wants %d", len(app.channels), 0)
+	}
+
+}
+
+func Test_add_channels(t *testing.T) {
+
+	app := newTestApp()
+
+	// Public
+
+	if len(app.PublicChannels()) != 0 {
+		t.Errorf("len(Application.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 0)
+	}
+
+	app.AddChannel(channel2.New("ID"))
+
+	if len(app.PublicChannels()) != 1 {
+		t.Errorf("len(Application.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 1)
+	}
+
+	// Presence
+
+	if len(app.PresenceChannels()) != 0 {
+		t.Errorf("len(Application.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 0)
+	}
+
+	app.AddChannel(channel2.New("presence-test"))
+
+	if len(app.PresenceChannels()) != 1 {
+		t.Errorf("len(Application.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 1)
+	}
+
+	// Private
+
+	if len(app.PrivateChannels()) != 0 {
+		t.Errorf("len(Application.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 0)
+	}
+
+	app.AddChannel(channel2.New("private-test"))
+
+	if len(app.PrivateChannels()) != 1 {
+		t.Errorf("len(Application.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 1)
+	}
+
+}
+
+func Test_AllChannels(t *testing.T) {
+	app := newTestApp()
+	app.AddChannel(channel2.New("private-test"))
+	app.AddChannel(channel2.New("presence-test"))
+	app.AddChannel(channel2.New("test"))
+
+	if len(app.channels) != 3 {
+		t.Errorf("len(Application.channels) == %d, wants %d", len(app.channels), 3)
+	}
+}
+
+func Test_New_Subscriber(t *testing.T) {
+	app := newTestApp()
+
+	if len(app.connections) != 0 {
+		t.Errorf("len(Application.connections) == %d, wants %d", len(app.connections), 0)
+	}
+
+	conn := connection.New("1", mocks.MockSocket{})
+	app.Connect(conn)
+
+	if len(app.connections) != 1 {
+		t.Errorf("len(Application.connections) == %d, wants %d", len(app.connections), 1)
+	}
+}
+
+func Test_find_subscriber(t *testing.T) {
+	app := newTestApp()
+	conn := connection.New("1", mocks.MockSocket{})
+	app.Connect(conn)
+
+	conn, err := app.FindConnection("1")
+
+	if err != nil {
+		t.Error(err)
+	}
+
+	if conn.SocketID != "1" {
+		t.Errorf("conn.SocketID == %s, wants %s", conn.SocketID, "1")
+	}
+
+	// Find a wrong subscriber
+
+	conn, err = app.FindConnection("DoesNotExists")
+
+	if err == nil {
+		t.Errorf("err == %q, wants !nil", err)
+	}
+
+	if conn != nil {
+		t.Errorf("conn == %q, wants nil", conn)
+	}
+}
+
+func Test_find_or_create_channels(t *testing.T) {
+	app := newTestApp()
+
+	// Public
+	if len(app.PublicChannels()) != 0 {
+		t.Errorf("len(Application.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 0)
+	}
+
+	c := app.FindOrCreateChannelByChannelID("id")
+
+	if len(app.PublicChannels()) != 1 {
+		t.Errorf("len(Application.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 1)
+	}
+
+	if c.ID != "id" {
+		t.Errorf("c.id == %s, wants %s", c.ID, "id")
+	}
+
+	// Presence
+	if len(app.PresenceChannels()) != 0 {
+		t.Errorf("len(Application.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 0)
+	}
+
+	c = app.FindOrCreateChannelByChannelID("presence-test")
+
+	if len(app.PresenceChannels()) != 1 {
+		t.Errorf("len(Application.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 1)
+	}
+
+	if c.ID != "presence-test" {
+		t.Errorf("c.id == %s, wants %s", c.ID, "presence-test")
+	}
+
+	// Private
+	if len(app.PrivateChannels()) != 0 {
+		t.Errorf("len(Application.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 0)
+	}
+
+	c = app.FindOrCreateChannelByChannelID("private-test")
+
+	if len(app.PrivateChannels()) != 1 {
+		t.Errorf("len(Application.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 1)
+	}
+
+	if c.ID != "private-test" {
+		t.Errorf("c.id == %s, wants %s", c.ID, "private-test")
+	}
+
+}

app/webhooks.go

@@ -2,19 +2,26 @@
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
 
-package ipe
+package app
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"time"
 
-	"github.com/dimiro1/ipe/utils"
 	log "github.com/golang/glog"
+
+	"ipe/channel"
+	"ipe/subscription"
+	"ipe/utils"
 )
 
-// A WebHook is sent as a HTTP POST request to the url which you specify.
+const maxTimeout = 3 * time.Second
+
+// A webHook is sent as a HTTP POST request to the url which you specify.
 // The POST request payload (body) contains a JSON document, and follows the following format:
 // {
 //   "time_ms": 1327078148132
@@ -29,7 +36,7 @@ import (
 // You may use a HTTP or a HTTPS url for WebHooks. In most cases HTTP is sufficient, but HTTPS can be useful if your data is sensitive or if you wish to protect against replay attacks for example.
 // Authentication
 //
-// Since anyone could in principle send WebHooks to your application, it’s important to verify that these WebHooks originated from Pusher. Valid WebHooks will therefore contain these headers which contain a HMAC signature of the WebHook payload (body):
+// Since anyone could in principle send WebHooks to your application, it’s important to verify that these WebHooks originated from Pusher. Valid WebHooks will therefore contain these headers which contain a HMAC signature of the webHook payload (body):
 //
 //     X-Pusher-Key: The App Key.
 //     X-Pusher-Signature: A HMAC SHA256 hex digest formed by signing the POST payload (body) with the token’s secret.
@@ -47,40 +54,51 @@ type hookEvent struct {
 	UserID   string      `json:"user_id,omitempty"`
 }
 
-func newChannelOcuppiedHook(channel *channel) hookEvent {
+func newChannelOcuppiedHook(channel *channel.Channel) hookEvent {
-	return hookEvent{Name: "channel_occupied", Channel: channel.ChannelID}
+	return hookEvent{Name: "channel_occupied", Channel: channel.ID}
 }
 
-func newChannelVacatedHook(channel *channel) hookEvent {
+func newChannelVacatedHook(channel *channel.Channel) hookEvent {
-	return hookEvent{Name: "channel_vacated", Channel: channel.ChannelID}
+	return hookEvent{Name: "channel_vacated", Channel: channel.ID}
 }
 
-func newMemberAddedHook(channel *channel, s *subscription) hookEvent {
+func newMemberAddedHook(channel *channel.Channel, s *subscription.Subscription) hookEvent {
-	return hookEvent{Name: "member_added", Channel: channel.ChannelID, UserID: s.ID}
+	return hookEvent{Name: "member_added", Channel: channel.ID, UserID: s.ID}
 }
 
-func newMemberRemovedHook(channel *channel, s *subscription) hookEvent {
+func newMemberRemovedHook(channel *channel.Channel, s *subscription.Subscription) hookEvent {
-	return hookEvent{Name: "member_removed", Channel: channel.ChannelID, UserID: s.ID}
+	return hookEvent{Name: "member_removed", Channel: channel.ID, UserID: s.ID}
 }
 
-func newClientHook(channel *channel, s *subscription, event string, data interface{}) hookEvent {
+func newClientHook(channel *channel.Channel, s *subscription.Subscription, event string, data interface{}) hookEvent {
-	return hookEvent{Name: "client_event", Channel: channel.ChannelID, Event: event, Data: data, SocketID: s.Connection.SocketID}
+	return hookEvent{Name: "client_event", Channel: channel.ID, Event: event, Data: data, SocketID: s.Connection.SocketID}
 }
 
-// channel_occupied
+// TriggerChannelOccupiedHook channel_occupied
 // { "name": "channel_occupied", "channel": "test_channel" }
-func (a *app) TriggerChannelOccupiedHook(c *channel) {
+func (a *Application) TriggerChannelOccupiedHook(c *channel.Channel) {
 	event := newChannelOcuppiedHook(c)
-	triggerHook(event.Name, a, c, event)
+	ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
+	defer cancel()
+
+	if err := triggerHook(ctx, a, event); err != nil {
+		log.Errorf("triggering webhook %+v", err)
+	}
 }
 
-// channel_vacated
+// TriggerChannelVacatedHook channel_vacated
 // { "name": "channel_vacated", "channel": "test_channel" }
-func (a *app) TriggerChannelVacatedHook(c *channel) {
+func (a *Application) TriggerChannelVacatedHook(c *channel.Channel) {
 	event := newChannelVacatedHook(c)
-	triggerHook(event.Name, a, c, event)
+	ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
+	defer cancel()
+
+	if err := triggerHook(ctx, a, event); err != nil {
+		log.Errorf("triggering webhook %+v", err)
+	}
 }
 
+// TriggerClientEventHook client_events
 // {
 //   "name": "client_event",
 //   "channel": "name of the channel the event was published on",
@@ -89,44 +107,63 @@ func (a *app) TriggerChannelVacatedHook(c *channel) {
 //   "socket_id": "socket_id of the sending socket",
 //   "user_id": "user_id associated with the sending socket" # Only for presence channels
 // }
-func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent string, data interface{}) {
+func (a *Application) TriggerClientEventHook(c *channel.Channel, s *subscription.Subscription, clientEvent string, data interface{}) {
 	event := newClientHook(c, s, clientEvent, data)
 
 	if c.IsPresence() {
 		event.UserID = s.ID
 	}
 
-	triggerHook(event.Name, a, c, event)
+	ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
+	defer cancel()
+
+	if err := triggerHook(ctx, a, event); err != nil {
+		log.Errorf("triggering webhook %+v", err)
+	}
 }
 
+// TriggerMemberAddedHook member_added
 // {
 //   "name": "member_added",
 //   "channel": "presence-your_channel_name",
 //   "user_id": "a_user_id"
 // }
-func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) {
+func (a *Application) TriggerMemberAddedHook(c *channel.Channel, s *subscription.Subscription) {
 	event := newMemberAddedHook(c, s)
-	triggerHook(event.Name, a, c, event)
+	ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
+	defer cancel()
+
+	if err := triggerHook(ctx, a, event); err != nil {
+		log.Errorf("triggering webhook %+v", err)
+	}
 }
 
+// TriggerMemberRemovedHook member_removed
 // {
 //   "name": "member_removed",
 //   "channel": "presence-your_channel_name",
 //   "user_id": "a_user_id"
 // }
-func (a *app) TriggerMemberRemovedHook(c *channel, s *subscription) {
+func (a *Application) TriggerMemberRemovedHook(c *channel.Channel, s *subscription.Subscription) {
 	event := newMemberRemovedHook(c, s)
-	triggerHook(event.Name, a, c, event)
+	ctx, cancel := context.WithTimeout(context.Background(), maxTimeout)
+	defer cancel()
+
+	if err := triggerHook(ctx, a, event); err != nil {
+		log.Errorf("triggering webhook %+v", err)
+	}
 }
 
-func triggerHook(name string, a *app, c *channel, event hookEvent) {
+func triggerHook(ctx context.Context, a *Application, event hookEvent) error {
 	if !a.WebHooks {
-		log.Infof("Webhooks are not enabled for app: %s", a.Name)
+		log.Infof("webhook are not enabled for app: %s", a.Name)
-		return
+		return fmt.Errorf("webhooks are not enabled for app: %s", a.Name)
 	}
 
+	done := make(chan bool)
+
 	go func() {
-		log.Infof("Triggering %s event", name)
+		log.Infof("Triggering %s event", event.Name)
 
 		hook := webHook{TimeMs: time.Now().Unix()}
 
@@ -145,11 +182,14 @@ func triggerHook(name string, a *app, c *channel, event hookEvent) {
 		var req *http.Request
 
 		req, err = http.NewRequest("POST", a.URLWebHook, bytes.NewReader(js))
+
 		if err != nil {
 			log.Errorf("Error creating request: %+v", err)
 			return
 		}
 
+		req = req.WithContext(ctx)
+
 		req.Header.Set("User-Agent", "Ipe UA; (+https://github.com/dimiro1/ipe)")
 		req.Header.Set("Content-Type", "application/json")
 		req.Header.Set("X-Pusher-Key", a.Key)
@@ -162,11 +202,25 @@ func triggerHook(name string, a *app, c *channel, event hookEvent) {
 
 		// See: http://devs.cloudimmunity.com/gotchas-and-common-mistakes-in-go-golang/index.html#close_http_resp_body
 		if resp != nil {
-			defer resp.Body.Close()
+			defer func() {
+				if err := resp.Body.Close(); err != nil {
+					log.Errorf("error closing response body %+v", err)
+				}
+			}()
 		}
 
 		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
+	}
 }

channel/channel.go

@@ -0,0 +1,324 @@
+// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package channel
+
+import (
+	"encoding/json"
+	"errors"
+	"sync"
+	"time"
+
+	log "github.com/golang/glog"
+
+	"ipe/connection"
+	"ipe/events"
+	"ipe/subscription"
+	"ipe/utils"
+)
+
+// Option constructor function for Channel
+type Option func(*Channel)
+
+// ListenerFunc listener function
+type ListenerFunc func(*Channel, *subscription.Subscription)
+
+// ClientEventListenerFunc listener for client events
+type ClientEventListenerFunc func(*Channel, *subscription.Subscription, string, interface{})
+
+// Channel represents an application channel
+type Channel struct {
+	sync.RWMutex
+
+	ID            string
+	subscriptions map[string]*subscription.Subscription
+
+	createdAt time.Time
+
+	memberAddedListeners     []ListenerFunc
+	memberRemovedListeners   []ListenerFunc
+	channelOccupiedListeners []ListenerFunc
+	channelVacatedListeners  []ListenerFunc
+	clientEventListeners     []ClientEventListenerFunc
+}
+
+// New Create a new Channel
+func New(channelID string, options ...Option) *Channel {
+	log.Infof("Creating a new Channel: %s", channelID)
+
+	c := &Channel{ID: channelID, createdAt: time.Now(), subscriptions: make(map[string]*subscription.Subscription)}
+
+	for _, option := range options {
+		option(c)
+	}
+
+	return c
+}
+
+// WithMemberAddedListener appends the given ListenerFunc into the memberAddedListeners list
+func WithMemberAddedListener(f ListenerFunc) func(*Channel) {
+	return func(c *Channel) {
+		c.memberAddedListeners = append(c.memberAddedListeners, f)
+	}
+}
+
+// WithMemberRemovedListener appends the given ListenerFunc into the memberRemovedListeners list
+func WithMemberRemovedListener(f ListenerFunc) func(*Channel) {
+	return func(c *Channel) {
+		c.memberRemovedListeners = append(c.memberRemovedListeners, f)
+	}
+}
+
+// WithChannelOccupiedListener appends the given ListenerFunc into the channelOccupiedListeners list
+func WithChannelOccupiedListener(f ListenerFunc) func(*Channel) {
+	return func(c *Channel) {
+		c.channelOccupiedListeners = append(c.channelOccupiedListeners, f)
+	}
+}
+
+// WithChannelVacatedListener appends the given ListenerFunc into the channelVacatedListeners list
+func WithChannelVacatedListener(f ListenerFunc) func(*Channel) {
+	return func(c *Channel) {
+		c.channelVacatedListeners = append(c.channelVacatedListeners, f)
+	}
+}
+
+// WithClientEventListener appends the given ListenerFunc into the clientEventListeners list
+func WithClientEventListener(f ClientEventListenerFunc) func(*Channel) {
+	return func(c *Channel) {
+		c.clientEventListeners = append(c.clientEventListeners, f)
+	}
+}
+
+// Subscriptions returns a slice of subscriptions
+func (c *Channel) Subscriptions() []*subscription.Subscription {
+	c.RLock()
+	defer c.RUnlock()
+
+	var subscriptions []*subscription.Subscription
+
+	for _, sub := range c.subscriptions {
+		subscriptions = append(subscriptions, sub)
+	}
+
+	return subscriptions
+}
+
+// IsOccupied Return true if the Channel has at least one subscriber
+func (c *Channel) IsOccupied() bool {
+	return c.TotalSubscriptions() > 0
+}
+
+// IsPresenceOrPrivate Check if the type of the Channel is presence or is private
+func (c *Channel) IsPresenceOrPrivate() bool {
+	return c.IsPresence() || c.IsPrivate()
+}
+
+// IsPublic Check if the type of the Channel is public
+func (c *Channel) IsPublic() bool {
+	return !c.IsPresenceOrPrivate()
+}
+
+// IsPresence Check if the type of the Channel is presence
+func (c *Channel) IsPresence() bool {
+	return utils.IsPresenceChannel(c.ID)
+}
+
+// IsPrivate Check if the type of the Channel is private
+func (c *Channel) IsPrivate() bool {
+	return utils.IsPrivateChannel(c.ID)
+}
+
+// TotalSubscriptions Get the total of subscribers
+func (c *Channel) TotalSubscriptions() int {
+	c.RLock()
+	defer c.RUnlock()
+
+	return len(c.subscriptions)
+}
+
+// TotalUsers Get the total of users.
+func (c *Channel) TotalUsers() int {
+	c.RLock()
+	defer c.RUnlock()
+
+	total := make(map[string]int)
+
+	for _, s := range c.subscriptions {
+		total[s.ID]++
+	}
+
+	return len(total)
+}
+
+// Subscribe Add a new subscriber to the Channel
+func (c *Channel) Subscribe(conn *connection.Connection, channelData string) error {
+	log.Infof("Subscribing %s to Channel %s", conn.SocketID, c.ID)
+
+	_subscription := subscription.New(conn, channelData)
+	c.Lock()
+	c.subscriptions[conn.SocketID] = _subscription
+	c.Unlock()
+
+	if c.IsPresence() {
+		// User Info Data
+		var info struct {
+			UserID   string          `json:"user_id"`
+			UserInfo json.RawMessage `json:"user_info"`
+		}
+
+		log.Infof("%+v", channelData)
+
+		if err := json.Unmarshal([]byte(channelData), &info); err != nil {
+			log.Error(err)
+			return err
+		}
+
+		js, err := info.UserInfo.MarshalJSON()
+
+		if err != nil {
+			log.Error(err)
+			return err
+		}
+
+		c.Lock()
+		// Update the Subscription
+		_subscription.ID = info.UserID
+		_subscription.Data = string(js)
+		c.Unlock()
+
+		// Publish pusher_internal:member_added
+		c.PublishMemberAddedEvent(channelData, _subscription)
+
+		for _, hook := range c.memberAddedListeners {
+			hook(c, _subscription)
+		}
+
+		// pusher_internal:subscription_succeeded
+		data := make(map[string]events.SubscriptionSucceededPresenceData)
+		data["presence"] = events.NewSubscriptionSucceedPresenceData(c.subscriptions)
+
+		js, err = json.Marshal(data)
+
+		if err != nil {
+			log.Error(err)
+			return err
+		}
+
+		conn.Publish(events.NewSubscriptionSucceeded(c.ID, string(js)))
+	} else {
+		conn.Publish(events.NewSubscriptionSucceeded(c.ID, "{}"))
+	}
+
+	if c.TotalSubscriptions() == 1 {
+		for _, hook := range c.channelOccupiedListeners {
+			hook(c, _subscription)
+		}
+	}
+
+	return nil
+}
+
+// IsSubscribed check if the user is subscribed
+func (c *Channel) IsSubscribed(conn *connection.Connection) bool {
+	c.RLock()
+	defer c.RUnlock()
+
+	_, exists := c.subscriptions[conn.SocketID]
+	return exists
+}
+
+// Unsubscribe Remove the subscriber from the Channel
+// It destroy the Channel if the channels does not have any subscribers.
+func (c *Channel) Unsubscribe(conn *connection.Connection) error {
+	log.Infof("unsubscribe %s from Channel %s", conn.SocketID, c.ID)
+
+	c.RLock()
+	_subscription, exists := c.subscriptions[conn.SocketID]
+	c.RUnlock()
+
+	if !exists {
+		return errors.New("_subscription not found")
+	}
+
+	c.Lock()
+	delete(c.subscriptions, conn.SocketID)
+	c.Unlock()
+
+	if c.IsPresence() {
+		// Publish pusher_internal:member_removed
+		c.PublishMemberRemovedEvent(_subscription)
+
+		for _, hook := range c.memberRemovedListeners {
+			hook(c, _subscription)
+		}
+	}
+
+	if !c.IsOccupied() {
+		for _, hook := range c.channelVacatedListeners {
+			hook(c, _subscription)
+		}
+	}
+
+	return nil
+}
+
+// PublishMemberAddedEvent Publish a MemberAddedEvent to all subscriptions
+func (c *Channel) PublishMemberAddedEvent(data string, subscription *subscription.Subscription) {
+	c.RLock()
+	defer c.RUnlock()
+
+	for _, subs := range c.subscriptions {
+		if subs != subscription {
+			subs.Connection.Publish(events.NewMemberAdded(c.ID, data))
+		}
+	}
+}
+
+// PublishMemberRemovedEvent Publish a MemberRemovedEvent to all subscriptions
+func (c *Channel) PublishMemberRemovedEvent(subscription *subscription.Subscription) {
+	c.RLock()
+	defer c.RUnlock()
+
+	for _, subs := range c.subscriptions {
+		if subs != subscription {
+			subs.Connection.Publish(events.NewMemberRemoved(c.ID, subscription.ID))
+		}
+	}
+}
+
+// Publish messages to all Subscribers
+// skip the ignore connection
+func (c *Channel) Publish(event events.Raw, ignore string) error {
+	c.RLock()
+	defer c.RUnlock()
+
+	b, err := event.Data.MarshalJSON()
+
+	if err != nil {
+		return err
+	}
+
+	var v interface{}
+
+	if err := json.Unmarshal(b, &v); err != nil {
+		return err
+	}
+
+	log.Infof("Publishing message %+v to Channel %s", v, c.ID)
+
+	for _, subs := range c.subscriptions {
+		if subs.Connection.SocketID != ignore {
+			subs.Connection.Publish(events.NewResponse(event.Event, event.Channel, v))
+		} else {
+			if utils.IsClientEvent(event.Event) {
+				for _, hook := range c.clientEventListeners {
+					hook(c, subs, event.Event, v)
+				}
+			}
+		}
+	}
+
+	return nil
+}

channel/channel_test.go

@@ -2,18 +2,23 @@
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
 
-package ipe
+package channel
 
-import "testing"
+import (
+	"ipe/connection"
+	"ipe/mocks"
+	"ipe/subscription"
+	"testing"
+)
 
 func TestIsOccupied(t *testing.T) {
-	c := newChannel("ID")
+	c := New("ID")
 
 	if c.IsOccupied() {
 		t.Errorf("c.IsOccupied() == %t, wants %t", c.IsOccupied(), false)
 	}
 
-	c.Subscriptions["ID"] = newSubscription(newConnection("ID", mockSocket{}), "")
+	c.subscriptions["ID"] = subscription.New(connection.New("ID", mocks.MockSocket{}), "")
 
 	if !c.IsOccupied() {
 		t.Errorf("c.IsOccupied() == %t, wants %t", c.IsOccupied(), true)
@@ -21,7 +26,7 @@ func TestIsOccupied(t *testing.T) {
 }
 
 func TestIsPrivate(t *testing.T) {
-	c := newChannel("private-channel")
+	c := New("private-Channel")
 
 	if !c.IsPrivate() {
 		t.Errorf("c.IsPrivate() == %t, wants %t", c.IsPrivate(), true)
@@ -29,7 +34,7 @@ func TestIsPrivate(t *testing.T) {
 }
 
 func TestIsPresence(t *testing.T) {
-	c := newChannel("presence-channel")
+	c := New("presence-Channel")
 
 	if !c.IsPresence() {
 		t.Errorf("c.IsPresence() == %t, wants %t", c.IsPresence(), true)
@@ -37,7 +42,7 @@ func TestIsPresence(t *testing.T) {
 }
 
 func TestIsPublic(t *testing.T) {
-	c := newChannel("channel")
+	c := New("Channel")
 
 	if !c.IsPublic() {
 		t.Errorf("c.IsPublic() == %t, wants %t", c.IsPublic(), true)
@@ -45,13 +50,13 @@ func TestIsPublic(t *testing.T) {
 }
 
 func TestIsPrivateOrPresence(t *testing.T) {
-	c := newChannel("private-channel")
+	c := New("private-Channel")
 
 	if !c.IsPresenceOrPrivate() {
 		t.Errorf("c.IsPresenceOrPrivate() == %t, wants %t", c.IsPresenceOrPrivate(), true)
 	}
 
-	c = newChannel("presence-channel")
+	c = New("presence-Channel")
 
 	if !c.IsPresenceOrPrivate() {
 		t.Errorf("c.IsPresenceOrPrivate() == %t, wants %t", c.IsPresenceOrPrivate(), true)
@@ -59,21 +64,21 @@ func TestIsPrivateOrPresence(t *testing.T) {
 }
 
 func TestTotalSubscriptions(t *testing.T) {
-	c := newChannel("ID")
+	c := New("ID")
 
-	if c.TotalSubscriptions() != len(c.Subscriptions) {
+	if c.TotalSubscriptions() != len(c.subscriptions) {
-		t.Errorf("c.TotalSubscriptions() == %d, wants %d", c.TotalSubscriptions(), len(c.Subscriptions))
+		t.Errorf("c.TotalSubscriptions() == %d, wants %d", c.TotalSubscriptions(), len(c.subscriptions))
 	}
 }
 
 func TestTotalUsers(t *testing.T) {
-	c := newChannel("ID")
+	c := New("ID")
 
-	c.Subscriptions["1"] = newSubscription(newConnection("ID", mockSocket{}), "")
+	c.subscriptions["1"] = subscription.New(connection.New("ID", mocks.MockSocket{}), "")
-	c.Subscriptions["2"] = newSubscription(newConnection("ID", mockSocket{}), "")
+	c.subscriptions["2"] = subscription.New(connection.New("ID", mocks.MockSocket{}), "")
 
-	if c.TotalSubscriptions() != len(c.Subscriptions) {
+	if c.TotalSubscriptions() != len(c.subscriptions) {
-		t.Errorf("c.TotalSubscriptions() == %d, wants %d", c.TotalSubscriptions(), len(c.Subscriptions))
+		t.Errorf("c.TotalSubscriptions() == %d, wants %d", c.TotalSubscriptions(), len(c.subscriptions))
 	}
 
 	if c.TotalUsers() != 1 {
@@ -83,14 +88,14 @@ func TestTotalUsers(t *testing.T) {
 }
 
 func TestIsSubscribed(t *testing.T) {
-	c := newChannel("ID")
+	c := New("ID")
-	conn := newConnection("ID", mockSocket{})
+	conn := connection.New("ID", mocks.MockSocket{})
 
 	if c.IsSubscribed(conn) {
 		t.Errorf("c.IsSubscribed(%q) == %t, wants %t", conn, c.IsSubscribed(conn), false)
 	}
 
-	c.Subscriptions["ID"] = newSubscription(conn, "")
+	c.subscriptions["ID"] = subscription.New(conn, "")
 
 	if !c.IsSubscribed(conn) {
 		t.Errorf("c.IsSubscribed(%q) == %t, wants %t", conn, c.IsSubscribed(conn), true)

cmd/main.go

@@ -8,20 +8,20 @@ import (
 	"flag"
 	"fmt"
 
-	"github.com/dimiro1/ipe/ipe"
+	"ipe"
 )
 
 // These variables are generated by the linker
 // please see the makefile for mor information.
 var (
-	version    string = "version"
+	version    = "version"
-	buildstamp string = "buildstamp"
+	buildstamp = "buildstamp"
-	githash    string = "githash"
+	githash    = "githash"
 )
 
 // Main function, initialize the system
 func main() {
-	var filename = flag.String("config", "config.json", "Config file location")
+	var filename = flag.String("config", "config.yml", "Config file location")
 	flag.Parse()
 
 	printBanner()

config/config.go

@@ -0,0 +1,39 @@
+// Copyright 2014, 2016 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 config
+
+// File config file
+type File struct {
+	Host      string        `yaml:"host"` // The host, eg: :8080 will start on 0.0.0.0:8080
+	SSL       SSL           `yaml:"ssl"`
+	Profiling bool          `yaml:"profiling"`
+	Apps      []Application `yaml:"apps"`
+}
+
+// SSL related configuration options
+type SSL struct {
+	Enabled  bool   `yaml:"enabled"`
+	Host     string `yaml:"host"`
+	KeyFile  string `yaml:"key_file"`
+	CertFile string `yaml:"cert_file"`
+}
+
+// Application related configuration options
+type Application struct {
+	Name       string   `yaml:"name"`
+	AppID      string   `yaml:"app_id"`
+	Key        string   `yaml:"key"`
+	Secret     string   `yaml:"secret"`
+	OnlySSL    bool     `yaml:"only_ssl"`
+	Enabled    bool     `yaml:"enabled"`
+	UserEvents bool     `yaml:"user_events"`
+	WebHooks   Webhooks `yaml:"webhooks"`
+}
+
+// Webhooks related configuration options
+type Webhooks struct {
+	Enabled bool   `yaml:"enabled"`
+	URL     string `yaml:"url"`
+}

connection/connection.go

@@ -0,0 +1,43 @@
+// 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 connection
+
+import (
+	"sync"
+	"time"
+
+	log "github.com/golang/glog"
+)
+
+// Socket interface to write to the client
+type Socket interface {
+	WriteJSON(interface{}) error
+}
+
+// Connection An user connection
+type Connection struct {
+	sync.Mutex
+
+	SocketID  string
+	Socket    Socket
+	CreatedAt time.Time
+}
+
+// New Create a new Subscriber
+func New(socketID string, s Socket) *Connection {
+	log.Infof("Creating a new Subscriber %+v", socketID)
+
+	return &Connection{SocketID: socketID, Socket: s, CreatedAt: time.Now()}
+}
+
+// Publish the message to websocket attached to this client
+func (conn *Connection) Publish(m interface{}) {
+	conn.Lock()
+	defer conn.Unlock()
+
+	if err := conn.Socket.WriteJSON(m); err != nil {
+		log.Errorf("error writing json into Socket, %+v", err)
+	}
+}

connection/connection_test.go

@@ -2,15 +2,18 @@
 // Use of this source code is governed by a MIT-style
 // license that can be found in the LICENSE file.
 
-package ipe
+package connection
 
-import "testing"
+import (
+	"ipe/mocks"
+	"testing"
+)
 
 func TestNewConnection(t *testing.T) {
 	expectedSocketID := "socketID"
-	expectedSocket := mockSocket{}
+	expectedSocket := mocks.MockSocket{}
 
-	c := newConnection(expectedSocketID, expectedSocket)
+	c := New(expectedSocketID, expectedSocket)
 
 	if c.SocketID != expectedSocketID {
 		t.Errorf("c.SocketID == %s, wants %s", c.SocketID, expectedSocketID)
@@ -21,6 +24,6 @@ func TestNewConnection(t *testing.T) {
 	}
 
 	if c.CreatedAt.IsZero() {
-		t.Errorf("c.CreatedAt.IsZero() == %t, wants %t", c.CreatedAt.IsZero(), false)
+		t.Errorf("c.createdAt.IsZero() == %t, wants %t", c.CreatedAt.IsZero(), false)
 	}
 }

entrypoint.sh

@@ -0,0 +1,6 @@
+#!/bin/sh
+if [ -z "$(ls -A /config)" ]; then
+   cp /app/config-example.yml /config/config.yml
+fi
+
+/app/ipe --config=/config/config.yml

events/events.go

@@ -0,0 +1,289 @@
+// 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 events
+
+import (
+	"encoding/json"
+
+	log "github.com/golang/glog"
+
+	"ipe/subscription"
+)
+
+// SubscribeData data for Subscribe event
+type SubscribeData struct {
+	Channel     string `json:"channel"`
+	Auth        string `json:"auth,omitempty"`
+	ChannelData string `json:"channel_data,omitempty"`
+}
+
+// Subscribe event
+// {
+//     "event": "pusher:subscribe",
+//     "data": {
+//         "channel": "the channel",
+//         "auth": "the auth",
+//         "channelData": "extra data"
+//     }
+// }
+type Subscribe struct {
+	Event string        `json:"event"`
+	Data  SubscribeData `json:"data"`
+}
+
+// NewSubscribe Create a new subscribe event with the specified channel and data
+func NewSubscribe(channel, auth, channelData string) Subscribe {
+	data := SubscribeData{Channel: channel, Auth: auth, ChannelData: channelData}
+	return Subscribe{Event: "pusher:subscribe", Data: data}
+}
+
+// UnsubscribeData event data
+type UnsubscribeData struct {
+	Channel string `json:"channel"`
+}
+
+// Unsubscribe event
+// {
+//     "event": "pusher:unsubscribe",
+//     "data": {
+//         "channel": "The channel"
+//     }
+// }
+type Unsubscribe struct {
+	Event string          `json:"event"`
+	Data  UnsubscribeData `json:"data"`
+}
+
+// NewUnsubscribe Create a new unsubscribe event for the specified channel
+func NewUnsubscribe(channel string) Unsubscribe {
+	data := UnsubscribeData{Channel: channel}
+	return Unsubscribe{Event: "pusher:unsubscribe", Data: data}
+}
+
+// SubscriptionSucceeded event
+// {
+//     "event": "pusher_internal:subscription_succeeded",
+//     "channel": "the channel"
+// }
+type SubscriptionSucceeded struct {
+	Event   string `json:"event"`
+	Channel string `json:"channel"`
+	Data    string `json:"data"`
+}
+
+// NewSubscriptionSucceeded Create a new subscription succeed event for the specified channel
+func NewSubscriptionSucceeded(channel, data string) SubscriptionSucceeded {
+	return SubscriptionSucceeded{Event: "pusher_internal:subscription_succeeded", Channel: channel, Data: data}
+}
+
+// SubscriptionSucceededPresenceData Data Subscription Succeed
+// "{
+//     \"presence\": {
+//        \"ids\": [\"11814b369700141b222a3f3791cec2d9\",\"71dd6a29da2a4833336d2a964becf820\"],
+//        \"hash\": {
+//           \"11814b369700141b222a3f3791cec2d9\": {
+//              \"name\":\"Phil Leggetter\",
+//              \"twitter\": \"@leggetter\"
+//           },
+//           \"71dd6a29da2a4833336d2a964becf820\": {
+//              \"name\":\"Max Williams\",
+//              \"twitter\": \"@maxthelion\"
+//           }
+//        },
+//        \"count\": 2
+//     }
+// }"
+type SubscriptionSucceededPresenceData struct {
+	Ids   []string               `json:"ids"`
+	Hash  map[string]interface{} `json:"hash"`
+	Count int                    `json:"count"`
+}
+
+// NewSubscriptionSucceedPresenceData returns new SubscriptionSucceededPresenceData
+func NewSubscriptionSucceedPresenceData(subscriptions map[string]*subscription.Subscription) SubscriptionSucceededPresenceData {
+	event := SubscriptionSucceededPresenceData{}
+
+	var (
+		ids  []string
+		hash = make(map[string]interface{}, len(subscriptions))
+	)
+
+	for _, s := range subscriptions {
+		// Do you have any other idea?
+		var js interface{}
+		if err := json.Unmarshal([]byte(s.Data), &js); err != nil {
+			continue
+		}
+
+		hash[s.ID] = js
+		ids = append(ids, s.ID)
+	}
+
+	event.Ids = ids
+	event.Hash = hash
+	event.Count = len(subscriptions)
+
+	return event
+}
+
+// Pong event
+// {
+//     "event": "pusher:pong",
+//     "data": {}
+// }
+type Pong struct {
+	Event string `json:"event"`
+	Data  string `json:"data"`
+}
+
+// NewPong Create a new pong event
+func NewPong() Pong {
+	return Pong{Event: "pusher:pong", Data: "{}"}
+}
+
+// Ping event
+// {
+//     "event": "pusher:ping",
+//     "data": {}
+// }
+type Ping struct {
+	Event string `json:"event"`
+	Data  string `json:"data"`
+}
+
+// NewPing Create a new ping event
+func NewPing() Ping {
+	return Ping{Event: "pusher:ping", Data: "{}"}
+}
+
+// Error event
+// {
+//     "event": "pusher:error",
+//     "data": {
+//         "message": "A Message",
+//         "code": 4000
+//     }
+// }
+type Error struct {
+	Event string      `json:"event"`
+	Data  interface{} `json:"data"`
+}
+
+// NewError Create a new error event
+// Pusher protocol is very strange in some parts
+// It send null in some errors.
+func NewError(code int, message string) Error {
+	var data = struct {
+		Code    *int   `json:"code"`
+		Message string `json:"message"`
+	}{
+		Message: message,
+	}
+
+	if code == 0 {
+		data.Code = nil
+	} else {
+		data.Code = &code
+	}
+
+	return Error{Event: "pusher:error", Data: data}
+}
+
+// ConnectionEstablished event
+// {
+//     "event" : "pusher:connection_established",
+//     "data" : {
+//       "socket_id" : "123456",
+//       "activity_timeout" : 120
+//     }
+// }
+type ConnectionEstablished struct {
+	Event string `json:"event"`
+	Data  string `json:"data"`
+}
+
+// NewConnectionEstablished Create a new connection established event using the specified socketId
+func NewConnectionEstablished(socketID string) ConnectionEstablished {
+	b, err := json.Marshal(struct {
+		SocketID        string `json:"socket_id"`
+		ActivityTimeout int    `json:"activity_timeout"`
+	}{
+		SocketID: socketID, ActivityTimeout: 120,
+	})
+
+	if err != nil {
+		panic("events: Could not Marshal json ConnectionEstablishedEvent")
+	}
+
+	return ConnectionEstablished{Event: "pusher:connection_established", Data: string(b)}
+}
+
+// MemberAdded event
+// {
+//   "event": "pusher_internal:member_added",
+//   "channel": "presence-example-channel",
+//   "data": String
+// }
+type MemberAdded struct {
+	Event   string `json:"event"`
+	Channel string `json:"channel"`
+	Data    string `json:"data"`
+}
+
+// NewMemberAdded creates a new MemberAdded event
+func NewMemberAdded(channel, data string) MemberAdded {
+	return MemberAdded{Event: "pusher_internal:member_added", Channel: channel, Data: data}
+}
+
+// MemberRemoved event
+// {
+//   "event": "pusher_internal:member_removed",
+//   "channel": "presence-example-channel",
+//   "data": String
+// }
+type MemberRemoved struct {
+	Event   string `json:"event"`
+	Channel string `json:"channel"`
+	Data    string `json:"data"`
+}
+
+// NewMemberRemoved returns a new MemberRemoved event
+func NewMemberRemoved(channel string, userID string) MemberRemoved {
+	data, err := json.Marshal(struct {
+		UserID string `json:"user_id"`
+	}{
+		UserID: userID,
+	})
+
+	if err != nil {
+		log.Error(err)
+	}
+
+	return MemberRemoved{Event: "pusher_internal:member_removed", Channel: channel, Data: string(data)}
+}
+
+// Raw event, usually used for client events
+// {
+//     "event": "client-?",
+//     "channel": "The channel",
+//     "data": {}
+// }
+type Raw struct {
+	Event   string          `json:"event"`
+	Channel string          `json:"channel"`
+	Data    json.RawMessage `json:"data"`
+}
+
+// Response event
+type Response struct {
+	Event   string      `json:"event"`
+	Channel string      `json:"channel"`
+	Data    interface{} `json:"data"`
+}
+
+// NewResponse The response event that is broadcasted to the client sockets
+func NewResponse(name, channel string, data interface{}) Response {
+	return Response{Event: name, Channel: channel, Data: data}
+}

events/events_test.go

@@ -0,0 +1,29 @@
+package events
+
+import (
+	"bytes"
+	"encoding/json"
+	"testing"
+)
+
+func Test_newErrorEvent_with_invalid_code(t *testing.T) {
+	event := NewError(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 := NewError(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)
+	}
+}

functional/Procfile

@@ -1,2 +1,2 @@
 client: go run client.go
-server: go run ../main.go -config ./functional-config.json -alsologtostderr
+server: go run ../cmd/main.go -config ./functional-config.yml -alsologtostderr

functional/SpecRunner.html

@@ -1,24 +1,23 @@
 <html>
 <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="//cdnjs.cloudflare.com/ajax/libs/mocha/5.2.0/mocha.min.css" rel="stylesheet"/>
 </head>
 <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="//cdnjs.cloudflare.com/ajax/libs/mocha/5.2.0/mocha.min.js"></script>
-  <script src="https://cdnjs.cloudflare.com/ajax/libs/mocha/2.3.4/mocha.min.js"></script>
+<script src="//cdnjs.cloudflare.com/ajax/libs/chai/4.2.0/chai.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/pusher/3.0.0/pusher.js"></script>
+<script src="//js.pusher.com/4.3.1/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.globals(['jQuery', 'Pusher']);
+    mocha.globals(['Pusher']);
     mocha.run();
-  </script>
+</script>
 </body>
 </html>

functional/client.go

@@ -1,9 +1,12 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"log"
 	"net/http"
+	"net/http/httputil"
 
 	"github.com/pusher/pusher-http-go"
 )
@@ -20,6 +23,7 @@ func init() {
 }
 
 func pusherPresenceAuth(res http.ResponseWriter, req *http.Request) {
+	log.Println("Presence Request")
 	presenceData := pusher.MemberData{
 		UserId:   "1",
 		UserInfo: map[string]string{},
@@ -32,30 +36,63 @@ func pusherPresenceAuth(res http.ResponseWriter, req *http.Request) {
 		panic(err)
 	}
 
-	fmt.Fprintf(res, string(response))
+	_, _ = fmt.Fprint(res, string(response))
 }
 
 func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) {
 	params, _ := ioutil.ReadAll(req.Body)
 	response, err := client.AuthenticatePrivateChannel(params)
 
+	log.Printf("Private Request %s", params)
+	log.Printf("Auth %s", response)
+
 	if err != nil {
 		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")
+	_, err := client.Trigger("private-messages", "messages", "The message from server")
+	if err != nil {