Factor out component setup from commands (#384)
* Add base component * Convert clientapi to using base component * Convert federationapi to using base component * Convert federationsender to using base component * Convert mediaapi to using base component * Convert publicroomsapi to using base component * Convert roomserver to using base component * Convert syncapi to using base component * Convert monolith to using base component * Split out config parsing and roomserver API creationmain
Erik Johnston
GitHub
parent
fa362ecef2
commit
d22fb24a66
|
|
@ -0,0 +1,66 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package clientapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/clientapi/consumers"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/clientapi/routing"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupClientAPIComponent sets up and registers HTTP handlers for the ClientAPI
|
||||
// component.
|
||||
func SetupClientAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
accountsDB *accounts.Database,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
keyRing *gomatrixserverlib.KeyRing,
|
||||
aliasAPI api.RoomserverAliasAPI,
|
||||
inputAPI api.RoomserverInputAPI,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
userUpdateProducer := &producers.UserUpdateProducer{
|
||||
Producer: base.KafkaProducer,
|
||||
Topic: string(base.Cfg.Kafka.Topics.UserUpdates),
|
||||
}
|
||||
|
||||
syncProducer := &producers.SyncAPIProducer{
|
||||
Producer: base.KafkaProducer,
|
||||
Topic: string(base.Cfg.Kafka.Topics.OutputClientData),
|
||||
}
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(
|
||||
base.Cfg, base.KafkaConsumer, accountsDB, queryAPI,
|
||||
)
|
||||
if err := consumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start room server consumer")
|
||||
}
|
||||
|
||||
routing.Setup(
|
||||
base.APIMux, *base.Cfg, roomserverProducer,
|
||||
queryAPI, aliasAPI, accountsDB, deviceDB,
|
||||
federation, *keyRing,
|
||||
userUpdateProducer, syncProducer,
|
||||
)
|
||||
}
|
||||
|
|
@ -15,113 +15,29 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/clientapi/consumers"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/clientapi/routing"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/clientapi"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
)
|
||||
|
||||
var (
|
||||
logDir = os.Getenv("LOG_DIR")
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file, For more information see the config file in this repository")
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(logDir)
|
||||
cfg := basecomponent.ParseFlags()
|
||||
|
||||
flag.Parse()
|
||||
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteClientAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
aliasAPI := api.NewRoomserverAliasAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
inputAPI := api.NewRoomserverInputAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
kafkaProducer, err := sarama.NewSyncProducer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka producers")
|
||||
}
|
||||
|
||||
userUpdateProducer := &producers.UserUpdateProducer{
|
||||
Producer: kafkaProducer,
|
||||
Topic: string(cfg.Kafka.Topics.UserUpdates),
|
||||
}
|
||||
|
||||
syncProducer := &producers.SyncAPIProducer{
|
||||
Producer: kafkaProducer,
|
||||
Topic: string(cfg.Kafka.Topics.OutputClientData),
|
||||
}
|
||||
|
||||
federation := gomatrixserverlib.NewFederationClient(
|
||||
cfg.Matrix.ServerName, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey,
|
||||
)
|
||||
|
||||
accountDB, err := accounts.NewDatabase(string(cfg.Database.Account), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup account database(%q): %s", cfg.Database.Account, err.Error())
|
||||
}
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup device database(%q): %s", cfg.Database.Device, err.Error())
|
||||
}
|
||||
keyDB, err := keydb.NewDatabase(string(cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup key database(%q): %s", cfg.Database.ServerKey, err.Error())
|
||||
}
|
||||
base := basecomponent.NewBaseDendrite(cfg, "ClientAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
accountDB := base.CreateAccountsDB()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
alias, input, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, accountDB, queryAPI)
|
||||
if err = consumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer")
|
||||
}
|
||||
|
||||
log.Info("Starting client API server on ", cfg.Listen.ClientAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(
|
||||
api, *cfg, roomserverProducer,
|
||||
queryAPI, aliasAPI, accountDB, deviceDB, federation, keyRing,
|
||||
userUpdateProducer, syncProducer,
|
||||
clientapi.SetupClientAPIComponent(
|
||||
base, deviceDB, accountDB, federation, &keyRing,
|
||||
alias, input, query,
|
||||
)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.ClientAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.ClientAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,84 +15,27 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/dendrite/federationapi/routing"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var (
|
||||
logDir = os.Getenv("LOG_DIR")
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
"github.com/matrix-org/dendrite/federationapi"
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(logDir)
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "FederationAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
accountDB := base.CreateAccountsDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
alias, input, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteFederationAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
federation := gomatrixserverlib.NewFederationClient(
|
||||
cfg.Matrix.ServerName, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey,
|
||||
federationapi.SetupFederationAPIComponent(
|
||||
base, accountDB, federation, &keyRing,
|
||||
alias, input, query,
|
||||
)
|
||||
|
||||
keyDB, err := keydb.NewDatabase(string(cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup key database(%q): %s", cfg.Database.ServerKey, err.Error())
|
||||
}
|
||||
|
||||
accountDB, err := accounts.NewDatabase(string(cfg.Database.Account), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup account database(%q): %s", cfg.Database.Account, err.Error())
|
||||
}
|
||||
|
||||
keyRing := gomatrixserverlib.KeyRing{
|
||||
KeyFetchers: []gomatrixserverlib.KeyFetcher{
|
||||
// TODO: Use perspective key fetchers for production.
|
||||
&gomatrixserverlib.DirectKeyFetcher{Client: federation.Client},
|
||||
},
|
||||
KeyDatabase: keyDB,
|
||||
}
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
inputAPI := api.NewRoomserverInputAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
aliasAPI := api.NewRoomserverAliasAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup kafka producers(%s): %s", cfg.Kafka.Addresses, err)
|
||||
}
|
||||
|
||||
log.Info("Starting federation API server on ", cfg.Listen.FederationAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, *cfg, queryAPI, aliasAPI, roomserverProducer, keyRing, federation, accountDB)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, api)
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.FederationAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.FederationAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,74 +15,22 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/federationsender/consumers"
|
||||
"github.com/matrix-org/dendrite/federationsender/queue"
|
||||
"github.com/matrix-org/dendrite/federationsender/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/federationsender"
|
||||
)
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(os.Getenv("LOG_DIR"))
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "FederationSender")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
federation := base.CreateFederationClient()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
_, _, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteFederationSender")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
db, err := storage.NewDatabase(string(cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create federation sender database with data source %s : %s", cfg.Database.FederationSender, err)
|
||||
}
|
||||
|
||||
federation := gomatrixserverlib.NewFederationClient(
|
||||
cfg.Matrix.ServerName, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey,
|
||||
federationsender.SetupFederationSenderComponent(
|
||||
base, federation, query,
|
||||
)
|
||||
|
||||
queues := queue.NewOutgoingQueues(cfg.Matrix.ServerName, federation)
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, queues, db, queryAPI)
|
||||
if err = consumer.Start(); err != nil {
|
||||
log.WithError(err).Panicf("startup: failed to start room server consumer")
|
||||
}
|
||||
|
||||
api := mux.NewRouter()
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, api)
|
||||
|
||||
if err := http.ListenAndServe(string(cfg.Listen.FederationSender), nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.FederationSender))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,62 +15,18 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/mediaapi/routing"
|
||||
"github.com/matrix-org/dendrite/mediaapi/storage"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var (
|
||||
logDir = os.Getenv("LOG_DIR")
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/mediaapi"
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(logDir)
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "MediaAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
mediaapi.SetupMediaAPIComponent(base, deviceDB)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteMediaAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
db, err := storage.Open(string(cfg.Database.MediaAPI))
|
||||
if err != nil {
|
||||
log.WithError(err).Panic("Failed to open database")
|
||||
}
|
||||
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.WithError(err).Panicf("Failed to setup device database(%q)", cfg.Database.Device)
|
||||
}
|
||||
|
||||
client := gomatrixserverlib.NewClient()
|
||||
|
||||
log.Info("Starting media API server on ", cfg.Listen.MediaAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, cfg, db, deviceDB, client)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.MediaAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.MediaAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,56 +15,24 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/naffka"
|
||||
|
||||
mediaapi_routing "github.com/matrix-org/dendrite/mediaapi/routing"
|
||||
mediaapi_storage "github.com/matrix-org/dendrite/mediaapi/storage"
|
||||
|
||||
roomserver_alias "github.com/matrix-org/dendrite/roomserver/alias"
|
||||
roomserver_input "github.com/matrix-org/dendrite/roomserver/input"
|
||||
roomserver_query "github.com/matrix-org/dendrite/roomserver/query"
|
||||
roomserver_storage "github.com/matrix-org/dendrite/roomserver/storage"
|
||||
|
||||
clientapi_consumers "github.com/matrix-org/dendrite/clientapi/consumers"
|
||||
clientapi_routing "github.com/matrix-org/dendrite/clientapi/routing"
|
||||
|
||||
syncapi_consumers "github.com/matrix-org/dendrite/syncapi/consumers"
|
||||
syncapi_routing "github.com/matrix-org/dendrite/syncapi/routing"
|
||||
syncapi_storage "github.com/matrix-org/dendrite/syncapi/storage"
|
||||
syncapi_sync "github.com/matrix-org/dendrite/syncapi/sync"
|
||||
syncapi_types "github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
||||
federationapi_routing "github.com/matrix-org/dendrite/federationapi/routing"
|
||||
|
||||
federationsender_consumers "github.com/matrix-org/dendrite/federationsender/consumers"
|
||||
"github.com/matrix-org/dendrite/federationsender/queue"
|
||||
federationsender_storage "github.com/matrix-org/dendrite/federationsender/storage"
|
||||
|
||||
publicroomsapi_consumers "github.com/matrix-org/dendrite/publicroomsapi/consumers"
|
||||
publicroomsapi_routing "github.com/matrix-org/dendrite/publicroomsapi/routing"
|
||||
publicroomsapi_storage "github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/clientapi"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/federationapi"
|
||||
"github.com/matrix-org/dendrite/federationsender"
|
||||
"github.com/matrix-org/dendrite/mediaapi"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/syncapi"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var (
|
||||
logDir = os.Getenv("LOG_DIR")
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
httpBindAddr = flag.String("http-bind-address", ":8008", "The HTTP listening port for the server")
|
||||
httpsBindAddr = flag.String("https-bind-address", ":8448", "The HTTPS listening port for the server")
|
||||
certFile = flag.String("tls-cert", "", "The PEM formatted X509 certificate to use for TLS")
|
||||
|
|
@ -72,285 +40,40 @@ var (
|
|||
)
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(logDir)
|
||||
cfg := basecomponent.ParseMonolithFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "Monolith")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
accountDB := base.CreateAccountsDB()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.LoadMonolithic(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
alias, input, query := roomserver.SetupRoomServerComponent(base)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteMonolith")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
clientapi.SetupClientAPIComponent(base, deviceDB, accountDB, federation, &keyRing, alias, input, query)
|
||||
federationapi.SetupFederationAPIComponent(base, accountDB, federation, &keyRing, alias, input, query)
|
||||
federationsender.SetupFederationSenderComponent(base, federation, query)
|
||||
mediaapi.SetupMediaAPIComponent(base, deviceDB)
|
||||
publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB)
|
||||
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query)
|
||||
|
||||
m := newMonolith(cfg)
|
||||
m.setupDatabases()
|
||||
m.setupFederation()
|
||||
m.setupKafka()
|
||||
m.setupRoomServer()
|
||||
m.setupProducers()
|
||||
m.setupNotifiers()
|
||||
m.setupConsumers()
|
||||
m.setupAPIs()
|
||||
httpHandler := common.WrapHandlerInCORS(base.APIMux)
|
||||
|
||||
// Expose the matrix APIs directly rather than putting them under a /api path.
|
||||
go func() {
|
||||
log.Info("Listening on ", *httpBindAddr)
|
||||
log.Fatal(http.ListenAndServe(*httpBindAddr, common.WrapHandlerInCORS(m.api)))
|
||||
logrus.Info("Listening on ", *httpBindAddr)
|
||||
logrus.Fatal(http.ListenAndServe(*httpBindAddr, httpHandler))
|
||||
}()
|
||||
// Handle HTTPS if certificate and key are provided
|
||||
go func() {
|
||||
if *certFile != "" && *keyFile != "" {
|
||||
log.Info("Listening on ", *httpsBindAddr)
|
||||
log.Fatal(http.ListenAndServeTLS(*httpsBindAddr, *certFile, *keyFile, m.api))
|
||||
logrus.Info("Listening on ", *httpsBindAddr)
|
||||
logrus.Fatal(http.ListenAndServeTLS(*httpsBindAddr, *certFile, *keyFile, httpHandler))
|
||||
}
|
||||
}()
|
||||
|
||||
// We want to block forever to let the HTTP and HTTPS handler serve the APIs
|
||||
select {}
|
||||
}
|
||||
|
||||
// A monolith contains all the dendrite components.
|
||||
// Some of the setup functions depend on previous setup functions, so they must
|
||||
// be called in the same order as they are defined in the file.
|
||||
type monolith struct {
|
||||
cfg *config.Dendrite
|
||||
api *mux.Router
|
||||
|
||||
roomServerDB *roomserver_storage.Database
|
||||
accountDB *accounts.Database
|
||||
deviceDB *devices.Database
|
||||
keyDB *keydb.Database
|
||||
mediaAPIDB *mediaapi_storage.Database
|
||||
syncAPIDB *syncapi_storage.SyncServerDatabase
|
||||
federationSenderDB *federationsender_storage.Database
|
||||
publicRoomsAPIDB *publicroomsapi_storage.PublicRoomsServerDatabase
|
||||
|
||||
federation *gomatrixserverlib.FederationClient
|
||||
keyRing gomatrixserverlib.KeyRing
|
||||
|
||||
inputAPI *roomserver_input.RoomserverInputAPI
|
||||
queryAPI *roomserver_query.RoomserverQueryAPI
|
||||
aliasAPI *roomserver_alias.RoomserverAliasAPI
|
||||
|
||||
naffka *naffka.Naffka
|
||||
kafkaProducer sarama.SyncProducer
|
||||
|
||||
roomServerProducer *producers.RoomserverProducer
|
||||
userUpdateProducer *producers.UserUpdateProducer
|
||||
syncProducer *producers.SyncAPIProducer
|
||||
|
||||
syncAPINotifier *syncapi_sync.Notifier
|
||||
}
|
||||
|
||||
func newMonolith(cfg *config.Dendrite) *monolith {
|
||||
return &monolith{cfg: cfg, api: mux.NewRouter()}
|
||||
}
|
||||
|
||||
func (m *monolith) setupDatabases() {
|
||||
var err error
|
||||
m.roomServerDB, err = roomserver_storage.Open(string(m.cfg.Database.RoomServer))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
m.accountDB, err = accounts.NewDatabase(string(m.cfg.Database.Account), m.cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup account database(%q): %s", m.cfg.Database.Account, err.Error())
|
||||
}
|
||||
m.deviceDB, err = devices.NewDatabase(string(m.cfg.Database.Device), m.cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup device database(%q): %s", m.cfg.Database.Device, err.Error())
|
||||
}
|
||||
m.keyDB, err = keydb.NewDatabase(string(m.cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup key database(%q): %s", m.cfg.Database.ServerKey, err.Error())
|
||||
}
|
||||
m.mediaAPIDB, err = mediaapi_storage.Open(string(m.cfg.Database.MediaAPI))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup sync api database(%q): %s", m.cfg.Database.MediaAPI, err.Error())
|
||||
}
|
||||
m.syncAPIDB, err = syncapi_storage.NewSyncServerDatabase(string(m.cfg.Database.SyncAPI))
|
||||
if err != nil {
|
||||
log.Panicf("Failed to setup sync api database(%q): %s", m.cfg.Database.SyncAPI, err.Error())
|
||||
}
|
||||
m.federationSenderDB, err = federationsender_storage.NewDatabase(string(m.cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create federation sender database with data source %s : %s", m.cfg.Database.FederationSender, err)
|
||||
}
|
||||
m.publicRoomsAPIDB, err = publicroomsapi_storage.NewPublicRoomsServerDatabase(string(m.cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to setup public rooms api database with data source %s : %s", m.cfg.Database.PublicRoomsAPI, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupFederation() {
|
||||
m.federation = gomatrixserverlib.NewFederationClient(
|
||||
m.cfg.Matrix.ServerName, m.cfg.Matrix.KeyID, m.cfg.Matrix.PrivateKey,
|
||||
)
|
||||
|
||||
m.keyRing = keydb.CreateKeyRing(m.federation.Client, m.keyDB)
|
||||
}
|
||||
|
||||
func (m *monolith) setupKafka() {
|
||||
if m.cfg.Kafka.UseNaffka {
|
||||
db, err := sql.Open("postgres", string(m.cfg.Database.Naffka))
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
}).Panic("Failed to open naffka database")
|
||||
}
|
||||
|
||||
naffkaDB, err := naffka.NewPostgresqlDatabase(db)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
}).Panic("Failed to setup naffka database")
|
||||
}
|
||||
|
||||
naff, err := naffka.New(naffkaDB)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
}).Panic("Failed to setup naffka")
|
||||
}
|
||||
m.naffka = naff
|
||||
m.kafkaProducer = naff
|
||||
} else {
|
||||
var err error
|
||||
m.kafkaProducer, err = sarama.NewSyncProducer(m.cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": m.cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka producers")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) kafkaConsumer() sarama.Consumer {
|
||||
if m.cfg.Kafka.UseNaffka {
|
||||
return m.naffka
|
||||
}
|
||||
consumer, err := sarama.NewConsumer(m.cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": m.cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
return consumer
|
||||
}
|
||||
|
||||
func (m *monolith) setupRoomServer() {
|
||||
m.inputAPI = &roomserver_input.RoomserverInputAPI{
|
||||
DB: m.roomServerDB,
|
||||
Producer: m.kafkaProducer,
|
||||
OutputRoomEventTopic: string(m.cfg.Kafka.Topics.OutputRoomEvent),
|
||||
}
|
||||
|
||||
m.queryAPI = &roomserver_query.RoomserverQueryAPI{
|
||||
DB: m.roomServerDB,
|
||||
}
|
||||
|
||||
m.aliasAPI = &roomserver_alias.RoomserverAliasAPI{
|
||||
DB: m.roomServerDB,
|
||||
Cfg: m.cfg,
|
||||
InputAPI: m.inputAPI,
|
||||
QueryAPI: m.queryAPI,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupProducers() {
|
||||
m.roomServerProducer = producers.NewRoomserverProducer(m.inputAPI)
|
||||
m.userUpdateProducer = &producers.UserUpdateProducer{
|
||||
Producer: m.kafkaProducer,
|
||||
Topic: string(m.cfg.Kafka.Topics.UserUpdates),
|
||||
}
|
||||
m.syncProducer = &producers.SyncAPIProducer{
|
||||
Producer: m.kafkaProducer,
|
||||
Topic: string(m.cfg.Kafka.Topics.OutputClientData),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupNotifiers() {
|
||||
pos, err := m.syncAPIDB.SyncStreamPosition(context.Background())
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to get latest sync stream position : %s", err)
|
||||
}
|
||||
|
||||
m.syncAPINotifier = syncapi_sync.NewNotifier(syncapi_types.StreamPosition(pos))
|
||||
if err = m.syncAPINotifier.Load(context.Background(), m.syncAPIDB); err != nil {
|
||||
log.Panicf("startup: failed to set up notifier: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupConsumers() {
|
||||
var err error
|
||||
|
||||
clientAPIConsumer := clientapi_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.accountDB, m.queryAPI,
|
||||
)
|
||||
if err = clientAPIConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
syncAPIRoomConsumer := syncapi_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.syncAPINotifier, m.syncAPIDB, m.queryAPI,
|
||||
)
|
||||
if err = syncAPIRoomConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
syncAPIClientConsumer := syncapi_consumers.NewOutputClientDataConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.syncAPINotifier, m.syncAPIDB,
|
||||
)
|
||||
if err = syncAPIClientConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start client API server consumer: %s", err)
|
||||
}
|
||||
|
||||
publicRoomsAPIConsumer := publicroomsapi_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), m.publicRoomsAPIDB, m.queryAPI,
|
||||
)
|
||||
if err = publicRoomsAPIConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
federationSenderQueues := queue.NewOutgoingQueues(m.cfg.Matrix.ServerName, m.federation)
|
||||
|
||||
federationSenderRoomConsumer := federationsender_consumers.NewOutputRoomEventConsumer(
|
||||
m.cfg, m.kafkaConsumer(), federationSenderQueues, m.federationSenderDB, m.queryAPI,
|
||||
)
|
||||
if err = federationSenderRoomConsumer.Start(); err != nil {
|
||||
log.WithError(err).Panicf("startup: failed to start room server consumer")
|
||||
}
|
||||
}
|
||||
|
||||
func (m *monolith) setupAPIs() {
|
||||
clientapi_routing.Setup(
|
||||
m.api, *m.cfg, m.roomServerProducer,
|
||||
m.queryAPI, m.aliasAPI, m.accountDB, m.deviceDB, m.federation, m.keyRing,
|
||||
m.userUpdateProducer, m.syncProducer,
|
||||
)
|
||||
|
||||
mediaapi_routing.Setup(
|
||||
m.api, m.cfg, m.mediaAPIDB, m.deviceDB, &m.federation.Client,
|
||||
)
|
||||
|
||||
syncapi_routing.Setup(m.api, syncapi_sync.NewRequestPool(
|
||||
m.syncAPIDB, m.syncAPINotifier, m.accountDB,
|
||||
), m.syncAPIDB, m.deviceDB)
|
||||
|
||||
federationapi_routing.Setup(
|
||||
m.api, *m.cfg, m.queryAPI, m.aliasAPI, m.roomServerProducer, m.keyRing, m.federation,
|
||||
m.accountDB,
|
||||
)
|
||||
|
||||
publicroomsapi_routing.Setup(m.api, m.deviceDB, m.publicRoomsAPIDB)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,77 +15,18 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/consumers"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/routing"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi"
|
||||
)
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(os.Getenv("LOG_DIR"))
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "PublicRoomsAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB)
|
||||
|
||||
closer, err := cfg.SetupTracing("DendritePublicRoomsAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
db, err := storage.NewPublicRoomsServerDatabase(string(cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create public rooms server database with data source %s : %s", cfg.Database.PublicRoomsAPI, err)
|
||||
}
|
||||
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create device database with data source %s : %s", cfg.Database.Device, err)
|
||||
}
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, db, queryAPI)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create room server consumer: %s", err)
|
||||
}
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
|
||||
log.Info("Starting public rooms server on ", cfg.Listen.PublicRoomsAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, deviceDB, db)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.PublicRoomsAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.PublicRoomsAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,85 +15,18 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"net/http"
|
||||
_ "net/http/pprof"
|
||||
"os"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/alias"
|
||||
"github.com/matrix-org/dendrite/roomserver/input"
|
||||
"github.com/matrix-org/dendrite/roomserver/query"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
)
|
||||
|
||||
var (
|
||||
logDir = os.Getenv("LOG_DIR")
|
||||
configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
)
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(logDir)
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "RoomServerAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
roomserver.SetupRoomServerComponent(base)
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteRoomServer")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
|
||||
db, err := storage.Open(string(cfg.Database.RoomServer))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
kafkaProducer, err := sarama.NewSyncProducer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
inputAPI := input.RoomserverInputAPI{
|
||||
DB: db,
|
||||
Producer: kafkaProducer,
|
||||
OutputRoomEventTopic: string(cfg.Kafka.Topics.OutputRoomEvent),
|
||||
}
|
||||
|
||||
inputAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
queryAPI := query.RoomserverQueryAPI{DB: db}
|
||||
|
||||
queryAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
aliasAPI := alias.RoomserverAliasAPI{
|
||||
DB: db,
|
||||
Cfg: cfg,
|
||||
InputAPI: &inputAPI,
|
||||
QueryAPI: &queryAPI,
|
||||
}
|
||||
|
||||
aliasAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
// This is deprecated, but prometheus are still arguing on what to replace
|
||||
// it with. Alternatively we could set it up manually.
|
||||
http.DefaultServeMux.Handle("/metrics", prometheus.Handler()) // nolint: staticcheck, megacheck
|
||||
|
||||
log.Info("Started room server on ", cfg.Listen.RoomServer)
|
||||
|
||||
// TODO: Implement clean shutdown.
|
||||
if err := http.ListenAndServe(string(cfg.Listen.RoomServer), nil); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.RoomServer))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,97 +15,21 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi/consumers"
|
||||
"github.com/matrix-org/dendrite/syncapi/routing"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage"
|
||||
"github.com/matrix-org/dendrite/syncapi/sync"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/syncapi"
|
||||
)
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
func main() {
|
||||
common.SetupLogging(os.Getenv("LOG_DIR"))
|
||||
cfg := basecomponent.ParseFlags()
|
||||
base := basecomponent.NewBaseDendrite(cfg, "SyncAPI")
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
flag.Parse()
|
||||
deviceDB := base.CreateDeviceDB()
|
||||
accountDB := base.CreateAccountsDB()
|
||||
|
||||
if *configPath == "" {
|
||||
log.Fatal("--config must be supplied")
|
||||
}
|
||||
cfg, err := config.Load(*configPath)
|
||||
if err != nil {
|
||||
log.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
_, _, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
||||
closer, err := cfg.SetupTracing("DendriteSyncAPI")
|
||||
if err != nil {
|
||||
log.WithError(err).Fatalf("Failed to start tracer")
|
||||
}
|
||||
defer closer.Close() // nolint: errcheck
|
||||
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query)
|
||||
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP(cfg.RoomServerURL(), nil)
|
||||
|
||||
db, err := storage.NewSyncServerDatabase(string(cfg.Database.SyncAPI))
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create sync server database with data source %s : %s", cfg.Database.SyncAPI, err)
|
||||
}
|
||||
|
||||
deviceDB, err := devices.NewDatabase(string(cfg.Database.Device), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create device database with data source %s : %s", cfg.Database.Device, err)
|
||||
}
|
||||
|
||||
adb, err := accounts.NewDatabase(string(cfg.Database.Account), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to create account database with data source %s : %s", cfg.Database.Account, err)
|
||||
}
|
||||
|
||||
pos, err := db.SyncStreamPosition(context.Background())
|
||||
if err != nil {
|
||||
log.Panicf("startup: failed to get latest sync stream position : %s", err)
|
||||
}
|
||||
|
||||
n := sync.NewNotifier(types.StreamPosition(pos))
|
||||
if err = n.Load(context.Background(), db); err != nil {
|
||||
log.Panicf("startup: failed to set up notifier: %s", err)
|
||||
}
|
||||
|
||||
kafkaConsumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
log.ErrorKey: err,
|
||||
"addresses": cfg.Kafka.Addresses,
|
||||
}).Panic("Failed to setup kafka consumers")
|
||||
}
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(cfg, kafkaConsumer, n, db, queryAPI)
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start room server consumer: %s", err)
|
||||
}
|
||||
clientConsumer := consumers.NewOutputClientDataConsumer(cfg, kafkaConsumer, n, db)
|
||||
if err = clientConsumer.Start(); err != nil {
|
||||
log.Panicf("startup: failed to start client API server consumer: %s", err)
|
||||
}
|
||||
|
||||
log.Info("Starting sync server on ", cfg.Listen.SyncAPI)
|
||||
|
||||
api := mux.NewRouter()
|
||||
routing.Setup(api, sync.NewRequestPool(db, n, adb), db, deviceDB)
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(api))
|
||||
|
||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.SyncAPI), nil))
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Listen.SyncAPI))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,182 @@
|
|||
// Copyright 2017 New Vector Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package basecomponent
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/keydb"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/naffka"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
sarama "gopkg.in/Shopify/sarama.v1"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// BaseDendrite is a base for creating new instances of dendrite. It parses
|
||||
// command line flags and config, and exposes methods for creating various
|
||||
// resources. All errors are handled by logging then exiting, so all methods
|
||||
// should only be used during start up.
|
||||
// Must be closed when shutting down.
|
||||
type BaseDendrite struct {
|
||||
componentName string
|
||||
tracerCloser io.Closer
|
||||
|
||||
// APIMux should be used to register new public matrix api endpoints
|
||||
APIMux *mux.Router
|
||||
Cfg *config.Dendrite
|
||||
KafkaConsumer sarama.Consumer
|
||||
KafkaProducer sarama.SyncProducer
|
||||
}
|
||||
|
||||
// NewBaseDendrite creates a new instance to be used by a component.
|
||||
// The componentName is used for logging purposes, and should be a friendly name
|
||||
// of the compontent running, e.g. "SyncAPI"
|
||||
func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite {
|
||||
common.SetupLogging(os.Getenv("LOG_DIR"))
|
||||
|
||||
closer, err := cfg.SetupTracing("Dendrite" + componentName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start opentracing")
|
||||
}
|
||||
|
||||
kafkaConsumer, kafkaProducer := setupKafka(cfg)
|
||||
|
||||
return &BaseDendrite{
|
||||
componentName: componentName,
|
||||
tracerCloser: closer,
|
||||
Cfg: cfg,
|
||||
APIMux: mux.NewRouter(),
|
||||
KafkaConsumer: kafkaConsumer,
|
||||
KafkaProducer: kafkaProducer,
|
||||
}
|
||||
}
|
||||
|
||||
// Close implements io.Closer
|
||||
func (b *BaseDendrite) Close() error {
|
||||
return b.tracerCloser.Close()
|
||||
}
|
||||
|
||||
// CreateHTTPRoomserverAPIs returns the AliasAPI, InputAPI and QueryAPI to hit
|
||||
// the roomserver over HTTP.
|
||||
func (b *BaseDendrite) CreateHTTPRoomserverAPIs() (api.RoomserverAliasAPI, api.RoomserverInputAPI, api.RoomserverQueryAPI) {
|
||||
alias := api.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
input := api.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
query := api.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
return alias, input, query
|
||||
}
|
||||
|
||||
// CreateDeviceDB creates a new instance of the device database. Should only be
|
||||
// called once per component.
|
||||
func (b *BaseDendrite) CreateDeviceDB() *devices.Database {
|
||||
db, err := devices.NewDatabase(string(b.Cfg.Database.Device), b.Cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to devices db")
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
// CreateAccountsDB creates a new instance of the accounts database. Should only
|
||||
// be called once per component.
|
||||
func (b *BaseDendrite) CreateAccountsDB() *accounts.Database {
|
||||
db, err := accounts.NewDatabase(string(b.Cfg.Database.Account), b.Cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to accounts db")
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
// CreateKeyDB creates a new instance of the key database. Should only be called
|
||||
// once per component.
|
||||
func (b *BaseDendrite) CreateKeyDB() *keydb.Database {
|
||||
db, err := keydb.NewDatabase(string(b.Cfg.Database.ServerKey))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to keys db")
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
// CreateFederationClient creates a new federation client. Should only be called
|
||||
// once per component.
|
||||
func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
|
||||
return gomatrixserverlib.NewFederationClient(
|
||||
b.Cfg.Matrix.ServerName, b.Cfg.Matrix.KeyID, b.Cfg.Matrix.PrivateKey,
|
||||
)
|
||||
}
|
||||
|
||||
// SetupAndServeHTTP sets up the HTTP server to serve endpoints registered on
|
||||
// ApiMux under /api/ and adds a prometheus handler under /metrics.
|
||||
func (b *BaseDendrite) SetupAndServeHTTP(addr string) {
|
||||
common.SetupHTTPAPI(http.DefaultServeMux, common.WrapHandlerInCORS(b.APIMux))
|
||||
|
||||
logrus.Infof("Starting %s server on %s", b.componentName, addr)
|
||||
|
||||
err := http.ListenAndServe(addr, nil)
|
||||
|
||||
if err != nil {
|
||||
logrus.WithError(err).Fatal("failed to serve http")
|
||||
}
|
||||
|
||||
logrus.Infof("Stopped %s server on %s", b.componentName, addr)
|
||||
}
|
||||
|
||||
// setupKafka creates kafka consumer/producer pair from the config. Checks if
|
||||
// should use naffka.
|
||||
func setupKafka(cfg *config.Dendrite) (sarama.Consumer, sarama.SyncProducer) {
|
||||
if cfg.Kafka.UseNaffka {
|
||||
db, err := sql.Open("postgres", string(cfg.Database.Naffka))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to open naffka database")
|
||||
}
|
||||
|
||||
naffkaDB, err := naffka.NewPostgresqlDatabase(db)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to setup naffka database")
|
||||
}
|
||||
|
||||
naff, err := naffka.New(naffkaDB)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to setup naffka")
|
||||
}
|
||||
|
||||
return naff, naff
|
||||
}
|
||||
|
||||
consumer, err := sarama.NewConsumer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to start kafka consumer")
|
||||
}
|
||||
|
||||
producer, err := sarama.NewSyncProducer(cfg.Kafka.Addresses, nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to setup kafka producers")
|
||||
}
|
||||
|
||||
return consumer, producer
|
||||
}
|
||||
|
|
@ -0,0 +1,61 @@
|
|||
// Copyright 2017 New Vector Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package basecomponent
|
||||
|
||||
import (
|
||||
"flag"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var configPath = flag.String("config", "dendrite.yaml", "The path to the config file. For more information, see the config file in this repository.")
|
||||
|
||||
// ParseFlags parses the commandline flags and uses them to create a config.
|
||||
// If running as a monolith use `ParseMonolithFlags` instead.
|
||||
func ParseFlags() *config.Dendrite {
|
||||
flag.Parse()
|
||||
|
||||
if *configPath == "" {
|
||||
logrus.Fatal("--config must be supplied")
|
||||
}
|
||||
|
||||
cfg, err := config.Load(*configPath)
|
||||
|
||||
if err != nil {
|
||||
logrus.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
||||
// ParseMonolithFlags parses the commandline flags and uses them to create a
|
||||
// config. Should only be used if running a monolith. See `ParseFlags`.
|
||||
func ParseMonolithFlags() *config.Dendrite {
|
||||
flag.Parse()
|
||||
|
||||
if *configPath == "" {
|
||||
logrus.Fatal("--config must be supplied")
|
||||
}
|
||||
|
||||
cfg, err := config.LoadMonolithic(*configPath)
|
||||
|
||||
if err != nil {
|
||||
logrus.Fatalf("Invalid config file: %s", err)
|
||||
}
|
||||
|
||||
return cfg
|
||||
}
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package federationapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
// TODO: Are we really wanting to pull in the producer from clientapi
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/federationapi/routing"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
// SetupFederationAPIComponent sets up and registers HTTP handlers for the
|
||||
// FederationAPI component.
|
||||
func SetupFederationAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
accountsDB *accounts.Database,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
keyRing *gomatrixserverlib.KeyRing,
|
||||
aliasAPI api.RoomserverAliasAPI,
|
||||
inputAPI api.RoomserverInputAPI,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
|
||||
|
||||
routing.Setup(
|
||||
base.APIMux, *base.Cfg, queryAPI, aliasAPI,
|
||||
roomserverProducer, *keyRing, federation, accountsDB,
|
||||
)
|
||||
}
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package federationsender
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/federationsender/consumers"
|
||||
"github.com/matrix-org/dendrite/federationsender/queue"
|
||||
"github.com/matrix-org/dendrite/federationsender/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupFederationSenderComponent sets up and registers HTTP handlers for the
|
||||
// FederationSender component.
|
||||
func SetupFederationSenderComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
federationSenderDB, err := storage.NewDatabase(string(base.Cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to connect to federation sender db")
|
||||
}
|
||||
|
||||
queues := queue.NewOutgoingQueues(base.Cfg.Matrix.ServerName, federation)
|
||||
|
||||
consumer := consumers.NewOutputRoomEventConsumer(
|
||||
base.Cfg, base.KafkaConsumer, queues,
|
||||
federationSenderDB, queryAPI,
|
||||
)
|
||||
if err = consumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panic("failed to start room server consumer")
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package mediaapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/mediaapi/routing"
|
||||
"github.com/matrix-org/dendrite/mediaapi/storage"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupMediaAPIComponent sets up and registers HTTP handlers for the MediaAPI
|
||||
// component.
|
||||
func SetupMediaAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
) {
|
||||
mediaDB, err := storage.Open(string(base.Cfg.Database.MediaAPI))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to media db")
|
||||
}
|
||||
|
||||
routing.Setup(
|
||||
base.APIMux, base.Cfg, mediaDB, deviceDB, gomatrixserverlib.NewClient(),
|
||||
)
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package publicroomsapi
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/routing"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupPublicRoomsAPIComponent sets up and registers HTTP handlers for the PublicRoomsAPI
|
||||
// component.
|
||||
func SetupPublicRoomsAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
) {
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
|
||||
routing.Setup(base.APIMux, deviceDB, publicRoomsDB)
|
||||
}
|
||||
|
|
@ -0,0 +1,64 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package roomserver
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/alias"
|
||||
"github.com/matrix-org/dendrite/roomserver/input"
|
||||
"github.com/matrix-org/dendrite/roomserver/query"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// SetupRoomServerComponent sets up and registers HTTP handlers for the
|
||||
// RoomServer component. Returns instances of the various roomserver APIs,
|
||||
// allowing other components running in the same process to hit the query the
|
||||
// APIs directly instead of having to use HTTP.
|
||||
func SetupRoomServerComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
) (api.RoomserverAliasAPI, api.RoomserverInputAPI, api.RoomserverQueryAPI) {
|
||||
roomserverDB, err := storage.Open(string(base.Cfg.Database.RoomServer))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to room server db")
|
||||
}
|
||||
|
||||
inputAPI := input.RoomserverInputAPI{
|
||||
DB: roomserverDB,
|
||||
Producer: base.KafkaProducer,
|
||||
OutputRoomEventTopic: string(base.Cfg.Kafka.Topics.OutputRoomEvent),
|
||||
}
|
||||
|
||||
inputAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
queryAPI := query.RoomserverQueryAPI{DB: roomserverDB}
|
||||
|
||||
queryAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
aliasAPI := alias.RoomserverAliasAPI{
|
||||
DB: roomserverDB,
|
||||
Cfg: base.Cfg,
|
||||
InputAPI: &inputAPI,
|
||||
QueryAPI: &queryAPI,
|
||||
}
|
||||
|
||||
aliasAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
||||
return &aliasAPI, &inputAPI, &queryAPI
|
||||
}
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
// Copyright 2017 Vector Creations Ltd
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package syncapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
|
||||
"github.com/matrix-org/dendrite/syncapi/consumers"
|
||||
"github.com/matrix-org/dendrite/syncapi/routing"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage"
|
||||
"github.com/matrix-org/dendrite/syncapi/sync"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
)
|
||||
|
||||
// SetupSyncAPIComponent sets up and registers HTTP handlers for the SyncAPI
|
||||
// component.
|
||||
func SetupSyncAPIComponent(
|
||||
base *basecomponent.BaseDendrite,
|
||||
deviceDB *devices.Database,
|
||||
accountsDB *accounts.Database,
|
||||
queryAPI api.RoomserverQueryAPI,
|
||||
) {
|
||||
syncDB, err := storage.NewSyncServerDatabase(string(base.Cfg.Database.SyncAPI))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to sync db")
|
||||
}
|
||||
|
||||
pos, err := syncDB.SyncStreamPosition(context.Background())
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to get stream position")
|
||||
}
|
||||
|
||||
notifier := sync.NewNotifier(types.StreamPosition(pos))
|
||||
err = notifier.Load(context.Background(), syncDB)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start notifier")
|
||||
}
|
||||
|
||||
requestPool := sync.NewRequestPool(syncDB, notifier, accountsDB)
|
||||
|
||||
roomConsumer := consumers.NewOutputRoomEventConsumer(
|
||||
base.Cfg, base.KafkaConsumer, notifier, syncDB, queryAPI,
|
||||
)
|
||||
if err = roomConsumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start room server consumer")
|
||||
}
|
||||
|
||||
clientConsumer := consumers.NewOutputClientDataConsumer(
|
||||
base.Cfg, base.KafkaConsumer, notifier, syncDB,
|
||||
)
|
||||
if err = clientConsumer.Start(); err != nil {
|
||||
logrus.WithError(err).Panicf("failed to start client data consumer")
|
||||
}
|
||||
|
||||
routing.Setup(base.APIMux, requestPool, syncDB, deviceDB)
|
||||
}
|
||||
Loading…
Reference in New Issue