hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-08-01 22:02:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Consolidation of roomserver APIs (#994)

Browse source 
* Consolidation of roomserver APIs

* Comment out alias tests for now, they are broken

* Wire AS API into roomserver again

* Roomserver didn't take asAPI param before so return to that

* Prevent roomserver asking AS API for alias info

* Rename some files

* Remove alias_test, incoherent tests and unwanted appservice integration

* Remove FS API inject on syncapi component
This commit is contained in:
Neil Alexander 2020-05-01 10:48:17 +01:00 • committed by GitHub
parent ebbfc12592
commit e15f6676ac
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
72 changed files with 894 additions and 1170 deletions
Download patch file Download diff file Expand all files Collapse all files

22
syncapi/consumers/roomserver.go
View file

@ -32,10 +32,10 @@ import (
// OutputRoomEventConsumer consumes events that originated in the room server.
type OutputRoomEventConsumer struct {
roomServerConsumer *common.ContinualConsumer
db storage.Database
notifier *sync.Notifier
query api.RoomserverQueryAPI
rsAPI api.RoomserverInternalAPI
rsConsumer *common.ContinualConsumer
db storage.Database
notifier *sync.Notifier
}
// NewOutputRoomEventConsumer creates a new OutputRoomEventConsumer. Call Start() to begin consuming from room servers.
@ -44,7 +44,7 @@ func NewOutputRoomEventConsumer(
kafkaConsumer sarama.Consumer,
n *sync.Notifier,
store storage.Database,
queryAPI api.RoomserverQueryAPI,
rsAPI api.RoomserverInternalAPI,
) *OutputRoomEventConsumer {
consumer := common.ContinualConsumer{
@ -53,10 +53,10 @@ func NewOutputRoomEventConsumer(
PartitionStore: store,
}
s := &OutputRoomEventConsumer{
roomServerConsumer: &consumer,
db: store,
notifier: n,
query: queryAPI,
rsConsumer: &consumer,
db: store,
notifier: n,
rsAPI: rsAPI,
}
consumer.ProcessMessage = s.onMessage
@ -65,7 +65,7 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error {
return s.roomServerConsumer.Start()
return s.rsConsumer.Start()
}
// onMessage is called when the sync server receives a new event from the room server output log.
@ -226,7 +226,7 @@ func (s *OutputRoomEventConsumer) lookupStateEvents(
// from the roomserver using the query API.
eventReq := api.QueryEventsByIDRequest{EventIDs: missing}
var eventResp api.QueryEventsByIDResponse
if err := s.query.QueryEventsByID(context.TODO(), &eventReq, &eventResp); err != nil {
if err := s.rsAPI.QueryEventsByID(context.TODO(), &eventReq, &eventResp); err != nil {
return nil, err
}

8
syncapi/routing/messages.go
View file

@ -34,7 +34,7 @@ import (
type messagesReq struct {
ctx context.Context
db storage.Database
queryAPI api.RoomserverQueryAPI
rsAPI api.RoomserverInternalAPI
federation *gomatrixserverlib.FederationClient
cfg *config.Dendrite
roomID string
@ -59,7 +59,7 @@ const defaultMessagesLimit = 10
func OnIncomingMessagesRequest(
req *http.Request, db storage.Database, roomID string,
federation *gomatrixserverlib.FederationClient,
queryAPI api.RoomserverQueryAPI,
rsAPI api.RoomserverInternalAPI,
cfg *config.Dendrite,
) util.JSONResponse {
var err error
@ -135,7 +135,7 @@ func OnIncomingMessagesRequest(
mReq := messagesReq{
ctx: req.Context(),
db: db,
queryAPI: queryAPI,
rsAPI: rsAPI,
federation: federation,
cfg: cfg,
roomID: roomID,
@ -360,7 +360,7 @@ func (r *messagesReq) handleNonEmptyEventsSlice(streamEvents []types.StreamEvent
// the room or sending the request.
func (r *messagesReq) backfill(roomID string, fromEventIDs []string, limit int) ([]gomatrixserverlib.HeaderedEvent, error) {
var res api.QueryBackfillResponse
err := r.queryAPI.QueryBackfill(context.Background(), &api.QueryBackfillRequest{
err := r.rsAPI.QueryBackfill(context.Background(), &api.QueryBackfillRequest{
RoomID: roomID,
EarliestEventsIDs: fromEventIDs,
Limit: limit,

4
syncapi/routing/routing.go
View file

@ -40,7 +40,7 @@ const pathPrefixR0 = "/_matrix/client/r0"
func Setup(
apiMux *mux.Router, srp *sync.RequestPool, syncDB storage.Database,
deviceDB devices.Database, federation *gomatrixserverlib.FederationClient,
queryAPI api.RoomserverQueryAPI,
rsAPI api.RoomserverInternalAPI,
cfg *config.Dendrite,
) {
r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
@ -61,6 +61,6 @@ func Setup(
if err != nil {
return util.ErrorResponse(err)
}
return OnIncomingMessagesRequest(req, syncDB, vars["roomID"], federation, queryAPI, cfg)
return OnIncomingMessagesRequest(req, syncDB, vars["roomID"], federation, rsAPI, cfg)
})).Methods(http.MethodGet, http.MethodOptions)
}

6
syncapi/syncapi.go
View file

@ -38,7 +38,7 @@ func SetupSyncAPIComponent(
base *basecomponent.BaseDendrite,
deviceDB devices.Database,
accountsDB accounts.Database,
queryAPI api.RoomserverQueryAPI,
rsAPI api.RoomserverInternalAPI,
federation *gomatrixserverlib.FederationClient,
cfg *config.Dendrite,
) {
@ -61,7 +61,7 @@ func SetupSyncAPIComponent(
requestPool := sync.NewRequestPool(syncDB, notifier, accountsDB)
roomConsumer := consumers.NewOutputRoomEventConsumer(
base.Cfg, base.KafkaConsumer, notifier, syncDB, queryAPI,
base.Cfg, base.KafkaConsumer, notifier, syncDB, rsAPI,
)
if err = roomConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start room server consumer")
@ -81,5 +81,5 @@ func SetupSyncAPIComponent(
logrus.WithError(err).Panicf("failed to start typing server consumer")
}
routing.Setup(base.APIMux, requestPool, syncDB, deviceDB, federation, queryAPI, cfg)
routing.Setup(base.APIMux, requestPool, syncDB, deviceDB, federation, rsAPI, cfg)
}