Add structs for HTTP long-polling
DELETE
src/github.com/matrix-org/dendrite/clientapi/readers/sync.go
-package readers
-
-import (
- "net/http"
-
- "github.com/matrix-org/dendrite/clientapi/auth"
- "github.com/matrix-org/util"
-)
-
-// Sync implements /sync
-func Sync(req *http.Request) util.JSONResponse {
- logger := util.GetLogger(req.Context())
- userID, resErr := auth.VerifyAccessToken(req)
- if resErr != nil {
- return *resErr
- }
-
- logger.WithField("userID", userID).Info("Doing stuff...")
- return util.MessageResponse(404, "Not implemented yet")
-}

MODIFY
src/github.com/matrix-org/dendrite/clientapi/routing/routing.go
src/github.com/matrix-org/dendrite/clientapi/routing/routing.go
@@ -6,7 +6,7 @@ import (
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/config"
"github.com/matrix-org/dendrite/clientapi/producers"
- "github.com/matrix-org/dendrite/clientapi/readers"
+ "github.com/matrix-org/dendrite/clientapi/sync"
"github.com/matrix-org/dendrite/clientapi/writers"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/util"
@@ -49,11 +49,11 @@ func Setup(servMux *http.ServeMux, httpClient *http.Client, cfg config.ClientAPI
}
// SetupSyncServerListeners configures the given mux with sync-server listeners
-func SetupSyncServerListeners(servMux *http.ServeMux, httpClient *http.Client, cfg config.Sync) {
+func SetupSyncServerListeners(servMux *http.ServeMux, httpClient *http.Client, cfg config.Sync, srp sync.RequestPool) {
apiMux := mux.NewRouter()
r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
r0mux.Handle("/sync", make("sync", util.NewJSONRequestHandler(func(req *http.Request) util.JSONResponse {
- return readers.Sync(req)
+ return srp.OnIncomingSyncRequest(req)
})))
servMux.Handle("/metrics", prometheus.Handler())
servMux.Handle("/api/", http.StripPrefix("/api", apiMux))

MODIFY
src/github.com/matrix-org/dendrite/clientapi/storage/output_room_events_table.go
src/github.com/matrix-org/dendrite/clientapi/storage/output_room_events_table.go
@@ -72,7 +72,7 @@ func (s *outputRoomEventsStatements) Events(txn *sql.Tx, eventIDs []string) ([]g
}
defer rows.Close()
- result := make([]gomatrixserverlib.Event, len(eventIDs))
+ var result []gomatrixserverlib.Event
i := 0
for ; rows.Next(); i++ {
var eventBytes []byte
@@ -86,7 +86,7 @@ func (s *outputRoomEventsStatements) Events(txn *sql.Tx, eventIDs []string) ([]g
result = append(result, ev)
}
if i != len(eventIDs) {
- return nil, fmt.Errorf("failed to map all event IDs to events: (%d != %d)", i, len(eventIDs))
+ return nil, fmt.Errorf("failed to map all event IDs to events: (got %d, wanted %d)", i, len(eventIDs))
}
return result, nil
}

ADD
src/github.com/matrix-org/dendrite/clientapi/sync/requestpool.go
+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"))
+ fullState := req.URL.Query().Get("full_state")
+ wantFullState := fullState != "" && fullState != "false"
+ // 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 {
+ return []gomatrixserverlib.Event{}, nil
+ }
+
+ // TODO: wait for an event which affects this user or one of their rooms, then recheck for new
+ // sync data.
+ time.Sleep(req.timeout)
+
+ 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}
+}

MODIFY
src/github.com/matrix-org/dendrite/clientapi/sync/syncserver.go
src/github.com/matrix-org/dendrite/clientapi/sync/syncserver.go
@@ -16,10 +16,11 @@ import (
type Server struct {
roomServerConsumer *common.ContinualConsumer
db *storage.SyncServerDatabase
+ rp RequestPool
}
// NewServer creates a new sync server. Call Start() to begin consuming from room servers.
-func NewServer(cfg *config.Sync, store *storage.SyncServerDatabase) (*Server, error) {
+func NewServer(cfg *config.Sync, rp RequestPool, store *storage.SyncServerDatabase) (*Server, error) {
kafkaConsumer, err := sarama.NewConsumer(cfg.KafkaConsumerURIs, nil)
if err != nil {
return nil, err
@@ -33,6 +34,7 @@ func NewServer(cfg *config.Sync, store *storage.SyncServerDatabase) (*Server, er
s := &Server{
roomServerConsumer: &consumer,
db: store,
+ rp: rp,
}
consumer.ProcessMessage = s.onMessage
@@ -73,6 +75,7 @@ func (s *Server) onMessage(msg *sarama.ConsumerMessage) error {
}).Panicf("roomserver output log: write event failure")
return nil
}
+ s.rp.OnNewEvent(&ev)
return nil
}

MODIFY
src/github.com/matrix-org/dendrite/cmd/dendrite-sync-server/main.go
src/github.com/matrix-org/dendrite/cmd/dendrite-sync-server/main.go
@@ -74,7 +74,9 @@ func main() {
log.Panicf("startup: failed to create sync server database with data source %s : %s", cfg.DataSource, err)
}
- server, err := sync.NewServer(cfg, db)
+ rp := sync.NewRequestPool(db)
+
+ server, err := sync.NewServer(cfg, rp, db)
if err != nil {
log.Panicf("startup: failed to create sync server: %s", err)
}
@@ -83,6 +85,6 @@ func main() {
}
log.Info("Starting sync server on ", *bindAddr)
- routing.SetupSyncServerListeners(http.DefaultServeMux, http.DefaultClient, *cfg)
+ routing.SetupSyncServerListeners(http.DefaultServeMux, http.DefaultClient, *cfg, rp)
log.Fatal(http.ListenAndServe(*bindAddr, nil))
}

Add structs for HTTP long-polling (#56)