mirror of
https://github.com/hoernschen/dendrite.git
synced 2024-12-27 07:28:27 +00:00
Read roomserver output log and remember position across restarts (#52)
This commit is contained in:
parent
7ab0436689
commit
a423008987
6 changed files with 104 additions and 3 deletions
|
@ -25,4 +25,6 @@ type Sync struct {
|
||||||
RoomserverOutputTopic string
|
RoomserverOutputTopic string
|
||||||
// A list of URIs to consume events from. These kafka logs should be produced by a Room Server.
|
// A list of URIs to consume events from. These kafka logs should be produced by a Room Server.
|
||||||
KafkaConsumerURIs []string
|
KafkaConsumerURIs []string
|
||||||
|
// The postgres connection config for connecting to the database e.g a postgres:// URI
|
||||||
|
DataSource string
|
||||||
}
|
}
|
||||||
|
|
|
@ -48,8 +48,8 @@ func Setup(servMux *http.ServeMux, httpClient *http.Client, cfg config.ClientAPI
|
||||||
servMux.Handle("/api/", http.StripPrefix("/api", apiMux))
|
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) {
|
func SetupSyncServerListeners(servMux *http.ServeMux, httpClient *http.Client, cfg config.Sync) {
|
||||||
apiMux := mux.NewRouter()
|
apiMux := mux.NewRouter()
|
||||||
r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
|
r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
|
||||||
r0mux.Handle("/sync", make("sync", util.NewJSONRequestHandler(func(req *http.Request) util.JSONResponse {
|
r0mux.Handle("/sync", make("sync", util.NewJSONRequestHandler(func(req *http.Request) util.JSONResponse {
|
||||||
|
|
|
@ -0,0 +1,39 @@
|
||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
|
||||||
|
"github.com/matrix-org/dendrite/common"
|
||||||
|
// Import the postgres database driver.
|
||||||
|
_ "github.com/lib/pq"
|
||||||
|
)
|
||||||
|
|
||||||
|
// SyncServerDatabase represents a sync server database
|
||||||
|
type SyncServerDatabase struct {
|
||||||
|
db *sql.DB
|
||||||
|
partitions common.PartitionOffsetStatements
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewSyncServerDatabase creates a new sync server database
|
||||||
|
func NewSyncServerDatabase(dataSourceName string) (*SyncServerDatabase, error) {
|
||||||
|
var db *sql.DB
|
||||||
|
var err error
|
||||||
|
if db, err = sql.Open("postgres", dataSourceName); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
partitions := common.PartitionOffsetStatements{}
|
||||||
|
if err = partitions.Prepare(db); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return &SyncServerDatabase{db, partitions}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PartitionOffsets implements common.PartitionStorer
|
||||||
|
func (d *SyncServerDatabase) PartitionOffsets(topic string) ([]common.PartitionOffset, error) {
|
||||||
|
return d.partitions.SelectPartitionOffsets(topic)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetPartitionOffset implements common.PartitionStorer
|
||||||
|
func (d *SyncServerDatabase) SetPartitionOffset(topic string, partition int32, offset int64) error {
|
||||||
|
return d.partitions.UpsertPartitionOffset(topic, partition, offset)
|
||||||
|
}
|
|
@ -0,0 +1,43 @@
|
||||||
|
package sync
|
||||||
|
|
||||||
|
import (
|
||||||
|
log "github.com/Sirupsen/logrus"
|
||||||
|
"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
|
||||||
|
}
|
||||||
|
|
||||||
|
consumer := common.ContinualConsumer{
|
||||||
|
Topic: cfg.RoomserverOutputTopic,
|
||||||
|
Consumer: kafkaConsumer,
|
||||||
|
PartitionStore: store,
|
||||||
|
}
|
||||||
|
s := &Server{
|
||||||
|
roomServerConsumer: &consumer,
|
||||||
|
}
|
||||||
|
consumer.ProcessMessage = s.onMessage
|
||||||
|
|
||||||
|
return s, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Start consuming from room servers
|
||||||
|
func (s *Server) Start() error {
|
||||||
|
return s.roomServerConsumer.Start()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) onMessage(msg *sarama.ConsumerMessage) error {
|
||||||
|
log.WithField("key", string(msg.Key)).WithField("val", string(msg.Value)).Info("Recv")
|
||||||
|
return nil
|
||||||
|
}
|
|
@ -7,6 +7,8 @@ import (
|
||||||
|
|
||||||
"github.com/matrix-org/dendrite/clientapi/config"
|
"github.com/matrix-org/dendrite/clientapi/config"
|
||||||
"github.com/matrix-org/dendrite/clientapi/routing"
|
"github.com/matrix-org/dendrite/clientapi/routing"
|
||||||
|
"github.com/matrix-org/dendrite/clientapi/storage"
|
||||||
|
"github.com/matrix-org/dendrite/clientapi/sync"
|
||||||
|
|
||||||
log "github.com/Sirupsen/logrus"
|
log "github.com/Sirupsen/logrus"
|
||||||
"github.com/matrix-org/dugong"
|
"github.com/matrix-org/dugong"
|
||||||
|
@ -40,10 +42,24 @@ func main() {
|
||||||
cfg := config.Sync{
|
cfg := config.Sync{
|
||||||
KafkaConsumerURIs: []string{"localhost:9092"},
|
KafkaConsumerURIs: []string{"localhost:9092"},
|
||||||
RoomserverOutputTopic: "roomserverOutput",
|
RoomserverOutputTopic: "roomserverOutput",
|
||||||
|
DataSource: "postgres://dendrite:itsasecret@localhost/syncserver?sslmode=disable",
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Info("Starting sync server")
|
log.Info("Starting sync server")
|
||||||
|
|
||||||
routing.SetupSyncServer(http.DefaultServeMux, http.DefaultClient, cfg)
|
db, err := storage.NewSyncServerDatabase(cfg.DataSource)
|
||||||
|
if err != nil {
|
||||||
|
log.Panicf("startup: failed to create sync server database with data source %s : %s", cfg.DataSource, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
server, err := sync.NewServer(&cfg, db)
|
||||||
|
if err != nil {
|
||||||
|
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.SetupSyncServerListeners(http.DefaultServeMux, http.DefaultClient, cfg)
|
||||||
log.Fatal(http.ListenAndServe(bindAddr, nil))
|
log.Fatal(http.ListenAndServe(bindAddr, nil))
|
||||||
}
|
}
|
||||||
|
|
|
@ -57,6 +57,7 @@ func (s *PartitionOffsetStatements) SelectPartitionOffsets(topic string) ([]Part
|
||||||
if err := rows.Scan(&offset.Partition, &offset.Offset); err != nil {
|
if err := rows.Scan(&offset.Partition, &offset.Offset); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
results = append(results, offset)
|
||||||
}
|
}
|
||||||
return results, nil
|
return results, nil
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue