Basic context implementation in webhooks.

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() {

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
+	}
 }

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(
+func handleMessages(conn *websocket.Conn, sessionID string, app *app) {
-	conn *websocket.Conn, w http.ResponseWriter,
-	r *http.Request, 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(
+func onOpen(conn *websocket.Conn, r *http.Request, sessionID string, app *app) error {
-	conn *websocket.Conn, w http.ResponseWriter,
-	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(
+func onClientEvent(conn *websocket.Conn, sessionID string, app *app, message []byte) {
-	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(
+func onUnsubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) {
-	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(
+func onSubscribe(conn *websocket.Conn, sessionID string, app *app, message []byte) {
-	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)
 }