hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2024-12-27 07:28:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Split out config parsing and roomserver API creation

Browse source
This commit is contained in:
Erik Johnston 2017-12-20 14:22:57 +00:00
parent a80615dd54
commit 51bdba82d1
15 changed files with 134 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
src/github.com/matrix-org/dendrite/clientapi/clientapi.go
View file

@ -21,6 +21,7 @@ import (
"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"
)
@ -33,8 +34,11 @@ func SetupClientAPIComponent(
accountsDB *accounts.Database,
federation *gomatrixserverlib.FederationClient,
keyRing *gomatrixserverlib.KeyRing,
aliasAPI api.RoomserverAliasAPI,
inputAPI api.RoomserverInputAPI,
queryAPI api.RoomserverQueryAPI,
) {
roomserverProducer := producers.NewRoomserverProducer(base.InputAPI())
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
userUpdateProducer := &producers.UserUpdateProducer{
Producer: base.KafkaProducer,
@ -47,7 +51,7 @@ func SetupClientAPIComponent(
}
consumer := consumers.NewOutputRoomEventConsumer(
base.Cfg, base.KafkaConsumer, accountsDB, base.QueryAPI(),
base.Cfg, base.KafkaConsumer, accountsDB, queryAPI,
)
if err := consumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start room server consumer")
@ -55,7 +59,7 @@ func SetupClientAPIComponent(
routing.Setup(
base.APIMux, *base.Cfg, roomserverProducer,
base.QueryAPI(), base.AliasAPI(), accountsDB, deviceDB,
queryAPI, aliasAPI, accountsDB, deviceDB,
federation, *keyRing,
userUpdateProducer, syncProducer,
)

7
src/github.com/matrix-org/dendrite/cmd/dendrite-client-api-server/main.go
View file

@ -21,7 +21,9 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("ClientAPI")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "ClientAPI")
defer base.Close() // nolint: errcheck
accountDB := base.CreateAccountsDB()
@ -30,8 +32,11 @@ func main() {
federation := base.CreateFederationClient()
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
alias, input, query := base.CreateHTTPRoomserverAPIs()
clientapi.SetupClientAPIComponent(
base, deviceDB, accountDB, federation, &keyRing,
alias, input, query,
)
base.SetupAndServeHTTP(string(base.Cfg.Listen.ClientAPI))

6
src/github.com/matrix-org/dendrite/cmd/dendrite-federation-api-server/main.go
View file

@ -21,7 +21,8 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("FederationAPI")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "FederationAPI")
defer base.Close() // nolint: errcheck
accountDB := base.CreateAccountsDB()
@ -29,8 +30,11 @@ func main() {
federation := base.CreateFederationClient()
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
alias, input, query := base.CreateHTTPRoomserverAPIs()
federationapi.SetupFederationAPIComponent(
base, accountDB, federation, &keyRing,
alias, input, query,
)
base.SetupAndServeHTTP(string(base.Cfg.Listen.FederationAPI))

7
src/github.com/matrix-org/dendrite/cmd/dendrite-federation-sender-server/main.go
View file

@ -20,13 +20,16 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("FederationSender")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "FederationSender")
defer base.Close() // nolint: errcheck
federation := base.CreateFederationClient()
_, _, query := base.CreateHTTPRoomserverAPIs()
federationsender.SetupFederationSenderComponent(
base, federation,
base, federation, query,
)
base.SetupAndServeHTTP(string(base.Cfg.Listen.FederationSender))

3
src/github.com/matrix-org/dendrite/cmd/dendrite-media-api-server/main.go
View file

@ -20,7 +20,8 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("MediaAPI")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "MediaAPI")
defer base.Close() // nolint: errcheck
deviceDB := base.CreateDeviceDB()

14
src/github.com/matrix-org/dendrite/cmd/dendrite-monolith-server/main.go
View file

@ -40,7 +40,8 @@ var (
)
func main() {
base, roomserverDB := basecomponent.NewBaseDendriteMonolith("Monolith")
cfg := basecomponent.ParseMonolithFlags()
base := basecomponent.NewBaseDendrite(cfg, "Monolith")
defer base.Close() // nolint: errcheck
accountDB := base.CreateAccountsDB()
@ -49,13 +50,14 @@ func main() {
federation := base.CreateFederationClient()
keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
clientapi.SetupClientAPIComponent(base, deviceDB, accountDB, federation, &keyRing)
federationapi.SetupFederationAPIComponent(base, accountDB, federation, &keyRing)
federationsender.SetupFederationSenderComponent(base, federation)
alias, input, query := roomserver.SetupRoomServerComponent(base)
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)
roomserver.SetupRoomServerComponentWithDB(base, roomserverDB)
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB)
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query)
httpHandler := common.WrapHandlerInCORS(base.APIMux)

3
src/github.com/matrix-org/dendrite/cmd/dendrite-public-rooms-api-server/main.go
View file

@ -20,7 +20,8 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("PublicRoomsAPI")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "PublicRoomsAPI")
defer base.Close() // nolint: errcheck
deviceDB := base.CreateDeviceDB()

