diff --git a/Rakefile b/Rakefile deleted file mode 100644 index f8846b8..0000000 --- a/Rakefile +++ /dev/null @@ -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.3.0-SNAPSHOT' -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 dependencies' -task :'deps' do - sh 'glide install -v -s' -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 \ No newline at end of file diff --git a/ipe/handlers.go b/api/handlers.go similarity index 72% rename from ipe/handlers.go rename to api/handlers.go index a80f539..8992a59 100644 --- a/ipe/handlers.go +++ b/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,17 @@ import ( "strings" log "github.com/golang/glog" - "github.com/pressly/chi" + "github.com/gorilla/mux" - "github.com/dimiro1/ipe/utils" + "ipe/events" + "ipe/storage" + "ipe/utils" ) +// // 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 @@ -47,12 +53,15 @@ 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 authenticationHandler(DB db) func(http.Handler) http.Handler { +func Authentication(storage storage.Storage) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { fn := func(w http.ResponseWriter, r *http.Request) { - appID := chi.URLParam(r, "app_id") + var ( + pathVars = mux.Vars(r) + appID = pathVars["app_id"] + ) - app, err := DB.GetAppByAppID(appID) + app, err := storage.GetAppByAppID(appID) if err != nil { log.Error(err) @@ -82,12 +91,15 @@ func authenticationHandler(DB db) func(http.Handler) http.Handler { } // Check if the application is disabled -func checkAppDisabled(DB db) func(http.Handler) http.Handler { +func CheckAppDisabled(storage storage.Storage) func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { fn := func(w http.ResponseWriter, r *http.Request) { - appID := chi.URLParam(r, "app_id") + var ( + pathVars = mux.Vars(r) + appID = pathVars["app_id"] + ) - currentApp, err := DB.GetAppByAppID(appID) + currentApp, err := storage.GetAppByAppID(appID) if err != nil { http.Error(w, fmt.Sprintf("Could not found an app with app_id: %s", appID), http.StatusForbidden) @@ -105,7 +117,11 @@ func checkAppDisabled(DB db) func(http.Handler) http.Handler { } } -type postEventsHandler struct{ DB db } +type PostEvents struct{ storage storage.Storage } + +func NewPostEvents(storage storage.Storage) *PostEvents { + return &PostEvents{storage: storage} +} // ServeHTTPC 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. @@ -122,10 +138,13 @@ type postEventsHandler struct{ DB db } // Response is an empty JSON hash. // // POST /apps/{app_id}/events -func (h *postEventsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - appID := chi.URLParam(r, "app_id") +func (h *PostEvents) ServeHTTP(w http.ResponseWriter, r *http.Request) { + var ( + pathVars = mux.Vars(r) + appID = pathVars["app_id"] + ) - app, err := h.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) @@ -160,15 +179,24 @@ func (h *postEventsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { 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) + } } -type getChannelsHandler struct{ DB db } +type GetChannels struct{ storage storage.Storage } + +func NewGetChannels(storage storage.Storage) *GetChannels { + return &GetChannels{storage: storage} +} // Allows fetching a hash of occupied channels (optionally filtered by prefix), // and optionally one or more attributes for each channel. @@ -189,14 +217,15 @@ type getChannelsHandler struct{ DB db } // } // // GET /apps/{app_id}/channels -func (h *getChannelsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - query := r.URL.Query() - - appID := chi.URLParam(r, "app_id") - filter := query.Get("filter_by_prefix") - info := query.Get("info") - - attributes := strings.Split(info, ",") +func (h *GetChannels) ServeHTTP(w http.ResponseWriter, r *http.Request) { + var ( + pathVars = mux.Vars(r) + queryVars = r.URL.Query() + appID = pathVars["app_id"] + filter = queryVars.Get("filter_by_prefix") + info = queryVars.Get("info") + attributes = strings.Split(info, ",") + ) requestedUserCount := false @@ -213,7 +242,7 @@ func (h *getChannelsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - app, err := h.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) @@ -225,30 +254,30 @@ func (h *getChannelsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { 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 { - channels[c.ChannelID] = struct{}{} + for _, c := range app.Channels() { + 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 @@ -259,7 +288,11 @@ func (h *getChannelsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } -type getChannelHandler struct{ DB db } +type GetChannel struct{ storage storage.Storage } + +func NewGetChannel(storage storage.Storage) *GetChannel { + return &GetChannel{storage: storage} +} // Fetch info for one channel // @@ -271,29 +304,28 @@ type getChannelHandler struct{ DB db } // } // // GET /apps/{app_id}/channels/{channel_name} -func (h *getChannelHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json;charset=UTF-8") +func (h *GetChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) { + 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() - - appID := chi.URLParam(r, "app_id") - app, err := h.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) } - channelName := chi.URLParam(r, "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 @@ -341,15 +373,18 @@ func (h *getChannelHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { 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) } } -type getChannelUsersHandler struct{ DB db } +type GetChannelUsers struct{ storage storage.Storage } + +func NewGetChannelUsers(storage storage.Storage) *GetChannelUsers { + return &GetChannelUsers{storage: storage} +} // Allowed only for presence-channels // @@ -362,9 +397,12 @@ type getChannelUsersHandler struct{ DB db } // } // // GET /apps/{app_id}/channels/{channel_name}/users -func (h *getChannelUsersHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - appID := chi.URLParam(r, "app_id") - channelName := chi.URLParam(r, "channel_name") +func (h *GetChannelUsers) ServeHTTP(w http.ResponseWriter, r *http.Request) { + var ( + pathVars = mux.Vars(r) + appID = pathVars["app_id"] + channelName = pathVars["channel_name"] + ) isPresence := utils.IsPresenceChannel(channelName) @@ -373,7 +411,7 @@ func (h *getChannelUsersHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques return } - app, err := h.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) @@ -392,7 +430,7 @@ func (h *getChannelUsersHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques var users []interface{} - for _, s := range channel.Subscriptions { + for _, s := range channel.Subscriptions() { users = append(users, struct { ID string `json:"id"` }{s.ID}) @@ -400,8 +438,7 @@ func (h *getChannelUsersHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques 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) diff --git a/ipe/handlers_test.go b/api/handlers_test.go similarity index 67% rename from ipe/handlers_test.go rename to api/handlers_test.go index 971c941..ae1b88c 100644 --- a/ipe/handlers_test.go +++ b/api/handlers_test.go @@ -1,53 +1,66 @@ -package ipe +package api import ( - "context" "encoding/json" "fmt" "net/http" "net/http/httptest" + "strconv" "testing" - "github.com/pressly/chi" + "github.com/gorilla/mux" + + "ipe/app" + channel2 "ipe/channel" + "ipe/connection" + "ipe/mocks" + "ipe/storage" ) var ( - testApp *app - database db + testApp *app.Application + 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(newChannel("private-c3")) + testApp.AddChannel(channel2.New("c2")) + testApp.AddChannel(channel2.New("private-c3")) - conn := newConnection("123.456", mockSocket{}) - testApp.Subscribe(channel, conn, "{}") + conn := connection.New("123.456", mocks.MockSocket{}) + _ = testApp.Subscribe(channel, conn, "{}") - conn = newConnection("321.654", mockSocket{}) - testApp.Subscribe(channel, conn, "{}") + conn = connection.New("321.654", mocks.MockSocket{}) + _ = testApp.Subscribe(channel, conn, "{}") - db := newMemdb() - db.AddApp(testApp) + _storage := storage.NewInMemory() + _ = _storage.AddApp(testApp) - database = db + database = _storage } -// All Channels +// All channels func Test_getChannels_all(t *testing.T) { appID := testApp.AppID - rctx := chi.NewRouteContext() - rctx.URLParams.Add("app_id", appID) - r, _ := http.NewRequest("GET", fmt.Sprintf("/apps/%s/channels", appID), nil) - r = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx)) + r = mux.SetURLVars(r, map[string]string{ + "app_id": appID, + }) w := httptest.NewRecorder() - handler := &getChannelsHandler{database} + handler := &GetChannels{database} handler.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -55,7 +68,7 @@ func Test_getChannels_all(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{}) @@ -67,14 +80,13 @@ func Test_getChannels_all(t *testing.T) { func Test_getChannels_filter_by_presence_prefix(t *testing.T) { appID := testApp.AppID - rctx := chi.NewRouteContext() - rctx.URLParams.Add("app_id", appID) - 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)) + r = mux.SetURLVars(r, map[string]string{ + "app_id": appID, + }) w := httptest.NewRecorder() - handler := &getChannelsHandler{database} + handler := &GetChannels{database} handler.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -82,7 +94,7 @@ func Test_getChannels_filter_by_presence_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{}) @@ -95,14 +107,13 @@ func Test_getChannels_filter_by_presence_prefix(t *testing.T) { func Test_getChannels_filter_by_presence_prefix_and_user_count(t *testing.T) { appID := testApp.AppID - rctx := chi.NewRouteContext() - 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 = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx)) + r = mux.SetURLVars(r, map[string]string{ + "app_id": appID, + }) w := httptest.NewRecorder() - handler := &getChannelsHandler{database} + handler := &GetChannels{database} handler.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -110,7 +121,7 @@ func Test_getChannels_filter_by_presence_prefix_and_user_count(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{}) @@ -135,14 +146,13 @@ 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) { appID := testApp.AppID - rctx := chi.NewRouteContext() - 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 = r.WithContext(context.WithValue(context.Background(), chi.RouteCtxKey, rctx)) + r = mux.SetURLVars(r, map[string]string{ + "app_id": appID, + }) w := httptest.NewRecorder() - handler := &getChannelsHandler{database} + handler := &GetChannels{database} handler.ServeHTTP(w, r) if w.Code != http.StatusBadRequest { @@ -153,14 +163,13 @@ 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 - rctx := chi.NewRouteContext() - rctx.URLParams.Add("app_id", appID) - 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)) + r = mux.SetURLVars(r, map[string]string{ + "app_id": appID, + }) w := httptest.NewRecorder() - handler := &getChannelsHandler{database} + handler := &GetChannels{database} handler.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -169,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{}) @@ -187,14 +196,13 @@ func Test_getChannels_filter_by_public_prefix(t *testing.T) { func Test_getChannels_filter_by_private_prefix(t *testing.T) { appID := testApp.AppID - rctx := chi.NewRouteContext() - rctx.URLParams.Add("app_id", appID) - 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)) + r = mux.SetURLVars(r, map[string]string{ + "app_id": appID, + }) w := httptest.NewRecorder() - handler := &getChannelsHandler{database} + handler := &GetChannels{database} handler.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -203,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{}) diff --git a/app/app.go b/app/app.go new file mode 100644 index 0000000..4b67e48 --- /dev/null +++ b/app/app.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 app + +import ( + "errors" + "expvar" + "fmt" + "sync" + + log "github.com/golang/glog" + + "ipe/channel" + "ipe/connection" + "ipe/events" + "ipe/subscription" +) + +// An App +type Application struct { + sync.Mutex + + Name string + AppID string + Key string + Secret string + OnlySSL bool + ApplicationDisabled bool + UserEvents bool + WebHooks bool + URLWebHook string + + channels map[string]*channel.Channel `json:"-"` + connections map[string]*connection.Connection `json:"-"` + + Stats *expvar.Map `json:"-"` +} + +func NewApplication( + name, + appID, + key, + secret string, + onlySSL, + disabled, + userEvents, + webHooks bool, + webHookURL string, +) *Application { + + a := &Application{ + Name: name, + AppID: appID, + Key: key, + Secret: secret, + OnlySSL: onlySSL, + ApplicationDisabled: disabled, + 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 { + var channels []*channel.Channel + + for _, c := range a.channels { + channels = append(channels, c) + } + + return channels +} + +// Only Presence channels +func (a *Application) PresenceChannels() []*channel.Channel { + var channels []*channel.Channel + + for _, c := range a.channels { + if c.IsPresence() { + channels = append(channels, c) + } + } + + return channels +} + +// Only Private channels +func (a *Application) PrivateChannels() []*channel.Channel { + var channels []*channel.Channel + + for _, c := range a.channels { + if c.IsPrivate() { + channels = append(channels, c) + } + } + + return channels +} + +// Only Public channels +func (a *Application) PublicChannels() []*channel.Channel { + 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() + defer a.Unlock() + + _, exists := a.connections[conn.SocketID] + + if !exists { + return + } + + delete(a.connections, conn.SocketID) + + 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) +} + +// Find a Connection on this Application +func (a *Application) FindConnection(socketID string) (*connection.Connection, error) { + conn, exists := a.connections[socketID] + + if exists { + return conn, nil + } + + return nil, errors.New("connection not found") +} + +// DeleteChannel 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) +} + +// 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) +} + +// 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 +} + +// Find the Channel by Channel ID +func (a *Application) FindChannelByChannelID(n string) (*channel.Channel, error) { + c, exists := a.channels[n] + + if exists { + return c, nil + } + + return nil, errors.New("channel does not exists") +} + +func (a *Application) Publish(c *channel.Channel, event events.Raw, ignore string) error { + a.Stats.Add("TotalUniqueMessages", 1) + + return c.Publish(event, ignore) +} + +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 +} + +func (a *Application) Subscribe(c *channel.Channel, conn *connection.Connection, data string) error { + return c.Subscribe(conn, data) +} diff --git a/app/app_test.go b/app/app_test.go new file mode 100644 index 0000000..770c46c --- /dev/null +++ b/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") + } + +} diff --git a/ipe/webhooks.go b/app/webhooks.go similarity index 63% rename from ipe/webhooks.go rename to app/webhooks.go index e8d349e..d6c2145 100644 --- a/ipe/webhooks.go +++ b/app/webhooks.go @@ -2,24 +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" - "context" - "fmt" - - "github.com/dimiro1/ipe/utils" log "github.com/golang/glog" + + "ipe/channel" + "ipe/subscription" + "ipe/utils" ) 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: // { // "time_ms": 1327078148132 @@ -34,7 +36,7 @@ const maxTimeout = 3 * time.Second // 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. @@ -52,43 +54,48 @@ type hookEvent struct { UserID string `json:"user_id,omitempty"` } -func newChannelOcuppiedHook(channel *channel) hookEvent { - return hookEvent{Name: "channel_occupied", Channel: channel.ChannelID} +func newChannelOcuppiedHook(channel *channel.Channel) hookEvent { + return hookEvent{Name: "channel_occupied", Channel: channel.ID} } -func newChannelVacatedHook(channel *channel) hookEvent { - return hookEvent{Name: "channel_vacated", Channel: channel.ChannelID} +func newChannelVacatedHook(channel *channel.Channel) hookEvent { + return hookEvent{Name: "channel_vacated", Channel: channel.ID} } -func newMemberAddedHook(channel *channel, s *subscription) hookEvent { - return hookEvent{Name: "member_added", Channel: channel.ChannelID, UserID: s.ID} +func newMemberAddedHook(channel *channel.Channel, s *subscription.Subscription) hookEvent { + return hookEvent{Name: "member_added", Channel: channel.ID, UserID: s.ID} } -func newMemberRemovedHook(channel *channel, s *subscription) hookEvent { - return hookEvent{Name: "member_removed", Channel: channel.ChannelID, UserID: s.ID} +func newMemberRemovedHook(channel *channel.Channel, s *subscription.Subscription) hookEvent { + return hookEvent{Name: "member_removed", Channel: channel.ID, UserID: s.ID} } -func newClientHook(channel *channel, s *subscription, event string, data interface{}) hookEvent { - return hookEvent{Name: "client_event", Channel: channel.ChannelID, Event: event, Data: data, SocketID: s.Connection.SocketID} +func newClientHook(channel *channel.Channel, s *subscription.Subscription, event string, data interface{}) hookEvent { + return hookEvent{Name: "client_event", Channel: channel.ID, Event: event, Data: data, SocketID: s.Connection.SocketID} } // channel_occupied // { "name": "channel_occupied", "channel": "test_channel" } -func (a *app) TriggerChannelOccupiedHook(c *channel) { +func (a *Application) TriggerChannelOccupiedHook(c *channel.Channel) { event := newChannelOcuppiedHook(c) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) defer cancel() - triggerHook(ctx, a, event) + if err := triggerHook(ctx, a, event); err != nil { + log.Errorf("triggering webhook %+v", err) + } } // channel_vacated // { "name": "channel_vacated", "channel": "test_channel" } -func (a *app) TriggerChannelVacatedHook(c *channel) { +func (a *Application) TriggerChannelVacatedHook(c *channel.Channel) { event := newChannelVacatedHook(c) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) defer cancel() - triggerHook(ctx, a, event) + + if err := triggerHook(ctx, a, event); err != nil { + log.Errorf("triggering webhook %+v", err) + } } // { @@ -99,7 +106,7 @@ 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() { @@ -108,7 +115,10 @@ func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent st ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) defer cancel() - triggerHook(ctx, a, event) + + if err := triggerHook(ctx, a, event); err != nil { + log.Errorf("triggering webhook %+v", err) + } } // { @@ -116,11 +126,14 @@ func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent st // "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) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) defer cancel() - triggerHook(ctx, a, event) + + if err := triggerHook(ctx, a, event); err != nil { + log.Errorf("triggering webhook %+v", err) + } } // { @@ -128,21 +141,23 @@ func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) { // "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) ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) defer cancel() - triggerHook(ctx, a, event) + + if err := triggerHook(ctx, a, event); err != nil { + log.Errorf("triggering webhook %+v", err) + } } -func triggerHook(ctx context.Context, a *app, event hookEvent) error { +func triggerHook(ctx context.Context, a *Application, event hookEvent) error { if !a.WebHooks { - log.Infof("Webhooks are not enabled for app: %s", a.Name) - return fmt.Errorf("Webhooks are not enabled for app: %s", a.Name) + log.Infof("webhook are not enabled for app: %s", a.Name) + return fmt.Errorf("webhooks are not enabled for app: %s", a.Name) } done := make(chan bool) - defer close(done) go func() { log.Infof("Triggering %s event", event.Name) @@ -170,7 +185,7 @@ func triggerHook(ctx context.Context, a *app, event hookEvent) error { return } - req.WithContext(ctx) + req = req.WithContext(ctx) req.Header.Set("User-Agent", "Ipe UA; (+https://github.com/dimiro1/ipe)") req.Header.Set("Content-Type", "application/json") @@ -184,11 +199,15 @@ func triggerHook(ctx context.Context, a *app, event hookEvent) error { // 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", event.Name, err) + log.Errorf("error posting %s event: %+v", event.Name, err) } // Successfully terminated diff --git a/channel/channel.go b/channel/channel.go new file mode 100644 index 0000000..35c6af3 --- /dev/null +++ b/channel/channel.go @@ -0,0 +1,313 @@ +// 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" +) + +type Option func(*Channel) +type ListenerFunc func(*Channel, *subscription.Subscription) +type ClientEventListenerFunc func(*Channel, *subscription.Subscription, string, interface{}) + +// A 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 +} + +// 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 +} + +func WithMemberAddedListener(f ListenerFunc) func(*Channel) { + return func(c *Channel) { + c.memberAddedListeners = append(c.memberAddedListeners, f) + } +} + +func WithMemberRemovedListener(f ListenerFunc) func(*Channel) { + return func(c *Channel) { + c.memberRemovedListeners = append(c.memberRemovedListeners, f) + } +} + +func WithChannelOccupiedListener(f ListenerFunc) func(*Channel) { + return func(c *Channel) { + c.channelOccupiedListeners = append(c.channelOccupiedListeners, f) + } +} + +func WithChannelVacatedListener(f ListenerFunc) func(*Channel) { + return func(c *Channel) { + c.channelVacatedListeners = append(c.channelVacatedListeners, f) + } +} + +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 +} + +// Return true if the Channel has at least one subscriber +func (c *Channel) IsOccupied() bool { + return c.TotalSubscriptions() > 0 +} + +// Check if the type of the Channel is presence or is private +func (c *Channel) IsPresenceOrPrivate() bool { + return c.IsPresence() || c.IsPrivate() +} + +// Check if the type of the Channel is public +func (c *Channel) IsPublic() bool { + return !c.IsPresenceOrPrivate() +} + +// Check if the type of the Channel is presence +func (c *Channel) IsPresence() bool { + return utils.IsPresenceChannel(c.ID) +} + +// Check if the type of the Channel is private +func (c *Channel) IsPrivate() bool { + return utils.IsPrivateChannel(c.ID) +} + +// Get the total of subscribers +func (c *Channel) TotalSubscriptions() int { + c.RLock() + defer c.RUnlock() + + return len(c.subscriptions) +} + +// 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) +} + +// 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 +} + +// 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 +} + +// 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)) + } + } +} + +// 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 +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 +} diff --git a/ipe/channel_test.go b/channel/channel_test.go similarity index 66% rename from ipe/channel_test.go rename to channel/channel_test.go index 8eeb23c..3e9d0c8 100644 --- a/ipe/channel_test.go +++ b/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) { - t.Errorf("c.TotalSubscriptions() == %d, wants %d", c.TotalSubscriptions(), len(c.Subscriptions)) + if 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["2"] = newSubscription(newConnection("ID", mockSocket{}), "") + c.subscriptions["1"] = subscription.New(connection.New("ID", mocks.MockSocket{}), "") + c.subscriptions["2"] = subscription.New(connection.New("ID", mocks.MockSocket{}), "") - if c.TotalSubscriptions() != len(c.Subscriptions) { - t.Errorf("c.TotalSubscriptions() == %d, wants %d", c.TotalSubscriptions(), len(c.Subscriptions)) + if 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") - conn := newConnection("ID", mockSocket{}) + c := New("ID") + 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) diff --git a/main.go b/cmd/main.go similarity index 97% rename from main.go rename to cmd/main.go index 8afca43..6b10d75 100644 --- a/main.go +++ b/cmd/main.go @@ -8,7 +8,7 @@ import ( "flag" "fmt" - "github.com/dimiro1/ipe/ipe" + "ipe" ) // These variables are generated by the linker diff --git a/ipe/config-example.json b/config-example.json similarity index 100% rename from ipe/config-example.json rename to config-example.json diff --git a/ipe/config.go b/config/config.go similarity index 70% rename from ipe/config.go rename to config/config.go index ce413f9..822c429 100644 --- a/ipe/config.go +++ b/config/config.go @@ -2,10 +2,10 @@ // Use of this source code is governed by a MIT-style // license that can be found in the LICENSE file. -package ipe +package config // The config file -type configFile struct { +type File struct { Host string // The host, eg: :8080 will start on 0.0.0.0:8080 User string SSL bool @@ -14,10 +14,10 @@ type configFile struct { SSLKeyFile string SSLCertFile string - Apps []configApp + Apps []Application } -type configApp struct { +type Application struct { Name string AppID string Key string @@ -28,17 +28,3 @@ type configApp struct { WebHooks bool URLWebHook string } - -func newAppFromConfig(a configApp) *app { - return newApp( - a.Name, - a.AppID, - a.Key, - a.Secret, - a.OnlySSL, - a.ApplicationDisabled, - a.UserEvents, - a.WebHooks, - a.URLWebHook, - ) -} diff --git a/connection/connection.go b/connection/connection.go new file mode 100644 index 0000000..3dda056 --- /dev/null +++ b/connection/connection.go @@ -0,0 +1,37 @@ +// 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 ( + "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 { + SocketID string + Socket Socket + CreatedAt time.Time +} + +// 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{}) { + if err := conn.Socket.WriteJSON(m); err != nil { + log.Errorf("error writing json into Socket, %+v", err) + } +} diff --git a/ipe/connection_test.go b/connection/connection_test.go similarity index 59% rename from ipe/connection_test.go rename to connection/connection_test.go index d691ecc..9da3b37 100644 --- a/ipe/connection_test.go +++ b/connection/connection_test.go @@ -2,23 +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" - -// mockSocket is a mock implementation of socket -// used in the test suite -type mockSocket struct{} - -func (s mockSocket) WriteJSON(i interface{}) error { - return nil -} +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) @@ -29,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) } } diff --git a/ipe/events.go b/events/events.go similarity index 58% rename from ipe/events.go rename to events/events.go index 0cc3d25..08d404f 100644 --- a/ipe/events.go +++ b/events/events.go @@ -2,12 +2,14 @@ // Use of this source code is governed by a MIT-style // license that can be found in the LICENSE file. -package ipe +package events import ( "encoding/json" log "github.com/golang/glog" + + "ipe/subscription" ) // { @@ -18,24 +20,24 @@ import ( // "channelData": "extra data" // } // } -type subscribeEventData struct { +type SubscribeData struct { Channel string `json:"channel"` Auth string `json:"auth,omitempty"` ChannelData string `json:"channel_data,omitempty"` } -type subscribeEvent struct { - Event string `json:"event"` - Data subscribeEventData `json:"data"` +type Subscribe struct { + Event string `json:"event"` + Data SubscribeData `json:"data"` } // Create a new subscribe event with the specified channel and data -func newSubscribeEvent(channel, auth, channelData string) subscribeEvent { - data := subscribeEventData{Channel: channel, Auth: auth, ChannelData: channelData} - return subscribeEvent{Event: "pusher:subscribe", Data: data} +func NewSubscribe(channel, auth, channelData string) Subscribe { + data := SubscribeData{Channel: channel, Auth: auth, ChannelData: channelData} + return Subscribe{Event: "pusher:subscribe", Data: data} } -type unsubscribeEventData struct { +type UnsubscribeData struct { Channel string `json:"channel"` } @@ -45,30 +47,30 @@ type unsubscribeEventData struct { // "channel": "The channel" // } // } -type unsubscribeEvent struct { - Event string `json:"event"` - Data unsubscribeEventData `json:"data"` +type Unsubscribe struct { + Event string `json:"event"` + Data UnsubscribeData `json:"data"` } // Create a new unsubscribe event for the specified channel -func newUnsubscribeEvent(channel string) unsubscribeEvent { - data := unsubscribeEventData{Channel: channel} - return unsubscribeEvent{Event: "pusher:unsubscribe", Data: data} +func NewUnsubscribe(channel string) Unsubscribe { + data := UnsubscribeData{Channel: channel} + return Unsubscribe{Event: "pusher:unsubscribe", Data: data} } // { // "event": "pusher_internal:subscription_succeeded", // "channel": "the channel" // } -type subscriptionSucceededEvent struct { +type SubscriptionSucceeded struct { Event string `json:"event"` Channel string `json:"channel"` Data string `json:"data"` } // Create a new subscription succeed event for the specified channel -func newSubscriptionSucceededEvent(channel, data string) subscriptionSucceededEvent { - return subscriptionSucceededEvent{Event: "pusher_internal:subscription_succeeded", Channel: channel, Data: data} +func NewSubscriptionSucceeded(channel, data string) SubscriptionSucceeded { + return SubscriptionSucceeded{Event: "pusher_internal:subscription_succeeded", Channel: channel, Data: data} } // Data Subscription Succeed @@ -89,22 +91,26 @@ func newSubscriptionSucceededEvent(channel, data string) subscriptionSucceededEv // \"count\": 2 // } // }" -type subscriptionSucceededEventPresenceData struct { +type SubscriptionSucceededPresenceData struct { Ids []string `json:"ids"` Hash map[string]interface{} `json:"hash"` Count int `json:"count"` } -func newSubscriptionSucceedEventPresenceData(c *channel) subscriptionSucceededEventPresenceData { - event := subscriptionSucceededEventPresenceData{} +func NewSubscriptionSucceedPresenceData(subscriptions map[string]*subscription.Subscription) SubscriptionSucceededPresenceData { + event := SubscriptionSucceededPresenceData{} - var ids []string - hash := make(map[string]interface{}, c.TotalSubscriptions()) + var ( + ids []string + hash = make(map[string]interface{}, len(subscriptions)) + ) - for _, s := range c.Subscriptions { + for _, s := range subscriptions { // Do you have any other idea? var js interface{} - json.Unmarshal([]byte(s.Data), &js) + if err := json.Unmarshal([]byte(s.Data), &js); err != nil { + continue + } hash[s.ID] = js ids = append(ids, s.ID) @@ -112,7 +118,7 @@ func newSubscriptionSucceedEventPresenceData(c *channel) subscriptionSucceededEv event.Ids = ids event.Hash = hash - event.Count = c.TotalSubscriptions() + event.Count = len(subscriptions) return event } @@ -121,28 +127,28 @@ func newSubscriptionSucceedEventPresenceData(c *channel) subscriptionSucceededEv // "event": "pusher:pong", // "data": {} // } -type pongEvent struct { +type Pong struct { Event string `json:"event"` Data string `json:"data"` } // Create a new pong event -func newPongEvent() pongEvent { - return pongEvent{Event: "pusher:pong", Data: "{}"} +func NewPong() Pong { + return Pong{Event: "pusher:pong", Data: "{}"} } // { // "event": "pusher:ping", // "data": {} // } -type pingEvent struct { +type Ping struct { Event string `json:"event"` Data string `json:"data"` } // Create a new ping event -func newPingEvent() pingEvent { - return pingEvent{Event: "pusher:ping", Data: "{}"} +func NewPing() Ping { + return Ping{Event: "pusher:ping", Data: "{}"} } // { @@ -152,7 +158,7 @@ func newPingEvent() pingEvent { // "code": 4000 // } // } -type errorEvent struct { +type Error struct { Event string `json:"event"` Data interface{} `json:"data"` } @@ -160,14 +166,11 @@ type errorEvent struct { // Create a new error event // Pusher protocol is very strange in some parts // It send null in some errors. -func newErrorEvent(code int, message string) errorEvent { - - type dataErrorEvent struct { +func NewError(code int, message string) Error { + var data = struct { Code *int `json:"code"` Message string `json:"message"` - } - - var data = dataErrorEvent{ + }{ Message: message, } @@ -177,7 +180,7 @@ func newErrorEvent(code int, message string) errorEvent { data.Code = &code } - return errorEvent{Event: "pusher:error", Data: data} + return Error{Event: "pusher:error", Data: data} } // { @@ -187,27 +190,25 @@ func newErrorEvent(code int, message string) errorEvent { // "activity_timeout" : 120 // } // } -type connectionEstablishedEventData struct { - SocketID string `json:"socket_id"` - ActivityTimeout int `json:"activity_timeout"` -} - -type connectionEstablishedEvent struct { +type ConnectionEstablished struct { Event string `json:"event"` Data string `json:"data"` } // Create a new connection established event using the specified socketId -func newConnectionEstablishedEvent(socketID string) connectionEstablishedEvent { - data := connectionEstablishedEventData{SocketID: socketID, ActivityTimeout: 120} - - b, err := json.Marshal(data) +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 connectionEstablishedEvent{Event: "pusher:connection_established", Data: string(b)} + return ConnectionEstablished{Event: "pusher:connection_established", Data: string(b)} } // { @@ -215,14 +216,14 @@ func newConnectionEstablishedEvent(socketID string) connectionEstablishedEvent { // "channel": "presence-example-channel", // "data": String // } -type memberAddedEvent struct { +type MemberAdded struct { Event string `json:"event"` Channel string `json:"channel"` Data string `json:"data"` } -func newMemberAddedEvent(channel, data string) memberAddedEvent { - return memberAddedEvent{Event: "pusher_internal:member_added", Channel: channel, Data: data} +func NewMemberAdded(channel, data string) MemberAdded { + return MemberAdded{Event: "pusher_internal:member_added", Channel: channel, Data: data} } // { @@ -230,24 +231,24 @@ func newMemberAddedEvent(channel, data string) memberAddedEvent { // "channel": "presence-example-channel", // "data": String // } -type memberRemovedEvent struct { +type MemberRemoved struct { Event string `json:"event"` Channel string `json:"channel"` Data string `json:"data"` } -func newMemberRemovedEvent(channel string, s *subscription) memberRemovedEvent { +func NewMemberRemoved(channel string, userID string) MemberRemoved { data, err := json.Marshal(struct { UserID string `json:"user_id"` }{ - UserID: s.ID, + UserID: userID, }) if err != nil { log.Error(err) } - return memberRemovedEvent{Event: "pusher_internal:member_removed", Channel: channel, Data: string(data)} + return MemberRemoved{Event: "pusher_internal:member_removed", Channel: channel, Data: string(data)} } // { @@ -255,19 +256,19 @@ func newMemberRemovedEvent(channel string, s *subscription) memberRemovedEvent { // "channel": "The channel", // "data": {} // } -type rawEvent struct { +type Raw struct { Event string `json:"event"` Channel string `json:"channel"` Data json.RawMessage `json:"data"` } -type responseEvent struct { +type Response struct { Event string `json:"event"` Channel string `json:"channel"` Data interface{} `json:"data"` } // The response event that is broadcasted to the client sockets -func newResponseEvent(name, channel string, data interface{}) responseEvent { - return responseEvent{Event: name, Channel: channel, Data: data} +func NewResponse(name, channel string, data interface{}) Response { + return Response{Event: name, Channel: channel, Data: data} } diff --git a/ipe/events_test.go b/events/events_test.go similarity index 83% rename from ipe/events_test.go rename to events/events_test.go index ed7d7ef..fdb813c 100644 --- a/ipe/events_test.go +++ b/events/events_test.go @@ -1,4 +1,4 @@ -package ipe +package events import ( "bytes" @@ -7,7 +7,7 @@ import ( ) func Test_newErrorEvent_with_invalid_code(t *testing.T) { - event := newErrorEvent(0, "The error message") + event := NewError(0, "The error message") data, _ := json.Marshal(event) expected := `{"event":"pusher:error","data":{"code":null,"message":"The error message"}}` @@ -18,7 +18,7 @@ func Test_newErrorEvent_with_invalid_code(t *testing.T) { } func Test_newErrorEvent_with_valid_code(t *testing.T) { - event := newErrorEvent(4007, "Unsupported protocol version") + event := NewError(4007, "Unsupported protocol version") data, _ := json.Marshal(event) expected := `{"event":"pusher:error","data":{"code":4007,"message":"Unsupported protocol version"}}` diff --git a/functional/Procfile b/functional/Procfile index 807bb49..66048ed 100644 --- a/functional/Procfile +++ b/functional/Procfile @@ -1,2 +1,2 @@ client: go run client.go -server: go run ../main.go -config ./functional-config.json -alsologtostderr \ No newline at end of file +server: go run ../cmd/main.go -config ./functional-config.json -alsologtostderr \ No newline at end of file diff --git a/functional/client.go b/functional/client.go index 505ee3c..f02655d 100644 --- a/functional/client.go +++ b/functional/client.go @@ -5,6 +5,7 @@ import ( "io/ioutil" "log" "net/http" + "net/http/httputil" "github.com/pusher/pusher-http-go" ) @@ -13,10 +14,11 @@ var client pusher.Client func init() { client = pusher.Client{ - AppId: "1", - Key: "278d525bdf162c739803", - Secret: "7ad3753142a6693b25b9", - Host: ":8080", + AppId: "1", + Key: "278d525bdf162c739803", + Secret: "7ad3753142a6693b25b9", + Host: ":8080", + Cluster: "hello", } } @@ -34,7 +36,7 @@ func pusherPresenceAuth(res http.ResponseWriter, req *http.Request) { panic(err) } - fmt.Fprint(res, string(response)) + _, _ = fmt.Fprint(res, string(response)) } func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) { @@ -48,19 +50,38 @@ func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) { panic(err) } - fmt.Fprint(res, string(response)) + _, _ = fmt.Fprint(res, string(response)) } 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 { + panic(err) + } - fmt.Fprint(res, "OK") + _, _ = fmt.Fprint(res, "OK") +} + +func hookcallback(res http.ResponseWriter, r *http.Request) { + bytes, err := httputil.DumpRequest(r, true) + if err != nil { + panic(err) + } + fmt.Println(string(bytes)) + + _, err = client.Trigger("private-webhook", "mywebhook", "The Webhoook from server") + if err != nil { + panic(err) + } + + _, _ = fmt.Fprint(res, "OK") } func main() { http.HandleFunc("/pusher/presence/auth", pusherPresenceAuth) http.HandleFunc("/pusher/private/auth", pusherPrivateAuth) http.HandleFunc("/trigger", triggerMessage) + http.HandleFunc("/hook", hookcallback) http.Handle("/", http.FileServer(http.Dir("./"))) - http.ListenAndServe(":5000", nil) + _ = http.ListenAndServe(":5000", nil) } diff --git a/functional/functional-config.json b/functional/functional-config.json index fc88457..c26e528 100644 --- a/functional/functional-config.json +++ b/functional/functional-config.json @@ -14,8 +14,8 @@ "Name": "App for Functional Test", "AppID": "1", "UserEvents": true, - "WebHooks": false, - "URLWebHook": "http://127.0.0.1:4567/php/hook.php" + "WebHooks": true, + "URLWebHook": "http://127.0.0.1:5000/hook" } ] } diff --git a/functional/test.pusher.js b/functional/test.pusher.js index 3c0d1af..eab9f07 100644 --- a/functional/test.pusher.js +++ b/functional/test.pusher.js @@ -1,136 +1,157 @@ +let assert = chai.assert; -var assert = chai.assert; +let APP_KEY = "278d525bdf162c739803"; +let HOST = "localhost"; +let PORT = 8080; +let AUTH = "http://localhost:5000/pusher/private/auth"; +let AUTH_PRESENCE = "http://localhost:5000/pusher/presence/auth"; -var APP_KEY = "278d525bdf162c739803"; -var HOST = "localhost"; -var PORT = 8080; -var AUTH = "http://localhost:5000/pusher/private/auth" -var AUTH_PRESENCE = "http://localhost:5000/pusher/presence/auth" - -Pusher.log = function(msg) { - console.log(msg); +Pusher.log = function (msg) { + console.log(msg); }; function getPusher(auth) { - return new Pusher(APP_KEY, { - wsHost: HOST, - wsPort: PORT, - authEndpoint: auth, - enabledTransports: ["ws"], - disabledTransports: ["flash"], - cluster: "hello", // Should be ignored - }); + return new Pusher(APP_KEY, { + wsHost: HOST, + wsPort: PORT, + authEndpoint: auth, + enabledTransports: ["ws"], + disabledTransports: ["flash"], + cluster: "hello", // Should be ignored + }); } -describe("Pusher", function() { +describe("Pusher", function () { - describe("connection", function() { - it("should connect sucessfully with correct config", function(done) { - var pusher = getPusher(AUTH); + describe("connection", function () { + it("should connect sucessfully with correct config", function (done) { + let pusher = getPusher(AUTH); - pusher.connection.bind('connected', function() { - assert.ok(true, "Connected"); - done(); - }); - }); + pusher.connection.bind('connected', function () { + assert.ok(true, "Connected"); + done(); + }); + }); - it("should not connect without the correct config", function(done) { - var pusher = new Pusher("INVALID_APP_KEY", { - wsHost: HOST, - wsPort: PORT, - enabledTransports: ["ws"], - disabledTransports: ["flash"] - }); + it("should not connect without the correct config", function (done) { + let pusher = new Pusher("INVALID_APP_KEY", { + wsHost: HOST, + wsPort: PORT, + enabledTransports: ["ws"], + disabledTransports: ["flash"] + }); - pusher.connection.bind('disconnected', function() { - assert.ok(true, "Not Connected"); - done(); - }); - }); - }); // connection + pusher.connection.bind('disconnected', function () { + assert.ok(true, "Not Connected"); + done(); + }); + }); + }); // connection - describe("subscription", function() { - it("should subscribe to a public channel", function(done) { - var pusher = getPusher(AUTH); + describe("subscription", function () { + it("should subscribe to a public channel", function (done) { + let pusher = getPusher(AUTH); - var channel = pusher.subscribe('public-channel'); - channel.bind("pusher:subscription_succeeded", function(data) { - assert.ok(true, "Connected to the channel"); - done(); - }); - }); + let channel = pusher.subscribe('public-channel'); + channel.bind("pusher:subscription_succeeded", function (data) { + assert.ok(true, "Connected to the channel"); + done(); + }); + }); - it("should subscribe to a private channel", function(done) { - var pusher = getPusher(AUTH); + it("should subscribe to a private channel", function (done) { + let pusher = getPusher(AUTH); - var channel = pusher.subscribe('private-channel'); - channel.bind("pusher:subscription_succeeded", function(data) { - assert.ok(true, "Connected to the channel"); - done(); - }); - }); + let channel = pusher.subscribe('private-channel'); + channel.bind("pusher:subscription_succeeded", function (data) { + assert.ok(true, "Connected to the channel"); + done(); + }); + }); - it("should subscribe to a presence channel", function(done) { - var pusher = getPusher(AUTH_PRESENCE); + it("should subscribe to a presence channel", function (done) { + let pusher = getPusher(AUTH_PRESENCE); - var channel = pusher.subscribe('presence-channel'); - channel.bind("pusher:subscription_succeeded", function(data) { - assert.ok(true, "Connected to the channel"); - done(); - }); - }); - }); // subscription + let channel = pusher.subscribe('presence-channel'); + channel.bind("pusher:subscription_succeeded", function (data) { + assert.ok(true, "Connected to the channel"); + done(); + }); + }); + }); // subscription - describe("events", function() { - it('should not allowed client events on public channels', function(done) { - var pusher = getPusher(AUTH); - var channel = pusher.subscribe('public-channel'); + describe("hooks", function () { + it('should receive hook', function (done) { + let pusher = getPusher(AUTH); + let channel = pusher.subscribe('private-webhook'); - channel.bind("pusher:subscription_succeeded", function(data) { - channel.trigger("client-message", "The message"); - }); + channel.bind("pusher:subscription_succeeded", function (data) { + console.log("subscribed"); + }); - pusher.bind("pusher:error", function(data) { - assert.ok(true, "Expected error"); - done(); - }); - }); + channel.bind("mywebhook", function (data) { + console.log(data); + assert.equal(data, "The Webhoook from server"); + done(); + }); + }); + }); // hooks - it('should allow client events on private channels', function(done) { - var pusher_a = getPusher(AUTH); - var pusher_b = getPusher(AUTH); + describe("events", function () { + it('should not allowed client events on public channels', function (done) { + let pusher = getPusher(AUTH); + let channel = pusher.subscribe('public-channel'); - var channel_a = pusher_a.subscribe('private-channel'); - var channel_b = pusher_b.subscribe('private-channel'); + channel.bind("pusher:subscription_succeeded", function (data) { + channel.trigger("client-message", "The message"); + }); - channel_a.bind("pusher:subscription_succeeded", function() { - channel_a.trigger("client-message", "The message"); - }); + pusher.bind("pusher:error", function (data) { + assert.ok(true, "Expected error"); + done(); + }); + }); - channel_b.bind("client-message", function(data) { - assert.equal(data, "The message"); - done(); - }); - }); + it('should allow client events on private channels', function (done) { + let pusher_a = getPusher(AUTH); + let pusher_b = getPusher(AUTH); - it('should publish event on private channel', function(done) { - var pusher_a = getPusher(AUTH); - var pusher_b = getPusher(AUTH); + let channel_a = pusher_a.subscribe('private-channel'); + let channel_b = pusher_b.subscribe('private-channel'); - var channel_a = pusher_a.subscribe('private-messages'); - var channel_b = pusher_b.subscribe('private-messages'); + channel_a.bind("pusher:subscription_succeeded", function () { + channel_a.trigger("client-message", "The message"); + }); - channel_a.bind("pusher:subscription_succeeded", function() { - var xhttp = new XMLHttpRequest(); - xhttp.open("GET", "/trigger", true); - xhttp.send(); - }); + channel_b.bind("client-message", function (data) { + assert.equal(data, "The message"); + done(); + }); + }); - channel_b.bind("messages", function(data) { - assert.equal(data, "The message from server"); - done(); - }); - }); + it('should publish event on private channel', function (done) { + let pusher_a = getPusher(AUTH); + let pusher_b = getPusher(AUTH); - }); // events + let channel_a = pusher_a.subscribe('private-messages'); + let channel_b = pusher_b.subscribe('private-messages'); + + channel_a.bind("pusher:subscription_succeeded", function () { + console.log("channel_a connected"); + let xhttp = new XMLHttpRequest(); + xhttp.open("GET", "/trigger", true); + xhttp.send(); + }); + + channel_b.bind("pusher:subscription_succeeded", function () { + console.log("channel_b connected"); + }); + + channel_b.bind("messages", function (data) { + assert.equal(data, "The message from server"); + done(); + }); + }); + + }); // events }); diff --git a/glide.lock b/glide.lock deleted file mode 100644 index aa09448..0000000 --- a/glide.lock +++ /dev/null @@ -1,12 +0,0 @@ -hash: ee2de935e70f2a39cbfd2dd16add0736c698b32f5c02409d0c25844d883c91a1 -updated: 2016-08-22T22:55:07.279113421-03:00 -imports: -- name: github.com/golang/glog - version: 23def4e6c14b4da8ac2ed8007337bc5eb5007998 -- name: github.com/gorilla/websocket - version: a69d25be2fe2923a97c2af6849b2f52426f68fc0 -- name: github.com/pressly/chi - version: 12aad88c7d86de2affe686f855b6ed94a07cba9c - subpackages: - - middleware -testImports: [] diff --git a/glide.yaml b/glide.yaml deleted file mode 100644 index 92332d0..0000000 --- a/glide.yaml +++ /dev/null @@ -1,7 +0,0 @@ -package: github.com/dimiro1/ipe -import: -- package: github.com/golang/glog -- package: github.com/gorilla/websocket -- package: github.com/pressly/chi -excludeDirs: -- functional \ No newline at end of file diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..1557571 --- /dev/null +++ b/go.mod @@ -0,0 +1,14 @@ +module ipe + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b + github.com/gorilla/context v1.1.1 // indirect + github.com/gorilla/handlers v1.4.0 + github.com/gorilla/mux v1.6.2 + github.com/gorilla/websocket v1.4.0 + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/pusher/pusher-http-go v1.3.0 + github.com/stretchr/testify v1.2.2 // indirect + golang.org/x/crypto v0.0.0-20181112202954-3d3f9f413869 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..9b3e19c --- /dev/null +++ b/go.sum @@ -0,0 +1,20 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58= +github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= +github.com/gorilla/context v1.1.1 h1:AWwleXJkX/nhcU9bZSnZoi3h/qGYqQAGhq6zZe/aQW8= +github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg= +github.com/gorilla/handlers v1.4.0 h1:XulKRWSQK5uChr4pEgSE4Tc/OcmnU9GJuSwdog/tZsA= +github.com/gorilla/handlers v1.4.0/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ= +github.com/gorilla/mux v1.6.2 h1:Pgr17XVTNXAk3q/r4CpKzC5xBM/qW1uVLV+IhRZpIIk= +github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs= +github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q= +github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pusher/pusher-http-go v1.3.0 h1:dWrjIsNheCUEo6YE9qlS8pIl3XzJa5yM8VlBu/LUl3M= +github.com/pusher/pusher-http-go v1.3.0/go.mod h1:XAv1fxRmVTI++2xsfofDhg7whapsLRG/gH/DXbF3a18= +github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +golang.org/x/crypto v0.0.0-20181112202954-3d3f9f413869 h1:kkXA53yGe04D0adEYJwEVQjeBppL01Exg+fnMjfUraU= +golang.org/x/crypto v0.0.0-20181112202954-3d3f9f413869/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= diff --git a/ipe.go b/ipe.go new file mode 100644 index 0000000..d1fd2a4 --- /dev/null +++ b/ipe.go @@ -0,0 +1,108 @@ +// Copyright 2015 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 ipe + +import ( + "encoding/json" + "math/rand" + "net/http" + "os" + "time" + + log "github.com/golang/glog" + "github.com/gorilla/handlers" + "github.com/gorilla/mux" + + "ipe/api" + "ipe/app" + "ipe/config" + "ipe/storage" + "ipe/websockets" +) + +// Start Parse the configuration file and starts the ipe server +// It Panic if could not start the HTTP or HTTPS server +func Start(filename string) { + var conf config.File + + rand.Seed(time.Now().Unix()) + file, err := os.Open(filename) + + if err != nil { + log.Error(err) + return + } + + defer func() { + if err := file.Close(); err != nil { + log.Error(err) + } + }() + + // Reading config + if err := json.NewDecoder(file).Decode(&conf); err != nil { + log.Error(err) + return + } + + // Using a in memory database + inMemoryStorage := storage.NewInMemory() + + // Adding applications + for _, a := range conf.Apps { + application := app.NewApplication( + a.Name, + a.AppID, + a.Key, + a.Secret, + a.OnlySSL, + a.ApplicationDisabled, + a.UserEvents, + a.WebHooks, + a.URLWebHook, + ) + + if err := inMemoryStorage.AddApp(application); err != nil { + log.Error(err) + return + } + } + + router := mux.NewRouter() + router.Use(handlers.RecoveryHandler()) + + router.Path("/app/{key}").Methods("GET").Handler( + websockets.NewWebsocket(inMemoryStorage), + ) + + appsRouter := router.PathPrefix("/apps/{app_id}").Subrouter() + appsRouter.Use( + api.CheckAppDisabled(inMemoryStorage), + api.Authentication(inMemoryStorage), + ) + + appsRouter.Path("/events").Methods("POST").Handler( + api.NewPostEvents(inMemoryStorage), + ) + appsRouter.Path("/channels").Methods("GET").Handler( + api.NewGetChannels(inMemoryStorage), + ) + appsRouter.Path("/channels/{channel_name}").Methods("GET").Handler( + api.NewGetChannel(inMemoryStorage), + ) + appsRouter.Path("/channels/{channel_name}/users").Methods("GET").Handler( + api.NewGetChannelUsers(inMemoryStorage), + ) + + if conf.SSL { + go func() { + log.Infof("Starting HTTPS service on %s ...", conf.SSLHost) + log.Fatal(http.ListenAndServeTLS(conf.SSLHost, conf.SSLCertFile, conf.SSLKeyFile, router)) + }() + } + + log.Infof("Starting HTTP service on %s ...", conf.Host) + log.Fatal(http.ListenAndServe(conf.Host, router)) +} diff --git a/ipe/app.go b/ipe/app.go deleted file mode 100644 index ed7aec1..0000000 --- a/ipe/app.go +++ /dev/null @@ -1,234 +0,0 @@ -// 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 ipe - -import ( - "errors" - "expvar" - "fmt" - "sync" - - log "github.com/golang/glog" -) - -// An App -type app struct { - sync.Mutex - - Name string - AppID string - Key string - Secret string - OnlySSL bool - ApplicationDisabled bool - UserEvents bool - WebHooks bool - URLWebHook string - - Channels map[string]*channel - Connections map[string]*connection - - Stats *expvar.Map -} - -func newApp(name, appID, key, secret string, onlySSL, disabled, userEvents, webHooks bool, webHookURL string) *app { - - a := &app{ - Name: name, - AppID: appID, - Key: key, - Secret: secret, - OnlySSL: onlySSL, - ApplicationDisabled: disabled, - UserEvents: userEvents, - WebHooks: webHooks, - URLWebHook: webHookURL, - } - - a.Connections = make(map[string]*connection) - a.Channels = make(map[string]*channel) - a.Stats = expvar.NewMap(fmt.Sprintf("%s (%s)", a.Name, a.AppID)) - - return a -} - -// Only Presence channels -func (a *app) PresenceChannels() []*channel { - var channels []*channel - - for _, c := range a.Channels { - if c.IsPresence() { - channels = append(channels, c) - } - } - - return channels -} - -// Only Private channels -func (a *app) PrivateChannels() []*channel { - var channels []*channel - - for _, c := range a.Channels { - if c.IsPrivate() { - channels = append(channels, c) - } - } - - return channels -} - -// Only Public channels -func (a *app) PublicChannels() []*channel { - var channels []*channel - - for _, c := range a.Channels { - if c.IsPublic() { - channels = append(channels, c) - } - } - - return channels -} - -// Disconnect Socket -func (a *app) 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) { - c.Unsubscribe(a, conn) - } - } - - // Remove from app - a.Lock() - defer a.Unlock() - - _, exists := a.Connections[conn.SocketID] - - if !exists { - return - } - - delete(a.Connections, conn.SocketID) - - a.Stats.Add("TotalConnections", -1) -} - -// Connect a new Subscriber -func (a *app) Connect(conn *connection) { - log.Infof("Adding a new Connection %s to app %s", conn.SocketID, a.Name) - a.Lock() - defer a.Unlock() - - a.Connections[conn.SocketID] = conn - - a.Stats.Add("TotalConnections", 1) -} - -// Find a Connection on this app -func (a *app) FindConnection(socketID string) (*connection, error) { - conn, exists := a.Connections[socketID] - - if exists { - return conn, nil - } - - return nil, errors.New("Connection not found") -} - -// DeleteChannel removes the channel from app -func (a *app) RemoveChannel(c *channel) { - log.Infof("Remove the channel %s from app %s", c.ChannelID, a.Name) - a.Lock() - defer a.Unlock() - - delete(a.Channels, c.ChannelID) - - 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) -} - -// Add a new Channel to this APP -func (a *app) AddChannel(c *channel) { - log.Infof("Adding a new channel %s to app %s", c.ChannelID, a.Name) - - a.Lock() - defer a.Unlock() - - a.Channels[c.ChannelID] = 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) -} - -// Returns a Channel from this app -// If not found then the channel is created and added to this app -func (a *app) FindOrCreateChannelByChannelID(n string) *channel { - c, err := a.FindChannelByChannelID(n) - - if err != nil { - c = newChannel(n) - a.AddChannel(c) - } - - return c -} - -// Find the channel by channel ID -func (a *app) FindChannelByChannelID(n string) (*channel, error) { - c, exists := a.Channels[n] - - if exists { - return c, nil - } - - return nil, errors.New("Channel does not exists") -} - -func (a *app) Publish(c *channel, event rawEvent, ignore string) error { - a.Stats.Add("TotalUniqueMessages", 1) - - return c.Publish(a, event, ignore) -} - -func (a *app) Unsubscribe(c *channel, conn *connection) error { - return c.Unsubscribe(a, conn) -} - -func (a *app) Subscribe(c *channel, conn *connection, data string) error { - return c.Subscribe(a, conn, data) -} diff --git a/ipe/app_test.go b/ipe/app_test.go deleted file mode 100644 index a6e07a8..0000000 --- a/ipe/app_test.go +++ /dev/null @@ -1,252 +0,0 @@ -// 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 ipe - -import ( - "strconv" - "testing" -) - -var id = 0 - -func newTestApp() *app { - - a := newApp("Test", strconv.Itoa(id), "123", "123", false, false, true, false, "") - id++ - - return a -} - -func TestConnect(t *testing.T) { - app := newTestApp() - - app.Connect(newConnection("socketID", mockSocket{})) - - if len(app.Connections) != 1 { - t.Errorf("len(app.Connections) == %d, wants %d", len(app.Connections), 1) - } - -} - -func TestDisconnect(t *testing.T) { - app := newTestApp() - - app.Connect(newConnection("socketID", mockSocket{})) - app.Disconnect("socketID") - - if len(app.Connections) != 0 { - t.Errorf("len(app.Connections) == %d, wants %d", len(app.Connections), 0) - } - -} - -func TestFindConnection(t *testing.T) { - app := newTestApp() - - app.Connect(newConnection("socketID", mockSocket{})) - - if _, err := app.FindConnection("socketID"); err != nil { - t.Errorf("app.FindConnection('socketID') == _, %q, wants %v", err, nil) - } - - if _, err := app.FindConnection("NotFound"); err == nil { - t.Errorf("app.FindConnection('socketID') == _, %q, wants !nil", err) - } - -} - -func TestFindChannelByChannelID(t *testing.T) { - app := newTestApp() - - channel := newChannel("ID") - app.AddChannel(channel) - - if _, err := app.FindChannelByChannelID("ID"); err != nil { - t.Errorf("app.FindChannelByChannelID('ID') == _, %q, wants %v", err, nil) - } -} - -func TestFindOrCreateChannelByChannelID(t *testing.T) { - app := newTestApp() - - if len(app.Channels) != 0 { - t.Errorf("len(app.Channels) == %d, wants %d", len(app.Channels), 0) - } - - app.FindOrCreateChannelByChannelID("ID") - - if len(app.Channels) != 1 { - t.Errorf("len(app.Channels) == %d, wants %d", len(app.Channels), 1) - } - -} - -func TestRemoveChannel(t *testing.T) { - app := newTestApp() - - if len(app.Channels) != 0 { - t.Errorf("len(app.Channels) == %d, wants %d", len(app.Channels), 0) - } - - channel := newChannel("ID") - app.AddChannel(channel) - - if len(app.Channels) != 1 { - t.Errorf("len(app.Channels) == %d, wants %d", len(app.Channels), 1) - } - - app.RemoveChannel(channel) - - if len(app.Channels) != 0 { - t.Errorf("len(app.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(app.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 0) - } - - app.AddChannel(newChannel("ID")) - - if len(app.PublicChannels()) != 1 { - t.Errorf("len(app.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 1) - } - - // Presence - - if len(app.PresenceChannels()) != 0 { - t.Errorf("len(app.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 0) - } - - app.AddChannel(newChannel("presence-test")) - - if len(app.PresenceChannels()) != 1 { - t.Errorf("len(app.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 1) - } - - // Private - - if len(app.PrivateChannels()) != 0 { - t.Errorf("len(app.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 0) - } - - app.AddChannel(newChannel("private-test")) - - if len(app.PrivateChannels()) != 1 { - t.Errorf("len(app.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 1) - } - -} - -func Test_AllChannels(t *testing.T) { - app := newTestApp() - app.AddChannel(newChannel("private-test")) - app.AddChannel(newChannel("presence-test")) - app.AddChannel(newChannel("test")) - - if len(app.Channels) != 3 { - t.Errorf("len(app.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(app.Connections) == %d, wants %d", len(app.Connections), 0) - } - - conn := newConnection("1", mockSocket{}) - app.Connect(conn) - - if len(app.Connections) != 1 { - t.Errorf("len(app.Connections) == %d, wants %d", len(app.Connections), 1) - } -} - -func Test_find_subscriber(t *testing.T) { - app := newTestApp() - conn := newConnection("1", 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(app.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 0) - } - - c := app.FindOrCreateChannelByChannelID("id") - - if len(app.PublicChannels()) != 1 { - t.Errorf("len(app.PublicChannels()) == %d, wants %d", len(app.PublicChannels()), 1) - } - - if c.ChannelID != "id" { - t.Errorf("c.ChannelID == %s, wants %s", c.ChannelID, "id") - } - - // Presence - if len(app.PresenceChannels()) != 0 { - t.Errorf("len(app.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 0) - } - - c = app.FindOrCreateChannelByChannelID("presence-test") - - if len(app.PresenceChannels()) != 1 { - t.Errorf("len(app.PresenceChannels()) == %d, wants %d", len(app.PresenceChannels()), 1) - } - - if c.ChannelID != "presence-test" { - t.Errorf("c.ChannelID == %s, wants %s", c.ChannelID, "presence-test") - } - - // Private - if len(app.PrivateChannels()) != 0 { - t.Errorf("len(app.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 0) - } - - c = app.FindOrCreateChannelByChannelID("private-test") - - if len(app.PrivateChannels()) != 1 { - t.Errorf("len(app.PrivateChannels()) == %d, wants %d", len(app.PrivateChannels()), 1) - } - - if c.ChannelID != "private-test" { - t.Errorf("c.ChannelID == %s, wants %s", c.ChannelID, "private-test") - } - -} diff --git a/ipe/channel.go b/ipe/channel.go deleted file mode 100644 index b347357..0000000 --- a/ipe/channel.go +++ /dev/null @@ -1,224 +0,0 @@ -// 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 ipe - -import ( - "encoding/json" - "errors" - "sync" - "time" - - "github.com/dimiro1/ipe/utils" - log "github.com/golang/glog" -) - -// A Channel -type channel struct { - sync.Mutex - - CreatedAt time.Time - ChannelID string - Subscriptions map[string]*subscription -} - -// Return true if the channel has at least one subscriber -func (c *channel) IsOccupied() bool { - return c.TotalSubscriptions() > 0 -} - -// Check if the type of the channel is presence or is private -func (c *channel) IsPresenceOrPrivate() bool { - return c.IsPresence() || c.IsPrivate() -} - -// Check if the type of the channel is public -func (c *channel) IsPublic() bool { - return !c.IsPresenceOrPrivate() -} - -// Check if the type of the channel is presence -func (c *channel) IsPresence() bool { - return utils.IsPresenceChannel(c.ChannelID) -} - -// Check if the type of the channel is private -func (c *channel) IsPrivate() bool { - return utils.IsPrivateChannel(c.ChannelID) -} - -// Get the total of subscribers -func (c *channel) TotalSubscriptions() int { - return len(c.Subscriptions) -} - -// Get the total of users. -func (c *channel) TotalUsers() int { - total := make(map[string]int) - - for _, s := range c.Subscriptions { - total[s.ID]++ - } - - return len(total) -} - -// Add a new subscriber to the channel -func (c *channel) Subscribe(a *app, conn *connection, channelData string) error { - log.Infof("Subscribing %s to channel %s", conn.SocketID, c.ChannelID) - - c.Lock() - defer c.Unlock() - - subscription := newSubscription(conn, channelData) - c.Subscriptions[conn.SocketID] = subscription - - if !c.IsPresence() { - 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]subscriptionSucceededEventPresenceData) - 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 - if c.TotalSubscriptions() == 1 { - a.TriggerChannelOccupiedHook(c) - } - - return nil -} - -// IsSubscribed check if the user is subscribed -func (c *channel) IsSubscribed(conn *connection) bool { - _, exists := c.Subscriptions[conn.SocketID] - return exists -} - -// Remove the subscriber from the channel -// It destroy the channel if the channels does not have any subscribers. -func (c *channel) Unsubscribe(a *app, conn *connection) error { - log.Infof("Unsubscribing %s from channel %s", conn.SocketID, c.ChannelID) - - c.Lock() - defer c.Unlock() - - subscription, exists := c.Subscriptions[conn.SocketID] - - if !exists { - return errors.New("Subscription not found") - } - - delete(c.Subscriptions, conn.SocketID) - - if c.IsPresence() { - // Publish pusher_internal:member_removed - c.PublishMemberRemovedEvent(a, subscription) - // Webhook - a.TriggerMemberRemovedHook(c, subscription) - } - - if !c.IsOccupied() { - // WebHook - a.TriggerChannelVacatedHook(c) - - // Remove the empty Channel - a.RemoveChannel(c) - } - - return nil -} - -// Create a new Channel -func newChannel(channelID string) *channel { - log.Infof("Creating a new channel: %s", channelID) - - return &channel{ChannelID: channelID, CreatedAt: time.Now(), Subscriptions: make(map[string]*subscription)} -} - -// Publish a MemberAddedEvent to all subscriptions -func (c *channel) PublishMemberAddedEvent(a *app, data string, subscription *subscription) { - for _, subs := range c.Subscriptions { - if subs != subscription { - subs.Connection.Publish(newMemberAddedEvent(c.ChannelID, data)) - } - } -} - -// Publish a MemberRemovedEvent to all subscriptions -func (c *channel) PublishMemberRemovedEvent(a *app, subscription *subscription) { - for _, subs := range c.Subscriptions { - if subs != subscription { - subs.Connection.Publish(newMemberRemovedEvent(c.ChannelID, subscription)) - } - } -} - -// Publish messages to all Subscribers -func (c *channel) Publish(a *app, event rawEvent, ignore string) error { - b, err := event.Data.MarshalJSON() - - if err != nil { - return err - } - - var v interface{} - - if err := json.Unmarshal(b, &v); err != nil { - return err - } - - log.Infof("Publishing message %+v to channel %s", v, c.ChannelID) - - for _, subs := range c.Subscriptions { - if subs.Connection.SocketID != ignore { - subs.Connection.Publish(newResponseEvent(event.Event, event.Channel, v)) - } else { - // Webhook - if utils.IsClientEvent(event.Event) { - a.TriggerClientEventHook(c, subs, event.Event, v) - } - } - } - - return nil -} diff --git a/ipe/connection.go b/ipe/connection.go deleted file mode 100644 index fe904cf..0000000 --- a/ipe/connection.go +++ /dev/null @@ -1,35 +0,0 @@ -// 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 ipe - -import ( - "time" - - log "github.com/golang/glog" -) - -// socket interface to write to the client -type socket interface { - WriteJSON(interface{}) error -} - -// An User Connection -type connection struct { - SocketID string - Socket socket - CreatedAt time.Time -} - -// Create a new Subscriber -func newConnection(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 atached to this client -func (conn *connection) Publish(m interface{}) { - conn.Socket.WriteJSON(m) -} diff --git a/ipe/constants.go b/ipe/constants.go deleted file mode 100644 index d67d2b3..0000000 --- a/ipe/constants.go +++ /dev/null @@ -1,12 +0,0 @@ -// 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 ipe - -// Only this version is supported -const supportedProtocolVersion = 7 - -// // 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 diff --git a/ipe/db.go b/ipe/db.go deleted file mode 100644 index 800b92e..0000000 --- a/ipe/db.go +++ /dev/null @@ -1,55 +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. - -package ipe - -import ( - "errors" - "sync" -) - -// db represents a app database -// For now it there is only one memory database implementation -// but in the future I can write a sql implementation -type db interface { - GetAppByAppID(appID string) (*app, error) - GetAppByKey(key string) (*app, error) - AddApp(*app) error -} - -type memdb struct { - sync.Mutex - Apps []*app -} - -func newMemdb() db { - return &memdb{} -} - -func (db *memdb) AddApp(a *app) error { - db.Lock() - db.Apps = append(db.Apps, a) - db.Unlock() - return nil -} - -// GetAppByAppID returns an App with by appID -func (db *memdb) GetAppByAppID(appID string) (*app, error) { - for _, a := range db.Apps { - if a.AppID == appID { - return a, nil - } - } - return nil, errors.New("App not found") -} - -// GetAppByKey returns an App with by key -func (db *memdb) GetAppByKey(key string) (*app, error) { - for _, a := range db.Apps { - if a.Key == key { - return a, nil - } - } - return nil, errors.New("App not found") -} diff --git a/ipe/db_test.go b/ipe/db_test.go deleted file mode 100644 index 31b61c5..0000000 --- a/ipe/db_test.go +++ /dev/null @@ -1,72 +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. - -package ipe - -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) { - app := &app{AppID: "123456", Name: "Example"} - - db := newMemdb() - db.AddApp(app) - - a, err := db.GetAppByAppID("123456") - - if err != nil { - t.Errorf("GetAppByAppID(%q) == %q, want %q", "123456", a, app) - } -} - -func Test_db_GetAppByAppID__error(t *testing.T) { - app := &app{AppID: "123456", Name: "Example"} - - db := newMemdb() - db.AddApp(app) - - a, err := db.GetAppByAppID("not-found") - - if err == nil { - t.Errorf("GetAppByAppID(%q) == %q, want %q", "123456", a, app) - } -} - -func Test_db_GetAppByKey(t *testing.T) { - app := &app{AppID: "123456", Name: "Example", Key: "654321"} - - db := newMemdb() - db.AddApp(app) - - a, err := db.GetAppByKey("654321") - - if err != nil { - t.Errorf("GetAppByKey(%q) == %q, want %q", "654321", a, app) - } -} - -func Test_db_GetAppByKey__error(t *testing.T) { - app := &app{AppID: "123456", Name: "Example", Key: "654321"} - - db := newMemdb() - db.AddApp(app) - - a, err := db.GetAppByKey("not-found") - - if err == nil { - t.Errorf("GetAppByKey(%q) == %q, want %v", "not-found", a, nil) - } -} diff --git a/ipe/errors.go b/ipe/errors.go deleted file mode 100644 index 849f761..0000000 --- a/ipe/errors.go +++ /dev/null @@ -1,124 +0,0 @@ -// 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 ipe - -import "fmt" - -// Base interface -type websocketError interface { - GetCode() int - GetMsg() string -} - -// Base struct -type baseWebsocketError struct { - Code int - Msg string -} - -func (e baseWebsocketError) GetCode() int { - return e.Code -} - -func (e baseWebsocketError) GetMsg() string { - return e.Msg -} - -func (e baseWebsocketError) Error() string { - return fmt.Sprintf("%d: %s", e.Code, e.Msg) -} - -// Unsupprted protocol version -type unsupportedProtocolVersionError struct { - baseWebsocketError -} - -func newUnsupportedProtocolVersionError() unsupportedProtocolVersionError { - return unsupportedProtocolVersionError{ - baseWebsocketError{Code: 4007, Msg: "Unsupported protocol version"}, - } -} - -// The application does not exists -// See the configuration file -type applicationDoesNotExistsError struct { - baseWebsocketError -} - -func newApplicationDoesNotExistsError() applicationDoesNotExistsError { - return applicationDoesNotExistsError{ - baseWebsocketError{Code: 4001, Msg: "Could not found an app with the given key"}, - } -} - -// The user did not send the protocol version -type noProtocolVersionSuppliedError struct { - baseWebsocketError -} - -func newNoProtocolVersionSuppliedError() noProtocolVersionSuppliedError { - return noProtocolVersionSuppliedError{ - baseWebsocketError{Code: 4008, Msg: "No protocol version supplied"}, - } -} - -// When the application is disabled. -// See the configuration file -type applicationDisabledError struct { - baseWebsocketError -} - -func newApplicationDisabledError() noProtocolVersionSuppliedError { - return noProtocolVersionSuppliedError{ - baseWebsocketError{Code: 4003, Msg: "Application disabled"}, - } -} - -// When the application only accepts SSL connections -type applicationOnlyAccepsSSLError struct { - baseWebsocketError -} - -func newApplicationOnlyAccepsSSLError() applicationOnlyAccepsSSLError { - return applicationOnlyAccepsSSLError{ - baseWebsocketError{Code: 4000, Msg: "Application only accepts SSL connections, reconnect using wss://"}, - } -} - -// When the user send an invalid version -type invalidVersionStringFormatError struct { - baseWebsocketError -} - -func newInvalidVersionStringFormatError() invalidVersionStringFormatError { - return invalidVersionStringFormatError{ - baseWebsocketError{Code: 4006, 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: 4200, 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: 0, Msg: msg}, - } -} diff --git a/ipe/ipe.go b/ipe/ipe.go deleted file mode 100644 index b2030f1..0000000 --- a/ipe/ipe.go +++ /dev/null @@ -1,75 +0,0 @@ -// Copyright 2015 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 ipe - -import ( - "encoding/json" - "math/rand" - "net/http" - "os" - "time" - - log "github.com/golang/glog" - "github.com/pressly/chi" - "github.com/pressly/chi/middleware" -) - -// Start Parse the configuration file and starts the ipe server -// It Panic if could not start the HTTP or HTTPS server -func Start(filename string) { - var conf configFile - - rand.Seed(time.Now().Unix()) - file, err := os.Open(filename) - - if err != nil { - log.Error(err) - return - } - - defer file.Close() - - // Reading config - if err := json.NewDecoder(file).Decode(&conf); err != nil { - log.Error(err) - return - } - - // Using a in memory database - db := newMemdb() - - // Adding applications - for _, a := range conf.Apps { - db.AddApp(newAppFromConfig(a)) - } - - r := chi.NewRouter() - r.Use(middleware.Recoverer) - - r.Get("/app/:key", (&websocketHandler{db}).ServeHTTP) - r.Group(func(r chi.Router) { - 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 { - go func() { - log.Infof("Starting HTTPS service on %s ...", conf.SSLHost) - log.Fatal(http.ListenAndServeTLS(conf.SSLHost, conf.SSLCertFile, conf.SSLKeyFile, r)) - }() - } - - log.Infof("Starting HTTP service on %s ...", conf.Host) - log.Fatal(http.ListenAndServe(conf.Host, r)) -} diff --git a/ipe/websocket.go b/ipe/websocket.go deleted file mode 100644 index c821b73..0000000 --- a/ipe/websocket.go +++ /dev/null @@ -1,280 +0,0 @@ -// 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 ipe - -import ( - "encoding/json" - "fmt" - "io" - "net/http" - "strconv" - "strings" - - log "github.com/golang/glog" - "github.com/gorilla/websocket" - "github.com/pressly/chi" - - "github.com/dimiro1/ipe/utils" -) - -var upgrader = websocket.Upgrader{ - ReadBufferSize: 1024, - WriteBufferSize: 1024, - CheckOrigin: func(_ *http.Request) bool { - return true - }, -} - -func handleMessages(conn *websocket.Conn, sessionID string, app *app) { - 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() - p := params.Get("protocol") - - protocol, err := strconv.Atoi(p) - - if err != nil { - return newInvalidVersionStringFormatError() - } - - switch { - case strings.TrimSpace(p) == "": - return newNoProtocolVersionSuppliedError() - case protocol != supportedProtocolVersion: - return newUnsupportedProtocolVersionError() - case app.ApplicationDisabled: - return newApplicationDisabledError() - case app.OnlySSL: - if r.TLS == nil { - return newApplicationOnlyAccepsSSLError() - } - } - - // Create the new Subscriber - connection := newConnection(sessionID, conn) - app.Connect(connection) - - // Everything went fine. Huhu. - if err := conn.WriteJSON(newConnectionEstablishedEvent(connection.SocketID)); err != nil { - return newGenericReconnectImmediatelyError() - } - - return nil -} - -func onClose(sessionID string, app *app) { - app.Disconnect(sessionID) -} - -func onPing(conn *websocket.Conn) { - if err := conn.WriteJSON(newPongEvent()); err != nil { - emitWSError(newGenericReconnectImmediatelyError(), conn) - } -} - -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) - } - - clientEvent := rawEvent{} - - if err := json.Unmarshal(message, &clientEvent); err != nil { - log.Error(err) - emitWSError(newGenericReconnectImmediatelyError(), conn) - return - } - - 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) - return - } - - if err := app.Publish(channel, clientEvent, sessionID); err != nil { - log.Error(err) - emitWSError(newGenericReconnectImmediatelyError(), conn) - return - } -} - -func onUnsubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) { - 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 - } -} - -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} -func (h *websocketHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - conn, err := upgrader.Upgrade(w, r, nil) - defer func() { - if conn != nil { - conn.Close() - } - }() - - if err != nil { - log.Error(err) - return - } - - appKey := chi.URLParam(r, "key") - - app, err := h.DB.GetAppByKey(appKey) - - if err != nil { - log.Error(err) - emitWSError(newApplicationDoesNotExistsError(), conn) - return - } - - sessionID := utils.GenerateSessionID() - - if err := onOpen(conn, r, sessionID, app); err != nil { - emitWSError(err, conn) - return - } - - handleMessages(conn, sessionID, app) -} diff --git a/mocks/socket.go b/mocks/socket.go new file mode 100644 index 0000000..cbba28f --- /dev/null +++ b/mocks/socket.go @@ -0,0 +1,9 @@ +package mocks + +// MockSocket is a mock implementation of Socket +// used in the test suite +type MockSocket struct{} + +func (s MockSocket) WriteJSON(i interface{}) error { + return nil +} diff --git a/storage/storage.go b/storage/storage.go new file mode 100644 index 0000000..c260442 --- /dev/null +++ b/storage/storage.go @@ -0,0 +1,64 @@ +// 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. + +package storage + +import ( + "errors" + "ipe/app" + "sync" +) + +// storage represents a app database +// For now it there is only one memory database implementation +// but in the future I can write a sql implementation +type Storage interface { + GetAppByAppID(appID string) (*app.Application, error) + GetAppByKey(key string) (*app.Application, error) + AddApp(application *app.Application) error +} + +type InMemory struct { + sync.RWMutex + Apps []*app.Application +} + +func NewInMemory() Storage { + return &InMemory{} +} + +func (db *InMemory) AddApp(application *app.Application) error { + db.Lock() + defer db.Unlock() + + db.Apps = append(db.Apps, application) + return nil +} + +// GetAppByAppID returns an App with by appID +func (db *InMemory) GetAppByAppID(appID string) (*app.Application, error) { + db.RLock() + defer db.RUnlock() + + for _, a := range db.Apps { + if a.AppID == appID { + return a, nil + } + } + + return nil, errors.New("app not found") +} + +// GetAppByKey returns an App with by key +func (db *InMemory) GetAppByKey(key string) (*app.Application, error) { + db.RLock() + defer db.RUnlock() + + for _, a := range db.Apps { + if a.Key == key { + return a, nil + } + } + return nil, errors.New("app not found") +} diff --git a/storage/storage_test.go b/storage/storage_test.go new file mode 100644 index 0000000..b7b27b2 --- /dev/null +++ b/storage/storage_test.go @@ -0,0 +1,75 @@ +// 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. + +package storage + +import ( + "ipe/app" + "testing" +) + +func Benchmark_memdb_GetAppByAppID(b *testing.B) { + storage := NewInMemory() + _ = storage.AddApp(&app.Application{AppID: "123456", Name: "Example"}) + _ = storage.AddApp(&app.Application{AppID: "654321", Name: "Example2"}) + _ = storage.AddApp(&app.Application{AppID: "678901", Name: "Example3"}) + + b.ResetTimer() + + for i := 0; i < b.N; i++ { + _, _ = storage.GetAppByAppID("123456") + } +} + +func Test_db_GetAppByAppID(t *testing.T) { + _app := &app.Application{AppID: "123456", Name: "Example"} + + storage := NewInMemory() + _ = storage.AddApp(_app) + + a, err := storage.GetAppByAppID("123456") + + if err != nil { + t.Errorf("GetAppByAppID(%q) == %+v, want %+v", "123456", a, _app) + } +} + +func Test_db_GetAppByAppID__error(t *testing.T) { + _app := &app.Application{AppID: "123456", Name: "Example"} + + storage := NewInMemory() + _ = storage.AddApp(_app) + + a, err := storage.GetAppByAppID("not-found") + + if err == nil { + t.Errorf("GetAppByAppID(%q) == %+v, want %+v", "123456", a, _app) + } +} + +func Test_db_GetAppByKey(t *testing.T) { + _app := &app.Application{AppID: "123456", Name: "Example", Key: "654321"} + + storage := NewInMemory() + _ = storage.AddApp(_app) + + a, err := storage.GetAppByKey("654321") + + if err != nil { + t.Errorf("GetAppByKey(%q) == %+v, want %+v", "654321", a, _app) + } +} + +func Test_db_GetAppByKey__error(t *testing.T) { + _app := &app.Application{AppID: "123456", Name: "Example", Key: "654321"} + + storage := NewInMemory() + _ = storage.AddApp(_app) + + a, err := storage.GetAppByKey("not-found") + + if err == nil { + t.Errorf("GetAppByKey(%q) == %+v, want %+v", "not-found", a, nil) + } +} diff --git a/ipe/subscription.go b/subscription/subscription.go similarity index 54% rename from ipe/subscription.go rename to subscription/subscription.go index 45d12f3..cb7e677 100644 --- a/ipe/subscription.go +++ b/subscription/subscription.go @@ -2,16 +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 subscription + +import "ipe/connection" // A Channel Subscription -type subscription struct { - Connection *connection +type Subscription struct { + Connection *connection.Connection ID string Data string } // Create a new Subscription -func newSubscription(conn *connection, data string) *subscription { - return &subscription{Connection: conn, Data: data} +func New(conn *connection.Connection, data string) *Subscription { + return &Subscription{Connection: conn, Data: data} } diff --git a/websockets/errors.go b/websockets/errors.go new file mode 100644 index 0000000..bacb06a --- /dev/null +++ b/websockets/errors.go @@ -0,0 +1,20 @@ +// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package websockets + +type websocketError struct { + Code int + Msg string +} + +var ( + applicationOnlyAcceptsSSL = &websocketError{Code: 4000, Msg: "Application only accepts SSL connections, reconnect using wss://"} + applicationDoesNotExists = &websocketError{Code: 4001, Msg: "Could not found an app with the given key"} + applicationDisabled = &websocketError{Code: 4003, Msg: "Application disabled"} + invalidVersionStringFormat = &websocketError{Code: 4006, Msg: "Invalid version string format"} + unsupportedProtocolVersion = &websocketError{Code: 4007, Msg: "Unsupported protocol version"} + noProtocolVersionSupplied = &websocketError{Code: 4008, Msg: "No protocol version supplied"} + reconnectImmediately = &websocketError{Code: 4200, Msg: "Generic reconnect immediately"} +) diff --git a/websockets/websocket.go b/websockets/websocket.go new file mode 100644 index 0000000..87ff180 --- /dev/null +++ b/websockets/websocket.go @@ -0,0 +1,294 @@ +// Copyright 2014 Claudemiro Alves Feitosa Neto. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package websockets + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + + log "github.com/golang/glog" + "github.com/gorilla/mux" + "github.com/gorilla/websocket" + + "ipe/app" + "ipe/connection" + "ipe/events" + "ipe/storage" + "ipe/utils" +) + +// Only this version is supported +const supportedProtocolVersion = 7 + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, + CheckOrigin: func(_ *http.Request) bool { + return true + }, +} + +type Websocket struct { + storage storage.Storage +} + +func NewWebsocket(storage storage.Storage) *Websocket { + return &Websocket{storage: storage} +} + +// Websocket GET /app/{key} +func (h *Websocket) ServeHTTP(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + defer func() { + if conn != nil { + if err := conn.Close(); err != nil { + log.Errorf("closing the websocket connection %+v", err) + } + } + }() + + if err != nil { + log.Error(err) + return + } + + var ( + pathVars = mux.Vars(r) + appKey = pathVars["key"] + ) + + _app, err := h.storage.GetAppByKey(appKey) + + if err != nil { + log.Error(err) + emitError(applicationDoesNotExists, conn) + return + } + + sessionID := utils.GenerateSessionID() + + if err := onOpen(conn, r, sessionID, _app); err != nil { + emitError(err, conn) + return + } + + handleMessages(conn, sessionID, _app) +} + +func handleMessages(conn *websocket.Conn, sessionID string, app *app.Application) { + 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 { + emitError(reconnectImmediately, conn) + return + } + + log.Infof("websocket: Handling %s event", event.Event) + + switch event.Event { + case "pusher:ping": + handlePing(conn) + case "pusher:subscribe": + handleSubscribe(conn, sessionID, app, message) + case "pusher:unsubscribe": + handleUnsubscribe(conn, sessionID, app, message) + default: + if utils.IsClientEvent(event.Event) { + handleClientEvent(conn, sessionID, app, message) + } + } + } +} + +// Emit an Websocket ErrorEvent +func emitError(err *websocketError, conn *websocket.Conn) { + event := events.NewError(err.Code, err.Msg) + + if err := conn.WriteJSON(event); err != nil { + log.Error(err) + } +} + +func handleError(conn *websocket.Conn, sessionID string, app *app.Application, err error) { + log.Errorf("%+v", err) + if err == io.EOF { + onClose(sessionID, app) + } else if _, ok := err.(*websocket.CloseError); ok { + onClose(sessionID, app) + } else { + emitError(reconnectImmediately, conn) + } +} + +func onOpen(conn *websocket.Conn, r *http.Request, sessionID string, app *app.Application) *websocketError { + var ( + queryVars = r.URL.Query() + strProtocol = queryVars.Get("protocol") + ) + + protocol, err := strconv.Atoi(strProtocol) + if err != nil { + return invalidVersionStringFormat + } + + switch { + case strings.TrimSpace(strProtocol) == "": + return noProtocolVersionSupplied + case protocol != supportedProtocolVersion: + return unsupportedProtocolVersion + case app.ApplicationDisabled: + return applicationDisabled + case app.OnlySSL: + if r.TLS == nil { + return applicationOnlyAcceptsSSL + } + } + + // Create the new Subscriber + _connection := connection.New(sessionID, conn) + app.Connect(_connection) + + // Everything went fine. + if err := conn.WriteJSON(events.NewConnectionEstablished(_connection.SocketID)); err != nil { + return reconnectImmediately + } + + return nil +} + +func onClose(sessionID string, app *app.Application) { + app.Disconnect(sessionID) +} + +func handlePing(conn *websocket.Conn) { + if err := conn.WriteJSON(events.NewPong()); err != nil { + emitError(reconnectImmediately, conn) + } +} + +func handleClientEvent(conn *websocket.Conn, sessionID string, app *app.Application, message []byte) { + if !app.UserEvents { + emitError(&websocketError{Code: 0, Msg: "To send client events, you must enable this feature in the Settings."}, conn) + } + + clientEvent := events.Raw{} + + if err := json.Unmarshal(message, &clientEvent); err != nil { + log.Error(err) + emitError(reconnectImmediately, conn) + return + } + + channel, err := app.FindChannelByChannelID(clientEvent.Channel) + + if err != nil { + emitError(&websocketError{Code: 0, Msg: fmt.Sprintf("Could not find a channel with the id %s", clientEvent.Channel)}, conn) + return + } + + if !channel.IsPresenceOrPrivate() { + emitError(&websocketError{Code: 0, Msg: "Client event rejected - only supported on private and presence channels"}, conn) + return + } + + if err := app.Publish(channel, clientEvent, sessionID); err != nil { + log.Error(err) + emitError(reconnectImmediately, conn) + return + } +} + +func handleUnsubscribe(conn *websocket.Conn, sessionID string, app *app.Application, message []byte) { + unsubscribeEvent := events.Unsubscribe{} + + if err := json.Unmarshal(message, &unsubscribeEvent); err != nil { + emitError(reconnectImmediately, conn) + return + } + + _connection, err := app.FindConnection(sessionID) + + if err != nil { + emitError(&websocketError{Code: 0, Msg: fmt.Sprintf("Could not find a connection with the id %s", sessionID)}, conn) + return + } + + channel, err := app.FindChannelByChannelID(unsubscribeEvent.Data.Channel) + + if err != nil { + emitError(&websocketError{Code: 0, Msg: fmt.Sprintf("Could not find a channel with the id %s", unsubscribeEvent.Data.Channel)}, conn) + return + } + + if err := app.Unsubscribe(channel, _connection); err != nil { + emitError(reconnectImmediately, conn) + return + } +} + +func handleSubscribe(conn *websocket.Conn, sessionID string, app *app.Application, message []byte) { + subscribeEvent := events.Subscribe{} + + if err := json.Unmarshal(message, &subscribeEvent); err != nil { + emitError(reconnectImmediately, conn) + return + } + + _connection, err := app.FindConnection(sessionID) + + if err != nil { + emitError(reconnectImmediately, conn) + return + } + + channelName := strings.TrimSpace(subscribeEvent.Data.Channel) + + if !utils.IsChannelNameValid(channelName) { + emitError(&websocketError{Code: 0, Msg: "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) { + emitError(&websocketError{Code: 0, Msg: 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 { + emitError(reconnectImmediately, conn) + } +} + +func validateAuthKey(givenAuthKey string, toSign []string, app *app.Application) bool { + expectedAuthKey := fmt.Sprintf("%s:%s", app.Key, utils.HashMAC([]byte(strings.Join(toSign, ":")), []byte(app.Secret))) + return givenAuthKey == expectedAuthKey +}