mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-11 08:53:11 -06:00
Introduced new Kafka topic and producer
This commit is contained in:
parent
89b792cfad
commit
6a320d6ab6
|
|
@ -22,7 +22,7 @@ import (
|
||||||
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
|
||||||
"github.com/matrix-org/dendrite/clientapi/httputil"
|
"github.com/matrix-org/dendrite/clientapi/httputil"
|
||||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||||
// "github.com/matrix-org/gomatrixserverlib"
|
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||||
"github.com/matrix-org/util"
|
"github.com/matrix-org/util"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -70,6 +70,7 @@ func GetProfile(
|
||||||
// AvatarURL implements GET and PUT /profile/{userID}/avatar_url
|
// AvatarURL implements GET and PUT /profile/{userID}/avatar_url
|
||||||
func AvatarURL(
|
func AvatarURL(
|
||||||
req *http.Request, accountDB *accounts.Database, userID string,
|
req *http.Request, accountDB *accounts.Database, userID string,
|
||||||
|
userUpdatesProducer *producers.RoomserverProducer,
|
||||||
) util.JSONResponse {
|
) util.JSONResponse {
|
||||||
if req.Method == "GET" {
|
if req.Method == "GET" {
|
||||||
localpart := getLocalPart(userID)
|
localpart := getLocalPart(userID)
|
||||||
|
|
@ -120,6 +121,7 @@ func AvatarURL(
|
||||||
// DisplayName implements GET and PUT /profile/{userID}/displayname
|
// DisplayName implements GET and PUT /profile/{userID}/displayname
|
||||||
func DisplayName(
|
func DisplayName(
|
||||||
req *http.Request, accountDB *accounts.Database, userID string,
|
req *http.Request, accountDB *accounts.Database, userID string,
|
||||||
|
userUpdatesProducer *producers.RoomserverProducer,
|
||||||
) util.JSONResponse {
|
) util.JSONResponse {
|
||||||
if req.Method == "GET" {
|
if req.Method == "GET" {
|
||||||
localpart := getLocalPart(userID)
|
localpart := getLocalPart(userID)
|
||||||
|
|
|
||||||
|
|
@ -46,6 +46,7 @@ func Setup(
|
||||||
deviceDB *devices.Database,
|
deviceDB *devices.Database,
|
||||||
federation *gomatrixserverlib.FederationClient,
|
federation *gomatrixserverlib.FederationClient,
|
||||||
keyRing gomatrixserverlib.KeyRing,
|
keyRing gomatrixserverlib.KeyRing,
|
||||||
|
userUpdatesProducer *producers.RoomserverProducer,
|
||||||
) {
|
) {
|
||||||
apiMux := mux.NewRouter()
|
apiMux := mux.NewRouter()
|
||||||
|
|
||||||
|
|
@ -171,14 +172,14 @@ func Setup(
|
||||||
r0mux.Handle("/profile/{userID}/avatar_url",
|
r0mux.Handle("/profile/{userID}/avatar_url",
|
||||||
common.MakeAPI("profile_avatar_url", func(req *http.Request) util.JSONResponse {
|
common.MakeAPI("profile_avatar_url", func(req *http.Request) util.JSONResponse {
|
||||||
vars := mux.Vars(req)
|
vars := mux.Vars(req)
|
||||||
return readers.AvatarURL(req, accountDB, vars["userID"])
|
return readers.AvatarURL(req, accountDB, vars["userID"], userUpdatesProducer)
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
r0mux.Handle("/profile/{userID}/displayname",
|
r0mux.Handle("/profile/{userID}/displayname",
|
||||||
common.MakeAPI("profile_displayname", func(req *http.Request) util.JSONResponse {
|
common.MakeAPI("profile_displayname", func(req *http.Request) util.JSONResponse {
|
||||||
vars := mux.Vars(req)
|
vars := mux.Vars(req)
|
||||||
return readers.DisplayName(req, accountDB, vars["userID"])
|
return readers.DisplayName(req, accountDB, vars["userID"], userUpdatesProducer)
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -53,6 +53,9 @@ func main() {
|
||||||
roomserverProducer, err := producers.NewRoomserverProducer(
|
roomserverProducer, err := producers.NewRoomserverProducer(
|
||||||
cfg.Kafka.Addresses, string(cfg.Kafka.Topics.InputRoomEvent),
|
cfg.Kafka.Addresses, string(cfg.Kafka.Topics.InputRoomEvent),
|
||||||
)
|
)
|
||||||
|
userUpdatesProducer, err := producers.NewRoomserverProducer(
|
||||||
|
cfg.Kafka.Addresses, string(cfg.Kafka.Topics.UserUpdates),
|
||||||
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Panicf("Failed to setup kafka producers(%q): %s", cfg.Kafka.Addresses, err)
|
log.Panicf("Failed to setup kafka producers(%q): %s", cfg.Kafka.Addresses, err)
|
||||||
}
|
}
|
||||||
|
|
@ -86,7 +89,7 @@ func main() {
|
||||||
log.Info("Starting client API server on ", cfg.Listen.ClientAPI)
|
log.Info("Starting client API server on ", cfg.Listen.ClientAPI)
|
||||||
routing.Setup(
|
routing.Setup(
|
||||||
http.DefaultServeMux, http.DefaultClient, *cfg, roomserverProducer,
|
http.DefaultServeMux, http.DefaultClient, *cfg, roomserverProducer,
|
||||||
queryAPI, accountDB, deviceDB, federation, keyRing,
|
queryAPI, accountDB, deviceDB, federation, keyRing, userUpdatesProducer,
|
||||||
)
|
)
|
||||||
log.Fatal(http.ListenAndServe(string(cfg.Listen.ClientAPI), nil))
|
log.Fatal(http.ListenAndServe(string(cfg.Listen.ClientAPI), nil))
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -99,6 +99,8 @@ type Dendrite struct {
|
||||||
InputRoomEvent Topic `yaml:"input_room_event"`
|
InputRoomEvent Topic `yaml:"input_room_event"`
|
||||||
// Topic for roomserver/api.OutputRoomEvent events.
|
// Topic for roomserver/api.OutputRoomEvent events.
|
||||||
OutputRoomEvent Topic `yaml:"output_room_event"`
|
OutputRoomEvent Topic `yaml:"output_room_event"`
|
||||||
|
// Topic for user updates (profile, presence)
|
||||||
|
UserUpdates Topic `yaml:"user_updates"`
|
||||||
}
|
}
|
||||||
} `yaml:"kafka"`
|
} `yaml:"kafka"`
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue