mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-12 09:23:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into filter_sync_impl

Browse source
This commit is contained in:
Crom (Thibaut CHARLES) 2018-01-02 13:09:07 +01:00
parent 9af5e73fdf 27c335438f
commit cdd6b0a3d3
No known key found for this signature in database
GPG key ID: 45A3D5F880B9E6D0
19 changed files with 782 additions and 843 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

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

77
src/github.com/matrix-org/dendrite/clientapi/routing/createroom.go
View file

@ -15,7 +15,6 @@
package routing
import (
"encoding/json"
"fmt"
"net/http"
"strings"
@ -43,10 +42,28 @@ type createRoomRequest struct {
Topic string `json:"topic"`
Preset string `json:"preset"`
CreationContent map[string]interface{} `json:"creation_content"`
InitialState json.RawMessage `json:"initial_state"` // TODO
InitialState []fledglingEvent `json:"initial_state"`
RoomAliasName string `json:"room_alias_name"`
GuestCanJoin bool `json:"guest_can_join"`
}
const (
presetPrivateChat = "private_chat"
presetTrustedPrivateChat = "trusted_private_chat"
presetPublicChat = "public_chat"
)
const (
joinRulePublic = "public"
joinRuleInvite = "invite"
)
const (
historyVisibilityShared = "shared"
// TODO: These should be implemented once history visibility is implemented
// historyVisibilityWorldReadable = "world_readable"
// historyVisibilityInvited = "invited"
)
func (r createRoomRequest) Validate() *util.JSONResponse {
whitespace := "\t\n\x0b\x0c\r " // https://docs.python.org/2/library/string.html#string.whitespace
// https://github.com/matrix-org/synapse/blob/v0.19.2/synapse/handlers/room.py#L81
@ -70,6 +87,16 @@ func (r createRoomRequest) Validate() *util.JSONResponse {
}
}
}
switch r.Preset {
case presetPrivateChat, presetTrustedPrivateChat, presetPublicChat:
break
default:
return &util.JSONResponse{
Code: 400,
JSON: jsonerror.BadJSON("preset must be any of 'private_chat', 'trusted_private_chat', 'public_chat'"),
}
}
return nil
}
@ -81,9 +108,9 @@ type createRoomResponse struct {
// fledglingEvent is a helper representation of an event used when creating many events in succession.
type fledglingEvent struct {
Type string
StateKey string
Content interface{}
Type string `json:"type"`
StateKey string `json:"state_key"`
Content interface{} `json:"content"`
}
// CreateRoom implements /createRoom
@ -141,6 +168,20 @@ func createRoom(req *http.Request, device *authtypes.Device,
AvatarURL: profile.AvatarURL,
}
var joinRules, historyVisibility string
switch r.Preset {
case presetPrivateChat:
joinRules = joinRuleInvite
historyVisibility = historyVisibilityShared
case presetTrustedPrivateChat:
joinRules = joinRuleInvite
historyVisibility = historyVisibilityShared
// TODO If trusted_private_chat, all invitees are given the same power level as the room creator.
case presetPublicChat:
joinRules = joinRulePublic
historyVisibility = historyVisibilityShared
}
var builtEvents []gomatrixserverlib.Event
// send events into the room in order of:
@ -151,7 +192,7 @@ func createRoom(req *http.Request, device *authtypes.Device,
// 5- m.room.join_rules
// 6- m.room.history_visibility
// 7- m.room.guest_access (opt)
// 8- other initial state items TODO
// 8- other initial state items
// 9- m.room.name (opt)
// 10- m.room.topic (opt)
// 11- invite events (opt) - with is_direct flag if applicable TODO
@ -166,16 +207,22 @@ func createRoom(req *http.Request, device *authtypes.Device,
{"m.room.member", userID, membershipContent},
{"m.room.power_levels", "", common.InitialPowerLevelsContent(userID)},
// TODO: m.room.canonical_alias
{"m.room.join_rules", "", common.JoinRulesContent{JoinRule: "public"}}, // FIXME: Allow this to be changed
{"m.room.history_visibility", "", common.HistoryVisibilityContent{HistoryVisibility: "joined"}}, // FIXME: Allow this to be changed
{"m.room.guest_access", "", common.GuestAccessContent{GuestAccess: "can_join"}}, // FIXME: Allow this to be changed
// TODO: Other initial state items
{"m.room.name", "", common.NameContent{Name: r.Name}}, // FIXME: Only send the name event if a name is supplied, to avoid sending a false room name removal event
{"m.room.topic", "", common.TopicContent{Topic: r.Topic}},
// TODO: invite events
// TODO: 3pid invite events
// TODO: m.room.aliases
{"m.room.join_rules", "", common.JoinRulesContent{JoinRule: joinRules}},
{"m.room.history_visibility", "", common.HistoryVisibilityContent{HistoryVisibility: historyVisibility}},
}
if r.GuestCanJoin {
eventsToMake = append(eventsToMake, fledglingEvent{"m.room.guest_access", "", common.GuestAccessContent{GuestAccess: "can_join"}})
}
eventsToMake = append(eventsToMake, r.InitialState...)
if r.Name != "" {
eventsToMake = append(eventsToMake, fledglingEvent{"m.room.name", "", common.NameContent{Name: r.Name}})
}
if r.Topic != "" {
eventsToMake = append(eventsToMake, fledglingEvent{"m.room.topic", "", common.TopicContent{Topic: r.Topic}})
}
// TODO: invite events
// TODO: 3pid invite events
// TODO: m.room.aliases
authEvents := gomatrixserverlib.NewAuthEvents(nil)
for i, e := range eventsToMake {

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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
}

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

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

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

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

40
src/github.com/matrix-org/dendrite/mediaapi/mediaapi.go Normal file
View file

@ -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(),
)
}

37
src/github.com/matrix-org/dendrite/publicroomsapi/publicroomsapi.go Normal file
View file

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

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

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

13
src/github.com/matrix-org/dendrite/syncapi/storage/output_room_events_table.go
View file

@ -65,7 +65,7 @@ const selectEventsSQL = "" +
const selectRecentEventsSQL = "" +
"SELECT id, event_json, device_id, transaction_id FROM syncapi_output_room_events" +
" WHERE room_id = $1 AND id > $2 AND id <= $3" +
" ORDER BY id DESC LIMIT $4"
" ORDER BY id ASC LIMIT $4"
const selectMaxEventIDSQL = "" +
"SELECT MAX(id) FROM syncapi_output_room_events"
@ -234,9 +234,7 @@ func (s *outputRoomEventsStatements) selectRecentEvents(
if err != nil {
return nil, err
}
// reverse the order because [0] is the newest event due to the ORDER BY in SQL-land. The reverse order makes [0] the oldest event,
// which is correct for /sync responses.
return reverseEvents(events), nil
return events, nil
}
// Events returns the events for the given event IDs. Returns an error if any one of the event IDs given are missing
@ -287,10 +285,3 @@ func rowsToStreamEvents(rows *sql.Rows) ([]streamEvent, error) {
}
return result, nil
}
func reverseEvents(input []streamEvent) (output []streamEvent) {
for i := len(input) - 1; i >= 0; i-- {
output = append(output, input[i])
}
return
}

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

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