3
src/github.com/matrix-org/dendrite/cmd/dendrite-room-server/main.go
View file

@ -22,7 +22,8 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("RoomServerAPI")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "RoomServerAPI")
defer base.Close() // nolint: errcheck
roomserver.SetupRoomServerComponent(base)

7
src/github.com/matrix-org/dendrite/cmd/dendrite-sync-api-server/main.go
View file

@ -20,13 +20,16 @@ import (
)
func main() {
base := basecomponent.NewBaseDendrite("SyncAPI")
cfg := basecomponent.ParseFlags()
base := basecomponent.NewBaseDendrite(cfg, "SyncAPI")
defer base.Close() // nolint: errcheck
deviceDB := base.CreateDeviceDB()
accountDB := base.CreateAccountsDB()
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB)
_, _, query := base.CreateHTTPRoomserverAPIs()
syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query)
base.SetupAndServeHTTP(string(base.Cfg.Listen.SyncAPI))
}

133
src/github.com/matrix-org/dendrite/common/basecomponent/base.go
View file

@ -1,4 +1,4 @@
// Copyright 2017 Vector Creations Ltd
// 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.
@ -16,7 +16,6 @@ package basecomponent
import (
"database/sql"
"flag"
"io"
"net/http"
"os"
@ -29,11 +28,6 @@ import (
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
"github.com/matrix-org/dendrite/common"
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"
"github.com/gorilla/mux"
sarama "gopkg.in/Shopify/sarama.v1"
@ -42,8 +36,6 @@ import (
"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.")
// 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
@ -52,10 +44,6 @@ var configPath = flag.String("config", "dendrite.yaml", "The path to the config
type BaseDendrite struct {
componentName string
tracerCloser io.Closer
queryAPI api.RoomserverQueryAPI
inputAPI api.RoomserverInputAPI
aliasAPI api.RoomserverAliasAPI
monolith bool
// APIMux should be used to register new public matrix api endpoints
APIMux *mux.Router
@ -64,60 +52,12 @@ type BaseDendrite struct {
KafkaProducer sarama.SyncProducer
}
// NewBaseDendrite creates a new instance to be used by a component. If running
// as a monolith then `NewBaseDendriteMonolith` should be used.
// 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(componentName string) *BaseDendrite {
base := newBaseDendrite(componentName, false)
// We're not a monolith so we can only use the HTTP versions
base.useHTTPRoomserverAPIs()
return base
}
// NewBaseDendriteMonolith is the same NewBaseDendrite, but indicates that all
// components will be in the same process. Allows using naffka and in-process
// roomserver APIs.
//
// It also connects to the room server databsae so that the monolith can use
// in-process versions of QueryAPI and co.
func NewBaseDendriteMonolith(componentName string) (*BaseDendrite, *roomserver_storage.Database) {
base := newBaseDendrite(componentName, true)
roomserverDB, err := roomserver_storage.Open(string(base.Cfg.Database.RoomServer))
if err != nil {
logrus.WithError(err).Panicf("failed to connect to room server db")
}
base.useInProcessRoomserverAPIs(roomserverDB)
return base, roomserverDB
}
// newBaseDendrite does the bulk of the work of NewBaseDendrite*, except setting
// up the roomserver APIs, which must be done by the callers.
func newBaseDendrite(componentName string, monolith bool) *BaseDendrite {
func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite {
common.SetupLogging(os.Getenv("LOG_DIR"))
flag.Parse()
if *configPath == "" {
logrus.Fatal("--config must be supplied")
}
var cfg *config.Dendrite
var err error
if monolith {
cfg, err = config.LoadMonolithic(*configPath)
} else {
cfg, err = config.Load(*configPath)
}
if err != nil {
logrus.Fatalf("Invalid config file: %s", err)
}
closer, err := cfg.SetupTracing("Dendrite" + componentName)
if err != nil {
logrus.WithError(err).Panicf("failed to start opentracing")
@ -132,7 +72,6 @@ func newBaseDendrite(componentName string, monolith bool) *BaseDendrite {
APIMux: mux.NewRouter(),
KafkaConsumer: kafkaConsumer,
KafkaProducer: kafkaProducer,
monolith: monolith,
}
}
@ -141,65 +80,13 @@ func (b *BaseDendrite) Close() error {
return b.tracerCloser.Close()
}
// useInProcessRoomserverAPIs sets up the AliasAPI, InputAPI and QueryAPI to hit
// the functions directly, rather than going through an RPC mechanism. Can only
// be used in a monolith set up.
func (b *BaseDendrite) useInProcessRoomserverAPIs(roomserverDB *roomserver_storage.Database) {
if !b.monolith {
logrus.Panic("Can only use in-process roomserver APIs if running as a monolith")
}
b.inputAPI = &roomserver_input.RoomserverInputAPI{
DB: roomserverDB,
Producer: b.KafkaProducer,
OutputRoomEventTopic: string(b.Cfg.Kafka.Topics.OutputRoomEvent),
}
b.queryAPI = &roomserver_query.RoomserverQueryAPI{
DB: roomserverDB,
}
b.aliasAPI = &roomserver_alias.RoomserverAliasAPI{
DB: roomserverDB,
Cfg: b.Cfg,
InputAPI: b.inputAPI,
QueryAPI: b.queryAPI,
}
}
// useHTTPRoomserverAPIs sets up the AliasAPI, InputAPI and QueryAPI to hit the
// roomserver over HTTP.
func (b *BaseDendrite) useHTTPRoomserverAPIs() {
b.queryAPI = api.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
b.inputAPI = api.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), nil)
b.aliasAPI = api.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), nil)
}
// QueryAPI gets an implementation of RoomserverQueryAPI
func (b *BaseDendrite) QueryAPI() api.RoomserverQueryAPI {
if b.queryAPI == nil {
logrus.Panic("RoomserverAPIs not created")
}
return b.queryAPI
}
// AliasAPI gets an implementation of RoomserverAliasAPI
func (b *BaseDendrite) AliasAPI() api.RoomserverAliasAPI {
if b.aliasAPI == nil {
logrus.Panic("RoomserverAPIs not created")
}
return b.aliasAPI
}
// InputAPI gets an implementation of RoomserverInputAPI
func (b *BaseDendrite) InputAPI() api.RoomserverInputAPI {
if b.inputAPI == nil {
logrus.Panic("RoomserverAPIs not created")
}
return b.inputAPI
// 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

61
src/github.com/matrix-org/dendrite/common/basecomponent/flags.go Normal file
View file

@ -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
}

8
src/github.com/matrix-org/dendrite/federationapi/federationapi.go
View file

@ -17,6 +17,7 @@ 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"
@ -30,11 +31,14 @@ func SetupFederationAPIComponent(
accountsDB *accounts.Database,
federation *gomatrixserverlib.FederationClient,
keyRing *gomatrixserverlib.KeyRing,
aliasAPI api.RoomserverAliasAPI,
inputAPI api.RoomserverInputAPI,
queryAPI api.RoomserverQueryAPI,
) {
roomserverProducer := producers.NewRoomserverProducer(base.InputAPI())
roomserverProducer := producers.NewRoomserverProducer(inputAPI)
routing.Setup(
base.APIMux, *base.Cfg, base.QueryAPI(), base.AliasAPI(),
base.APIMux, *base.Cfg, queryAPI, aliasAPI,
roomserverProducer, *keyRing, federation, accountsDB,
)
}

4
src/github.com/matrix-org/dendrite/federationsender/federationsender.go
View file

@ -19,6 +19,7 @@ import (
"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"
)
@ -28,6 +29,7 @@ import (
func SetupFederationSenderComponent(
base *basecomponent.BaseDendrite,
federation *gomatrixserverlib.FederationClient,
queryAPI api.RoomserverQueryAPI,
) {
federationSenderDB, err := storage.NewDatabase(string(base.Cfg.Database.FederationSender))
if err != nil {
@ -38,7 +40,7 @@ func SetupFederationSenderComponent(
consumer := consumers.NewOutputRoomEventConsumer(
base.Cfg, base.KafkaConsumer, queues,
federationSenderDB, base.QueryAPI(),
federationSenderDB, queryAPI,
)
if err = consumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start room server consumer")

20
src/github.com/matrix-org/dendrite/roomserver/roomserver.go
View file

@ -17,6 +17,8 @@ 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"
@ -25,24 +27,18 @@ import (
"github.com/sirupsen/logrus"
)
// SetupRoomServerComponent sets up and registers HTTP handlers for the RoomServer
// component.
// 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")
}
SetupRoomServerComponentWithDB(base, roomserverDB)
}
// SetupRoomServerComponentWithDB sets up and registers HTTP handlers for the RoomServer
// component, reusing the given room server database instance.
func SetupRoomServerComponentWithDB(
base *basecomponent.BaseDendrite, roomserverDB *storage.Database,
) {
inputAPI := input.RoomserverInputAPI{
DB: roomserverDB,
Producer: base.KafkaProducer,
@ -63,4 +59,6 @@ func SetupRoomServerComponentWithDB(
}
aliasAPI.SetupHTTP(http.DefaultServeMux)
return &aliasAPI, &inputAPI, &queryAPI
}

4
src/github.com/matrix-org/dendrite/syncapi/syncapi.go
View file

@ -21,6 +21,7 @@ import (
"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"
@ -36,6 +37,7 @@ 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 {
@ -56,7 +58,7 @@ func SetupSyncAPIComponent(
requestPool := sync.NewRequestPool(syncDB, notifier, accountsDB)
roomConsumer := consumers.NewOutputRoomEventConsumer(
base.Cfg, base.KafkaConsumer, notifier, syncDB, base.QueryAPI(),
base.Cfg, base.KafkaConsumer, notifier, syncDB, queryAPI,
)
if err = roomConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start room server consumer")