From 4c5d5302ec4d02853e5e5a31a1ad3e1a7bd280ee Mon Sep 17 00:00:00 2001 From: Claudemiro Date: Sun, 21 Aug 2016 10:22:35 -0300 Subject: [PATCH] Basic context implementation in webhooks. --- functional/SpecRunner.html | 24 ++++++++--------- functional/client.go | 8 +++--- functional/functional-config.json | 36 ++++++++++++------------- ipe/config-example.json | 36 ++++++++++++------------- ipe/webhooks.go | 45 ++++++++++++++++++++++++++----- ipe/websocket.go | 38 +++++++++++--------------- 6 files changed, 105 insertions(+), 82 deletions(-) diff --git a/functional/SpecRunner.html b/functional/SpecRunner.html index dd1359a..43d0378 100644 --- a/functional/SpecRunner.html +++ b/functional/SpecRunner.html @@ -1,24 +1,24 @@ - - Pusher Spec - + + Pusher Spec + -
+
- - - + + + - + - - - + + + diff --git a/functional/client.go b/functional/client.go index 05e9388..505ee3c 100644 --- a/functional/client.go +++ b/functional/client.go @@ -34,7 +34,7 @@ func pusherPresenceAuth(res http.ResponseWriter, req *http.Request) { panic(err) } - fmt.Fprintf(res, string(response)) + fmt.Fprint(res, string(response)) } func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) { @@ -48,13 +48,13 @@ func pusherPrivateAuth(res http.ResponseWriter, req *http.Request) { panic(err) } - fmt.Fprintf(res, string(response)) + fmt.Fprint(res, string(response)) } -func triggerMessage(res http.ResponseWriter, req *http.Request) { +func triggerMessage(res http.ResponseWriter, _ *http.Request) { client.Trigger("private-messages", "messages", "The message from server") - fmt.Fprintf(res, "OK") + fmt.Fprint(res, "OK") } func main() { diff --git a/functional/functional-config.json b/functional/functional-config.json index 9900c3d..08e9bcf 100644 --- a/functional/functional-config.json +++ b/functional/functional-config.json @@ -1,20 +1,20 @@ { - "Host": ":8080", - "Encrypted": false, - "SSLHost": ":8090", - "SSLKeyFile": "key.pem", - "SSLCertFile": "cert.pem", - "Apps": [ - { - "ApplicationDisabled": false, - "OnlySSL": false, - "Secret": "7ad3753142a6693b25b9", - "Key": "278d525bdf162c739803", - "Name": "App for Functional Test", - "AppID": "1", - "UserEvents": true, - "WebHooks": false, - "URLWebHook": "http://127.0.0.1:4567/php/hook.php" - } - ] + "Host": ":8080", + "Encrypted": false, + "SSLHost": ":8090", + "SSLKeyFile": "key.pem", + "SSLCertFile": "cert.pem", + "Apps": [ + { + "ApplicationDisabled": false, + "OnlySSL": false, + "Secret": "7ad3753142a6693b25b9", + "Key": "278d525bdf162c739803", + "Name": "App for Functional Test", + "AppID": "1", + "UserEvents": true, + "WebHooks": false, + "URLWebHook": "http://127.0.0.1:4567/php/hook.php" + } + ] } diff --git a/ipe/config-example.json b/ipe/config-example.json index 201149d..f209d5f 100644 --- a/ipe/config-example.json +++ b/ipe/config-example.json @@ -1,20 +1,20 @@ { - "Host": ":8080", - "SSL": false, - "SSLHost": ":4433", - "SSLKeyFile": "A key.pem file", - "SSLCertFile": "A cert.pem file", - "Apps": [ - { - "ApplicationDisabled": false, - "Secret": "A really secret random string", - "Key": "A random Key string", - "OnlySSL": false, - "Name": "The app name", - "AppID": "The app ID", - "UserEvents": true, - "WebHooks": true, - "URLWebHook": "Some URL to send webhooks" - } - ] + "Host": ":8080", + "SSL": false, + "SSLHost": ":4433", + "SSLKeyFile": "A key.pem file", + "SSLCertFile": "A cert.pem file", + "Apps": [ + { + "ApplicationDisabled": false, + "Secret": "A really secret random string", + "Key": "A random Key string", + "OnlySSL": false, + "Name": "The app name", + "AppID": "The app ID", + "UserEvents": true, + "WebHooks": true, + "URLWebHook": "Some URL to send webhooks" + } + ] } diff --git a/ipe/webhooks.go b/ipe/webhooks.go index e5250df..1d19093 100644 --- a/ipe/webhooks.go +++ b/ipe/webhooks.go @@ -10,10 +10,14 @@ import ( "net/http" "time" + "context" + "fmt" "github.com/dimiro1/ipe/utils" log "github.com/golang/glog" ) +const maxTimeout = 3 * time.Second + // A WebHook is sent as a HTTP POST request to the url which you specify. // The POST request payload (body) contains a JSON document, and follows the following format: // { @@ -71,14 +75,19 @@ func newClientHook(channel *channel, s *subscription, event string, data interfa // { "name": "channel_occupied", "channel": "test_channel" } func (a *app) TriggerChannelOccupiedHook(c *channel) { event := newChannelOcuppiedHook(c) - triggerHook(event.Name, a, c, event) + ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) + defer cancel() + + triggerHook(ctx, event.Name, a, c, event) } // channel_vacated // { "name": "channel_vacated", "channel": "test_channel" } func (a *app) TriggerChannelVacatedHook(c *channel) { event := newChannelVacatedHook(c) - triggerHook(event.Name, a, c, event) + ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) + defer cancel() + triggerHook(ctx, event.Name, a, c, event) } // { @@ -96,7 +105,9 @@ func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent st event.UserID = s.ID } - triggerHook(event.Name, a, c, event) + ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) + defer cancel() + triggerHook(ctx, event.Name, a, c, event) } // { @@ -106,7 +117,9 @@ func (a *app) TriggerClientEventHook(c *channel, s *subscription, clientEvent st // } func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) { event := newMemberAddedHook(c, s) - triggerHook(event.Name, a, c, event) + ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) + defer cancel() + triggerHook(ctx, event.Name, a, c, event) } // { @@ -116,15 +129,20 @@ func (a *app) TriggerMemberAddedHook(c *channel, s *subscription) { // } func (a *app) TriggerMemberRemovedHook(c *channel, s *subscription) { event := newMemberRemovedHook(c, s) - triggerHook(event.Name, a, c, event) + ctx, cancel := context.WithTimeout(context.Background(), maxTimeout) + defer cancel() + triggerHook(ctx, event.Name, a, c, event) } -func triggerHook(name string, a *app, c *channel, event hookEvent) { +func triggerHook(ctx context.Context, name string, a *app, _ *channel, event hookEvent) error { if !a.WebHooks { log.Infof("Webhooks are not enabled for app: %s", a.Name) - return + return fmt.Errorf("Webhooks are not enabled for app: %s", a.Name) } + var done chan (bool) + defer close(done) + go func() { log.Infof("Triggering %s event", name) @@ -145,11 +163,14 @@ func triggerHook(name string, a *app, c *channel, event hookEvent) { var req *http.Request req, err = http.NewRequest("POST", a.URLWebHook, bytes.NewReader(js)) + if err != nil { log.Errorf("Error creating request: %+v", err) return } + req.WithContext(ctx) + req.Header.Set("User-Agent", "Ipe UA; (+https://github.com/dimiro1/ipe)") req.Header.Set("Content-Type", "application/json") req.Header.Set("X-Pusher-Key", a.Key) @@ -168,5 +189,15 @@ func triggerHook(name string, a *app, c *channel, event hookEvent) { if err != nil { log.Errorf("Error posting %s event: %+v", name, err) } + + // Successfully terminated + done <- true }() + + select { + case <-ctx.Done(): + return ctx.Err() + case <-done: + return nil + } } diff --git a/ipe/websocket.go b/ipe/websocket.go index 313acd0..8116f4c 100644 --- a/ipe/websocket.go +++ b/ipe/websocket.go @@ -26,13 +26,12 @@ import ( var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, - CheckOrigin: func(r *http.Request) bool { return true }, + CheckOrigin: func(_ *http.Request) bool { + return true + }, } -func handleMessages( - conn *websocket.Conn, w http.ResponseWriter, - r *http.Request, sessionID string, app *app) { - +func handleMessages(conn *websocket.Conn, sessionID string, app *app) { var event struct { Event string `json:"event"` } @@ -64,7 +63,7 @@ func handleMessages( onClientEvent(conn, sessionID, app, message) } } - } // For + } } func handleError(conn *websocket.Conn, sessionID string, app *app, err error) { @@ -78,10 +77,7 @@ func handleError(conn *websocket.Conn, sessionID string, app *app, err error) { } } -func onOpen( - conn *websocket.Conn, w http.ResponseWriter, - r *http.Request, sessionID string, app *app) error { - +func onOpen(conn *websocket.Conn, r *http.Request, sessionID string, app *app) error { params := r.URL.Query() p := params.Get("protocol") @@ -126,9 +122,7 @@ func onPing(conn *websocket.Conn) { } } -func onClientEvent( - conn *websocket.Conn, sessionID string, app *app, message []byte) { - +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) } @@ -159,9 +153,7 @@ func onClientEvent( } } -func onUnsubscribe( - conn *websocket.Conn, sessionID string, app *app, message []byte) { - +func onUnsubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) { unsubscribeEvent := unsubscribeEvent{} if err := json.Unmarshal(message, &unsubscribeEvent); err != nil { @@ -186,9 +178,7 @@ func onUnsubscribe( } } -func onSubscribe( - conn *websocket.Conn, sessionID string, app *app, message []byte) { - +func onSubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) { subscribeEvent := subscribeEvent{} if err := json.Unmarshal(message, &subscribeEvent); err != nil { @@ -206,7 +196,7 @@ func onSubscribe( channelName := strings.TrimSpace(subscribeEvent.Data.Channel) if !utils.IsChannelNameValid(channelName) { - emitWSError(newGenericError(fmt.Sprintf("This channel name is not valid")), conn) + emitWSError(newGenericError("This channel name is not valid"), conn) return } @@ -259,7 +249,9 @@ func newWebsocketHandler(DB db) goji.Handler { return &websocketHandler{DB} } -type websocketHandler struct{ DB db } +type websocketHandler struct { + DB db +} // Websocket GET /app/{key} func (h *websocketHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter, r *http.Request) { @@ -287,10 +279,10 @@ func (h *websocketHandler) ServeHTTPC(ctx context.Context, w http.ResponseWriter sessionID := utils.GenerateSessionID() - if err := onOpen(conn, w, r, sessionID, app); err != nil { + if err := onOpen(conn, r, sessionID, app); err != nil { emitWSError(err, conn) return } - handleMessages(conn, w, r, sessionID, app) + handleMessages(conn, sessionID, app) }