Add SyncServer and remove needless abstractions

src/github.com/matrix-org/dendrite/clientapi/consumers/roomserver.go

@@ -1,29 +0,0 @@
-package consumers
-
-import (
-	"github.com/matrix-org/dendrite/clientapi/config"
-	"github.com/matrix-org/dendrite/common"
-	sarama "gopkg.in/Shopify/sarama.v1"
-)
-
-// RoomserverConsumer consumes events from the room server
-type RoomserverConsumer struct {
-	Consumer common.ContinualConsumer
-}
-
-// NewRoomserverConsumer creates a new roomserver consumer
-func NewRoomserverConsumer(cfg *config.Sync, store common.PartitionStorer) (*RoomserverConsumer, error) {
-	kafkaConsumer, err := sarama.NewConsumer(cfg.KafkaConsumerURIs, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	return &RoomserverConsumer{
-		Consumer: common.ContinualConsumer{
-			Topic:          cfg.RoomserverOutputTopic,
-			Consumer:       kafkaConsumer,
-			PartitionStore: store,
-		},
-	}, nil
-
-}

src/github.com/matrix-org/dendrite/clientapi/routing/routing.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/clientapi/config"
-	"github.com/matrix-org/dendrite/clientapi/consumers"
 	"github.com/matrix-org/dendrite/clientapi/producers"
 	"github.com/matrix-org/dendrite/clientapi/readers"
 	"github.com/matrix-org/dendrite/clientapi/writers"
@@ -49,8 +48,8 @@ func Setup(servMux *http.ServeMux, httpClient *http.Client, cfg config.ClientAPI
 	servMux.Handle("/api/", http.StripPrefix("/api", apiMux))
 }
 
-// SetupSyncServer configures the given mux with sync-server listeners
+// SetupSyncServerListeners configures the given mux with sync-server listeners
-func SetupSyncServer(servMux *http.ServeMux, httpClient *http.Client, cfg config.Sync, consumer *consumers.RoomserverConsumer) {
+func SetupSyncServerListeners(servMux *http.ServeMux, httpClient *http.Client, cfg config.Sync) {
 	apiMux := mux.NewRouter()
 	r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
 	r0mux.Handle("/sync", make("sync", util.NewJSONRequestHandler(func(req *http.Request) util.JSONResponse {

src/github.com/matrix-org/dendrite/clientapi/storage/syncserver.go

@@ -1,19 +1,21 @@
-package sync
+package storage
 
 import (
 	"database/sql"
 
 	"github.com/matrix-org/dendrite/common"
+	// Import the postgres database driver.
+	_ "github.com/lib/pq"
 )
 
-// Database represents a sync server database
+// SyncServerDatabase represents a sync server database
-type Database struct {
+type SyncServerDatabase struct {
 	db         *sql.DB
 	partitions common.PartitionOffsetStatements
 }
 
-// NewDatabase creates a new sync server database
+// NewSyncServerDatabase creates a new sync server database
-func NewDatabase(dataSourceName string) (*Database, error) {
+func NewSyncServerDatabase(dataSourceName string) (*SyncServerDatabase, error) {
 	var db *sql.DB
 	var err error
 	if db, err = sql.Open("postgres", dataSourceName); err != nil {
@@ -23,15 +25,15 @@ func NewDatabase(dataSourceName string) (*Database, error) {
 	if err := partitions.Prepare(db); err != nil {
 		return nil, err
 	}
-	return &Database{db, partitions}, nil
+	return &SyncServerDatabase{db, partitions}, nil
 }
 
 // PartitionOffsets implements common.PartitionStorer
-func (d *Database) PartitionOffsets(topic string) ([]common.PartitionOffset, error) {
+func (d *SyncServerDatabase) PartitionOffsets(topic string) ([]common.PartitionOffset, error) {
 	return d.partitions.SelectPartitionOffsets(topic)
 }
 
 // SetPartitionOffset implements common.PartitionStorer
-func (d *Database) SetPartitionOffset(topic string, partition int32, offset int64) error {
+func (d *SyncServerDatabase) SetPartitionOffset(topic string, partition int32, offset int64) error {
 	return d.partitions.UpsertPartitionOffset(topic, partition, offset)
 }

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

@@ -0,0 +1,34 @@
+package sync
+
+import (
+	"github.com/matrix-org/dendrite/clientapi/config"
+	"github.com/matrix-org/dendrite/common"
+	sarama "gopkg.in/Shopify/sarama.v1"
+)
+
+// Server contains all the logic for running a sync server
+type Server struct {
+	roomServerConsumer *common.ContinualConsumer
+}
+
+// NewServer creates a new sync server. Call Start() to begin consuming from room servers.
+func NewServer(cfg *config.Sync, store common.PartitionStorer) (*Server, error) {
+	kafkaConsumer, err := sarama.NewConsumer(cfg.KafkaConsumerURIs, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Server{
+		roomServerConsumer: &common.ContinualConsumer{
+			Topic:          cfg.RoomserverOutputTopic,
+			Consumer:       kafkaConsumer,
+			PartitionStore: store,
+		},
+	}, nil
+
+}
+
+// Start consuming from room servers
+func (s *Server) Start() error {
+	return s.roomServerConsumer.Start()
+}

src/github.com/matrix-org/dendrite/cmd/dendrite-sync-server/main.go

@@ -6,9 +6,9 @@ import (
 	"path/filepath"
 
 	"github.com/matrix-org/dendrite/clientapi/config"
-	"github.com/matrix-org/dendrite/clientapi/consumers"
 	"github.com/matrix-org/dendrite/clientapi/routing"
-	"github.com/matrix-org/dendrite/clientapi/storage/sync"
+	"github.com/matrix-org/dendrite/clientapi/storage"
+	"github.com/matrix-org/dendrite/clientapi/sync"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/matrix-org/dugong"
@@ -42,21 +42,24 @@ func main() {
 	cfg := config.Sync{
 		KafkaConsumerURIs:     []string{"localhost:9092"},
 		RoomserverOutputTopic: "roomserverOutput",
-		DataSource:            "",
+		DataSource:            "postgres://user:pass@localhost/dendrite-sync-server?sslmode=disable",
 	}
 
 	log.Info("Starting sync server")
 
-	db, err := sync.NewDatabase(cfg.DataSource)
+	db, err := storage.NewSyncServerDatabase(cfg.DataSource)
 	if err != nil {
-		log.Panicf("startup: failed to create database with data source %s : %s", cfg.DataSource, err)
+		log.Panicf("startup: failed to create sync server database with data source %s : %s", cfg.DataSource, err)
 	}
 
-	consumer, err := consumers.NewRoomserverConsumer(&cfg, db)
+	server, err := sync.NewServer(&cfg, db)
 	if err != nil {
-		log.Panicf("startup: failed to create roomserver consumer: %s", err)
+		log.Panicf("startup: failed to create sync server: %s", err)
+	}
+	if err = server.Start(); err != nil {
+		log.Panicf("startup: failed to start sync server")
 	}
 
-	routing.SetupSyncServer(http.DefaultServeMux, http.DefaultClient, cfg, consumer)
+	routing.SetupSyncServerListeners(http.DefaultServeMux, http.DefaultClient, cfg)
 	log.Fatal(http.ListenAndServe(bindAddr, nil))
 }