Add timeouts to requests. Start fleshing out RequestPool.

src/github.com/matrix-org/dendrite/clientapi/sync/requestpool.go

@@ -0,0 +1,99 @@
+package sync
+
+import (
+	"net/http"
+	"strconv"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/matrix-org/dendrite/clientapi/auth"
+	"github.com/matrix-org/dendrite/clientapi/httputil"
+	"github.com/matrix-org/dendrite/clientapi/storage"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+)
+
+const defaultSyncTimeout = time.Duration(30) * time.Second
+
+type syncRequest struct {
+	userID        string
+	timeout       time.Duration
+	since         string
+	wantFullState bool
+}
+
+// RequestPool manages HTTP long-poll connections for /sync
+type RequestPool struct {
+	db *storage.SyncServerDatabase
+}
+
+// OnIncomingSyncRequest is called when a client makes a /sync request. This function MUST be
+// called in a dedicated goroutine for this request. This function will block the goroutine
+// until a response is ready, or it times out.
+func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request) util.JSONResponse {
+	// Extract values from request
+	logger := util.GetLogger(req.Context())
+	userID, resErr := auth.VerifyAccessToken(req)
+	if resErr != nil {
+		return *resErr
+	}
+	since := req.URL.Query().Get("since")
+	timeout := getTimeout(req.URL.Query().Get("timeout"))
+	wantFullState := req.URL.Query().Get("full_state") != ""
+	// TODO: Additional query params: set_presence, filter
+	syncReq := syncRequest{
+		userID:        userID,
+		timeout:       timeout,
+		since:         since,
+		wantFullState: wantFullState,
+	}
+	logger.WithFields(log.Fields{
+		"userID":  userID,
+		"since":   since,
+		"timeout": timeout,
+	}).Info("Incoming /sync request")
+
+	res, err := rp.currentSyncForUser(syncReq)
+	if err != nil {
+		return httputil.LogThenError(req, err)
+	}
+	return util.JSONResponse{
+		Code: 200,
+		JSON: res,
+	}
+}
+
+// OnNewEvent is called when a new event is received from the room server
+func (rp *RequestPool) OnNewEvent(ev *gomatrixserverlib.Event) {
+
+}
+
+func (rp *RequestPool) currentSyncForUser(req syncRequest) ([]gomatrixserverlib.Event, error) {
+	// https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L179
+	// Check if we are going to return immediately and if so, calculate the current
+	// sync for this user and return.
+	if req.since == "" || req.timeout == time.Duration(0) || req.wantFullState {
+
+	}
+
+	// wait for an event which affects this user or one of their rooms, then recheck for new
+	// sync data.
+
+	return nil, nil
+}
+
+func getTimeout(timeoutMS string) time.Duration {
+	if timeoutMS == "" {
+		return defaultSyncTimeout
+	}
+	i, err := strconv.Atoi(timeoutMS)
+	if err != nil {
+		return defaultSyncTimeout
+	}
+	return time.Duration(i) * time.Millisecond
+}
+
+// NewRequestPool makes a new RequestPool
+func NewRequestPool(db *storage.SyncServerDatabase) RequestPool {
+	return RequestPool{db}
+}

src/github.com/matrix-org/dendrite/clientapi/sync/syncrequestpool.go

@@ -1,33 +0,0 @@
-package sync
-
-import (
-	"net/http"
-
-	"github.com/matrix-org/dendrite/clientapi/storage"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/matrix-org/util"
-)
-
-// RequestPool manages HTTP long-poll connections for /sync
-type RequestPool struct {
-	db    *storage.SyncServerDatabase
-	conns map[string]chan interface{}
-}
-
-// OnIncomingSyncRequest is called when a client makes a /sync request
-func (srp *RequestPool) OnIncomingSyncRequest(req *http.Request) util.JSONResponse {
-	return util.JSONResponse{
-		Code: 200,
-		JSON: struct{}{},
-	}
-}
-
-// OnNewEvent is called when a new event is received from the room server
-func (srp *RequestPool) OnNewEvent(ev *gomatrixserverlib.Event) {
-
-}
-
-// NewRequestPool makes a new RequestPool
-func NewRequestPool(db *storage.SyncServerDatabase) RequestPool {
-	return RequestPool{db, make(map[string]chan interface{})}
-}