mirror of
https://github.com/hoernschen/dendrite.git
synced 2024-12-26 15:08:28 +00:00
Federation sender API remodel (#988)
* Define an input API for the federationsender * Wiring for rooomserver input API and federation sender input API * Whoops, commit common too * Merge input API into query API * Rename FederationSenderQueryAPI to FederationSenderInternalAPI * Fix dendritejs * Rename Input to Perform * Fix a couple of inputs -> performs * Remove needless storage interface, add comments
This commit is contained in:
parent
a4b9edb28e
commit
a308e61331
23 changed files with 293 additions and 136 deletions
|
@ -44,7 +44,7 @@ func SetupClientAPIComponent(
|
|||
eduInputAPI eduServerAPI.EDUServerInputAPI,
|
||||
asAPI appserviceAPI.AppServiceQueryAPI,
|
||||
transactionsCache *transactions.Cache,
|
||||
fedSenderAPI federationSenderAPI.FederationSenderQueryAPI,
|
||||
fsAPI federationSenderAPI.FederationSenderInternalAPI,
|
||||
) {
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI, queryAPI)
|
||||
eduProducer := producers.NewEDUServerProducer(eduInputAPI)
|
||||
|
@ -69,6 +69,6 @@ func SetupClientAPIComponent(
|
|||
routing.Setup(
|
||||
base.APIMux, base.Cfg, roomserverProducer, queryAPI, aliasAPI, asAPI,
|
||||
accountsDB, deviceDB, federation, *keyRing, userUpdateProducer,
|
||||
syncProducer, eduProducer, transactionsCache, fedSenderAPI,
|
||||
syncProducer, eduProducer, transactionsCache, fsAPI,
|
||||
)
|
||||
}
|
||||
|
|
|
@ -47,7 +47,7 @@ func DirectoryRoom(
|
|||
federation *gomatrixserverlib.FederationClient,
|
||||
cfg *config.Dendrite,
|
||||
rsAPI roomserverAPI.RoomserverAliasAPI,
|
||||
fedSenderAPI federationSenderAPI.FederationSenderQueryAPI,
|
||||
fedSenderAPI federationSenderAPI.FederationSenderInternalAPI,
|
||||
) util.JSONResponse {
|
||||
_, domain, err := gomatrixserverlib.SplitID('#', roomAlias)
|
||||
if err != nil {
|
||||
|
|
|
@ -60,7 +60,7 @@ func Setup(
|
|||
syncProducer *producers.SyncAPIProducer,
|
||||
eduProducer *producers.EDUServerProducer,
|
||||
transactionsCache *transactions.Cache,
|
||||
federationSender federationSenderAPI.FederationSenderQueryAPI,
|
||||
federationSender federationSenderAPI.FederationSenderInternalAPI,
|
||||
) {
|
||||
|
||||
apiMux.Handle("/_matrix/client/versions",
|
||||
|
|
|
@ -37,12 +37,12 @@ func main() {
|
|||
|
||||
asQuery := base.CreateHTTPAppServiceAPIs()
|
||||
alias, input, query := base.CreateHTTPRoomserverAPIs()
|
||||
fedSenderAPI := base.CreateHTTPFederationSenderAPIs()
|
||||
fsAPI := base.CreateHTTPFederationSenderAPIs()
|
||||
eduInputAPI := eduserver.SetupEDUServerComponent(base, cache.New())
|
||||
|
||||
clientapi.SetupClientAPIComponent(
|
||||
base, deviceDB, accountDB, federation, &keyRing,
|
||||
alias, input, query, eduInputAPI, asQuery, transactions.New(), fedSenderAPI,
|
||||
alias, input, query, eduInputAPI, asQuery, transactions.New(), fsAPI,
|
||||
)
|
||||
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Bind.ClientAPI), string(base.Cfg.Listen.ClientAPI))
|
||||
|
|
|
@ -153,15 +153,15 @@ func main() {
|
|||
asQuery := appservice.SetupAppServiceAPIComponent(
|
||||
&base.Base, accountDB, deviceDB, federation, alias, query, transactions.New(),
|
||||
)
|
||||
fedSenderAPI := federationsender.SetupFederationSenderComponent(&base.Base, federation, query, input)
|
||||
fsAPI := federationsender.SetupFederationSenderComponent(&base.Base, federation, query, input)
|
||||
|
||||
clientapi.SetupClientAPIComponent(
|
||||
&base.Base, deviceDB, accountDB,
|
||||
federation, &keyRing, alias, input, query,
|
||||
eduInputAPI, asQuery, transactions.New(), fedSenderAPI,
|
||||
eduInputAPI, asQuery, transactions.New(), fsAPI,
|
||||
)
|
||||
eduProducer := producers.NewEDUServerProducer(eduInputAPI)
|
||||
federationapi.SetupFederationAPIComponent(&base.Base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fedSenderAPI, eduProducer)
|
||||
federationapi.SetupFederationAPIComponent(&base.Base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fsAPI, eduProducer)
|
||||
mediaapi.SetupMediaAPIComponent(&base.Base, deviceDB)
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabaseWithPubSub(string(base.Base.Cfg.Database.PublicRoomsAPI), base.LibP2PPubsub)
|
||||
if err != nil {
|
||||
|
|
|
@ -32,7 +32,7 @@ func main() {
|
|||
deviceDB := base.CreateDeviceDB()
|
||||
keyDB := base.CreateKeyDB()
|
||||
federation := base.CreateFederationClient()
|
||||
federationSender := base.CreateHTTPFederationSenderAPIs()
|
||||
fsAPI := base.CreateHTTPFederationSenderAPIs()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB, cfg.Matrix.KeyPerspectives)
|
||||
|
||||
alias, input, query := base.CreateHTTPRoomserverAPIs()
|
||||
|
@ -42,7 +42,7 @@ func main() {
|
|||
|
||||
federationapi.SetupFederationAPIComponent(
|
||||
base, accountDB, deviceDB, federation, &keyRing,
|
||||
alias, input, query, asQuery, federationSender, eduProducer,
|
||||
alias, input, query, asQuery, fsAPI, eduProducer,
|
||||
)
|
||||
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Bind.FederationAPI), string(base.Cfg.Listen.FederationAPI))
|
||||
|
|
|
@ -62,15 +62,16 @@ func main() {
|
|||
asQuery := appservice.SetupAppServiceAPIComponent(
|
||||
base, accountDB, deviceDB, federation, alias, query, transactions.New(),
|
||||
)
|
||||
fedSenderAPI := federationsender.SetupFederationSenderComponent(base, federation, query, input)
|
||||
fsAPI := federationsender.SetupFederationSenderComponent(base, federation, query, input)
|
||||
input.SetFederationSenderAPI(fsAPI)
|
||||
|
||||
clientapi.SetupClientAPIComponent(
|
||||
base, deviceDB, accountDB,
|
||||
federation, &keyRing, alias, input, query,
|
||||
eduInputAPI, asQuery, transactions.New(), fedSenderAPI,
|
||||
eduInputAPI, asQuery, transactions.New(), fsAPI,
|
||||
)
|
||||
eduProducer := producers.NewEDUServerProducer(eduInputAPI)
|
||||
federationapi.SetupFederationAPIComponent(base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fedSenderAPI, eduProducer)
|
||||
federationapi.SetupFederationAPIComponent(base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fsAPI, eduProducer)
|
||||
mediaapi.SetupMediaAPIComponent(base, deviceDB)
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI))
|
||||
if err != nil {
|
||||
|
|
|
@ -30,7 +30,9 @@ func main() {
|
|||
federation := base.CreateFederationClient()
|
||||
keyRing := keydb.CreateKeyRing(federation.Client, keyDB, cfg.Matrix.KeyPerspectives)
|
||||
|
||||
roomserver.SetupRoomServerComponent(base, keyRing, federation)
|
||||
fsAPI := base.CreateHTTPFederationSenderAPIs()
|
||||
_, input, _ := roomserver.SetupRoomServerComponent(base, keyRing, federation)
|
||||
input.SetFederationSenderAPI(fsAPI)
|
||||
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Bind.RoomServer), string(base.Cfg.Listen.RoomServer))
|
||||
|
||||
|
|
|
@ -129,6 +129,7 @@ func main() {
|
|||
base, accountDB, deviceDB, federation, alias, query, transactions.New(),
|
||||
)
|
||||
fedSenderAPI := federationsender.SetupFederationSenderComponent(base, federation, query, input)
|
||||
input.SetFederationSenderAPI(fedSenderAPI)
|
||||
|
||||
clientapi.SetupClientAPIComponent(
|
||||
base, deviceDB, accountDB,
|
||||
|
|
|
@ -149,12 +149,12 @@ func (b *BaseDendrite) CreateHTTPEDUServerAPIs() eduServerAPI.EDUServerInputAPI
|
|||
return e
|
||||
}
|
||||
|
||||
// CreateHTTPFederationSenderAPIs returns FederationSenderQueryAPI for hitting
|
||||
// CreateHTTPFederationSenderAPIs returns FederationSenderInternalAPI for hitting
|
||||
// the federation sender over HTTP
|
||||
func (b *BaseDendrite) CreateHTTPFederationSenderAPIs() federationSenderAPI.FederationSenderQueryAPI {
|
||||
f, err := federationSenderAPI.NewFederationSenderQueryAPIHTTP(b.Cfg.FederationSenderURL(), b.httpClient)
|
||||
func (b *BaseDendrite) CreateHTTPFederationSenderAPIs() federationSenderAPI.FederationSenderInternalAPI {
|
||||
f, err := federationSenderAPI.NewFederationSenderInternalAPIHTTP(b.Cfg.FederationSenderURL(), b.httpClient)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewFederationSenderQueryAPIHTTP failed", b.httpClient)
|
||||
logrus.WithError(err).Panic("NewFederationSenderInternalAPIHTTP failed", b.httpClient)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
|
|
@ -40,7 +40,7 @@ func SetupFederationAPIComponent(
|
|||
inputAPI roomserverAPI.RoomserverInputAPI,
|
||||
queryAPI roomserverAPI.RoomserverQueryAPI,
|
||||
asAPI appserviceAPI.AppServiceQueryAPI,
|
||||
federationSenderAPI federationSenderAPI.FederationSenderQueryAPI,
|
||||
federationSenderAPI federationSenderAPI.FederationSenderInternalAPI,
|
||||
eduProducer *producers.EDUServerProducer,
|
||||
) {
|
||||
roomserverProducer := producers.NewRoomserverProducer(inputAPI, queryAPI)
|
||||
|
|
|
@ -33,7 +33,7 @@ func RoomAliasToID(
|
|||
federation *gomatrixserverlib.FederationClient,
|
||||
cfg *config.Dendrite,
|
||||
aliasAPI roomserverAPI.RoomserverAliasAPI,
|
||||
senderAPI federationSenderAPI.FederationSenderQueryAPI,
|
||||
senderAPI federationSenderAPI.FederationSenderInternalAPI,
|
||||
) util.JSONResponse {
|
||||
roomAlias := httpReq.FormValue("room_alias")
|
||||
if roomAlias == "" {
|
||||
|
|
|
@ -49,7 +49,7 @@ func Setup(
|
|||
asAPI appserviceAPI.AppServiceQueryAPI,
|
||||
producer *producers.RoomserverProducer,
|
||||
eduProducer *producers.EDUServerProducer,
|
||||
federationSenderAPI federationSenderAPI.FederationSenderQueryAPI,
|
||||
federationSenderAPI federationSenderAPI.FederationSenderInternalAPI,
|
||||
keys gomatrixserverlib.KeyRing,
|
||||
federation *gomatrixserverlib.FederationClient,
|
||||
accountDB accounts.Database,
|
||||
|
|
53
federationsender/api/api.go
Normal file
53
federationsender/api/api.go
Normal file
|
@ -0,0 +1,53 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// FederationSenderInternalAPI is used to query information from the federation sender.
|
||||
type FederationSenderInternalAPI interface {
|
||||
// Query the joined hosts and the membership events accounting for their participation in a room.
|
||||
// Note that if a server has multiple users in the room, it will have multiple entries in the returned slice.
|
||||
// See `QueryJoinedHostServerNamesInRoom` for a de-duplicated version.
|
||||
QueryJoinedHostsInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostsInRoomRequest,
|
||||
response *QueryJoinedHostsInRoomResponse,
|
||||
) error
|
||||
// Query the server names of the joined hosts in a room.
|
||||
// Unlike QueryJoinedHostsInRoom, this function returns a de-duplicated slice
|
||||
// containing only the server names (without information for membership events).
|
||||
QueryJoinedHostServerNamesInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostServerNamesInRoomRequest,
|
||||
response *QueryJoinedHostServerNamesInRoomResponse,
|
||||
) error
|
||||
// Handle an instruction to make_join & send_join with a remote server.
|
||||
PerformJoinRequest(
|
||||
ctx context.Context,
|
||||
request *PerformJoinRequest,
|
||||
response *PerformJoinResponse,
|
||||
) error
|
||||
// Handle an instruction to make_leave & send_leave with a remote server.
|
||||
PerformLeaveRequest(
|
||||
ctx context.Context,
|
||||
request *PerformLeaveRequest,
|
||||
response *PerformLeaveResponse,
|
||||
) error
|
||||
}
|
||||
|
||||
// NewFederationSenderInternalAPIHTTP creates a FederationSenderInternalAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil an error is returned
|
||||
func NewFederationSenderInternalAPIHTTP(federationSenderURL string, httpClient *http.Client) (FederationSenderInternalAPI, error) {
|
||||
if httpClient == nil {
|
||||
return nil, errors.New("NewFederationSenderInternalAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpFederationSenderInternalAPI{federationSenderURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpFederationSenderInternalAPI struct {
|
||||
federationSenderURL string
|
||||
httpClient *http.Client
|
||||
}
|
56
federationsender/api/perform.go
Normal file
56
federationsender/api/perform.go
Normal file
|
@ -0,0 +1,56 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
||||
const (
|
||||
// FederationSenderPerformJoinRequestPath is the HTTP path for the PerformJoinRequest API.
|
||||
FederationSenderPerformJoinRequestPath = "/api/federationsender/performJoinRequest"
|
||||
|
||||
// FederationSenderPerformLeaveRequestPath is the HTTP path for the PerformLeaveRequest API.
|
||||
FederationSenderPerformLeaveRequestPath = "/api/federationsender/performLeaveRequest"
|
||||
)
|
||||
|
||||
type PerformJoinRequest struct {
|
||||
RoomID string `json:"room_id"`
|
||||
}
|
||||
|
||||
type PerformJoinResponse struct {
|
||||
}
|
||||
|
||||
// Handle an instruction to make_join & send_join with a remote server.
|
||||
func (h *httpFederationSenderInternalAPI) PerformJoinRequest(
|
||||
ctx context.Context,
|
||||
request *PerformJoinRequest,
|
||||
response *PerformJoinResponse,
|
||||
) error {
|
||||
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformJoinRequest")
|
||||
defer span.Finish()
|
||||
|
||||
apiURL := h.federationSenderURL + FederationSenderPerformJoinRequestPath
|
||||
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
|
||||
}
|
||||
|
||||
type PerformLeaveRequest struct {
|
||||
RoomID string `json:"room_id"`
|
||||
}
|
||||
|
||||
type PerformLeaveResponse struct {
|
||||
}
|
||||
|
||||
// Handle an instruction to make_leave & send_leave with a remote server.
|
||||
func (h *httpFederationSenderInternalAPI) PerformLeaveRequest(
|
||||
ctx context.Context,
|
||||
request *PerformLeaveRequest,
|
||||
response *PerformLeaveResponse,
|
||||
) error {
|
||||
span, ctx := opentracing.StartSpanFromContext(ctx, "PerformLeaveRequest")
|
||||
defer span.Finish()
|
||||
|
||||
apiURL := h.federationSenderURL + FederationSenderPerformLeaveRequestPath
|
||||
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
|
||||
}
|
|
@ -2,16 +2,20 @@ package api
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
"github.com/matrix-org/dendrite/federationsender/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
"github.com/matrix-org/dendrite/federationsender/types"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
||||
// FederationSenderQueryJoinedHostsInRoomPath is the HTTP path for the QueryJoinedHostsInRoom API.
|
||||
const FederationSenderQueryJoinedHostsInRoomPath = "/api/federationsender/queryJoinedHostsInRoom"
|
||||
|
||||
// FederationSenderQueryJoinedHostServerNamesInRoomPath is the HTTP path for the QueryJoinedHostServerNamesInRoom API.
|
||||
const FederationSenderQueryJoinedHostServerNamesInRoomPath = "/api/federationsender/queryJoinedHostServerNamesInRoom"
|
||||
|
||||
// QueryJoinedHostsInRoomRequest is a request to QueryJoinedHostsInRoom
|
||||
type QueryJoinedHostsInRoomRequest struct {
|
||||
RoomID string `json:"room_id"`
|
||||
|
@ -22,6 +26,19 @@ type QueryJoinedHostsInRoomResponse struct {
|
|||
JoinedHosts []types.JoinedHost `json:"joined_hosts"`
|
||||
}
|
||||
|
||||
// QueryJoinedHostsInRoom implements FederationSenderInternalAPI
|
||||
func (h *httpFederationSenderInternalAPI) QueryJoinedHostsInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostsInRoomRequest,
|
||||
response *QueryJoinedHostsInRoomResponse,
|
||||
) error {
|
||||
span, ctx := opentracing.StartSpanFromContext(ctx, "QueryJoinedHostsInRoom")
|
||||
defer span.Finish()
|
||||
|
||||
apiURL := h.federationSenderURL + FederationSenderQueryJoinedHostsInRoomPath
|
||||
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
|
||||
}
|
||||
|
||||
// QueryJoinedHostServerNamesRequest is a request to QueryJoinedHostServerNames
|
||||
type QueryJoinedHostServerNamesInRoomRequest struct {
|
||||
RoomID string `json:"room_id"`
|
||||
|
@ -32,61 +49,8 @@ type QueryJoinedHostServerNamesInRoomResponse struct {
|
|||
ServerNames []gomatrixserverlib.ServerName `json:"server_names"`
|
||||
}
|
||||
|
||||
// FederationSenderQueryAPI is used to query information from the federation sender.
|
||||
type FederationSenderQueryAPI interface {
|
||||
// Query the joined hosts and the membership events accounting for their participation in a room.
|
||||
// Note that if a server has multiple users in the room, it will have multiple entries in the returned slice.
|
||||
// See `QueryJoinedHostServerNamesInRoom` for a de-duplicated version.
|
||||
QueryJoinedHostsInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostsInRoomRequest,
|
||||
response *QueryJoinedHostsInRoomResponse,
|
||||
) error
|
||||
// Query the server names of the joined hosts in a room.
|
||||
// Unlike QueryJoinedHostsInRoom, this function returns a de-duplicated slice
|
||||
// containing only the server names (without information for membership events).
|
||||
QueryJoinedHostServerNamesInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostServerNamesInRoomRequest,
|
||||
response *QueryJoinedHostServerNamesInRoomResponse,
|
||||
) error
|
||||
}
|
||||
|
||||
// FederationSenderQueryJoinedHostsInRoomPath is the HTTP path for the QueryJoinedHostsInRoom API.
|
||||
const FederationSenderQueryJoinedHostsInRoomPath = "/api/federationsender/queryJoinedHostsInRoom"
|
||||
|
||||
// FederationSenderQueryJoinedHostServerNamesInRoomPath is the HTTP path for the QueryJoinedHostServerNamesInRoom API.
|
||||
const FederationSenderQueryJoinedHostServerNamesInRoomPath = "/api/federationsender/queryJoinedHostServerNamesInRoom"
|
||||
|
||||
// NewFederationSenderQueryAPIHTTP creates a FederationSenderQueryAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil an error is returned
|
||||
func NewFederationSenderQueryAPIHTTP(federationSenderURL string, httpClient *http.Client) (FederationSenderQueryAPI, error) {
|
||||
if httpClient == nil {
|
||||
return nil, errors.New("NewFederationSenderQueryAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpFederationSenderQueryAPI{federationSenderURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpFederationSenderQueryAPI struct {
|
||||
federationSenderURL string
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
// QueryJoinedHostsInRoom implements FederationSenderQueryAPI
|
||||
func (h *httpFederationSenderQueryAPI) QueryJoinedHostsInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostsInRoomRequest,
|
||||
response *QueryJoinedHostsInRoomResponse,
|
||||
) error {
|
||||
span, ctx := opentracing.StartSpanFromContext(ctx, "QueryJoinedHostsInRoom")
|
||||
defer span.Finish()
|
||||
|
||||
apiURL := h.federationSenderURL + FederationSenderQueryJoinedHostsInRoomPath
|
||||
return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
|
||||
}
|
||||
|
||||
// QueryJoinedHostServerNamesInRoom implements FederationSenderQueryAPI
|
||||
func (h *httpFederationSenderQueryAPI) QueryJoinedHostServerNamesInRoom(
|
||||
// QueryJoinedHostServerNamesInRoom implements FederationSenderInternalAPI
|
||||
func (h *httpFederationSenderInternalAPI) QueryJoinedHostServerNamesInRoom(
|
||||
ctx context.Context,
|
||||
request *QueryJoinedHostServerNamesInRoomRequest,
|
||||
response *QueryJoinedHostServerNamesInRoomResponse,
|
||||
|
|
|
@ -36,7 +36,7 @@ func SetupFederationSenderComponent(
|
|||
federation *gomatrixserverlib.FederationClient,
|
||||
rsQueryAPI roomserverAPI.RoomserverQueryAPI,
|
||||
rsInputAPI roomserverAPI.RoomserverInputAPI,
|
||||
) api.FederationSenderQueryAPI {
|
||||
) api.FederationSenderInternalAPI {
|
||||
federationSenderDB, err := storage.NewDatabase(string(base.Cfg.Database.FederationSender))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to connect to federation sender db")
|
||||
|
@ -61,7 +61,7 @@ func SetupFederationSenderComponent(
|
|||
logrus.WithError(err).Panic("failed to start typing server consumer")
|
||||
}
|
||||
|
||||
queryAPI := query.FederationSenderQueryAPI{
|
||||
queryAPI := query.FederationSenderInternalAPI{
|
||||
DB: federationSenderDB,
|
||||
}
|
||||
queryAPI.SetupHTTP(http.DefaultServeMux)
|
||||
|
|
77
federationsender/query/api.go
Normal file
77
federationsender/query/api.go
Normal file
|
@ -0,0 +1,77 @@
|
|||
package query
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/federationsender/api"
|
||||
"github.com/matrix-org/dendrite/federationsender/storage"
|
||||
rsAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/util"
|
||||
)
|
||||
|
||||
// FederationSenderInternalAPI is an implementation of api.FederationSenderInternalAPI
|
||||
type FederationSenderInternalAPI struct {
|
||||
api.FederationSenderInternalAPI
|
||||
DB storage.Database
|
||||
RoomserverInputAPI rsAPI.RoomserverInputAPI
|
||||
}
|
||||
|
||||
// SetupHTTP adds the FederationSenderInternalAPI handlers to the http.ServeMux.
|
||||
func (f *FederationSenderInternalAPI) SetupHTTP(servMux *http.ServeMux) {
|
||||
servMux.Handle(
|
||||
api.FederationSenderQueryJoinedHostsInRoomPath,
|
||||
common.MakeInternalAPI("QueryJoinedHostsInRoom", func(req *http.Request) util.JSONResponse {
|
||||
var request api.QueryJoinedHostsInRoomRequest
|
||||
var response api.QueryJoinedHostsInRoomResponse
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
if err := f.QueryJoinedHostsInRoom(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
servMux.Handle(
|
||||
api.FederationSenderQueryJoinedHostServerNamesInRoomPath,
|
||||
common.MakeInternalAPI("QueryJoinedHostServerNamesInRoom", func(req *http.Request) util.JSONResponse {
|
||||
var request api.QueryJoinedHostServerNamesInRoomRequest
|
||||
var response api.QueryJoinedHostServerNamesInRoomResponse
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
if err := f.QueryJoinedHostServerNamesInRoom(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
servMux.Handle(api.FederationSenderPerformJoinRequestPath,
|
||||
common.MakeInternalAPI("PerformJoinRequest", func(req *http.Request) util.JSONResponse {
|
||||
var request api.PerformJoinRequest
|
||||
var response api.PerformJoinResponse
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.MessageResponse(http.StatusBadRequest, err.Error())
|
||||
}
|
||||
if err := f.PerformJoinRequest(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
servMux.Handle(api.FederationSenderPerformLeaveRequestPath,
|
||||
common.MakeInternalAPI("PerformLeaveRequest", func(req *http.Request) util.JSONResponse {
|
||||
var request api.PerformLeaveRequest
|
||||
var response api.PerformLeaveResponse
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.MessageResponse(http.StatusBadRequest, err.Error())
|
||||
}
|
||||
if err := f.PerformLeaveRequest(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
}
|
25
federationsender/query/perform.go
Normal file
25
federationsender/query/perform.go
Normal file
|
@ -0,0 +1,25 @@
|
|||
package query
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/matrix-org/dendrite/federationsender/api"
|
||||
)
|
||||
|
||||
// PerformJoinRequest implements api.FederationSenderInternalAPI
|
||||
func (r *FederationSenderInternalAPI) PerformJoinRequest(
|
||||
ctx context.Context,
|
||||
request *api.PerformJoinRequest,
|
||||
response *api.PerformJoinResponse,
|
||||
) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
// PerformLeaveRequest implements api.FederationSenderInternalAPI
|
||||
func (r *FederationSenderInternalAPI) PerformLeaveRequest(
|
||||
ctx context.Context,
|
||||
request *api.PerformLeaveRequest,
|
||||
response *api.PerformLeaveResponse,
|
||||
) (err error) {
|
||||
return nil
|
||||
}
|
|
@ -2,30 +2,13 @@ package query
|
|||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/federationsender/api"
|
||||
"github.com/matrix-org/dendrite/federationsender/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
)
|
||||
|
||||
// FederationSenderQueryDatabase has the APIs needed to implement the query API.
|
||||
type FederationSenderQueryDatabase interface {
|
||||
GetJoinedHosts(
|
||||
ctx context.Context, roomID string,
|
||||
) ([]types.JoinedHost, error)
|
||||
}
|
||||
|
||||
// FederationSenderQueryAPI is an implementation of api.FederationSenderQueryAPI
|
||||
type FederationSenderQueryAPI struct {
|
||||
DB FederationSenderQueryDatabase
|
||||
}
|
||||
|
||||
// QueryJoinedHostsInRoom implements api.FederationSenderQueryAPI
|
||||
func (f *FederationSenderQueryAPI) QueryJoinedHostsInRoom(
|
||||
// QueryJoinedHostsInRoom implements api.FederationSenderInternalAPI
|
||||
func (f *FederationSenderInternalAPI) QueryJoinedHostsInRoom(
|
||||
ctx context.Context,
|
||||
request *api.QueryJoinedHostsInRoomRequest,
|
||||
response *api.QueryJoinedHostsInRoomResponse,
|
||||
|
@ -34,8 +17,8 @@ func (f *FederationSenderQueryAPI) QueryJoinedHostsInRoom(
|
|||
return
|
||||
}
|
||||
|
||||
// QueryJoinedHostServerNamesInRoom implements api.FederationSenderQueryAPI
|
||||
func (f *FederationSenderQueryAPI) QueryJoinedHostServerNamesInRoom(
|
||||
// QueryJoinedHostServerNamesInRoom implements api.FederationSenderInternalAPI
|
||||
func (f *FederationSenderInternalAPI) QueryJoinedHostServerNamesInRoom(
|
||||
ctx context.Context,
|
||||
request *api.QueryJoinedHostServerNamesInRoomRequest,
|
||||
response *api.QueryJoinedHostServerNamesInRoomResponse,
|
||||
|
@ -54,35 +37,3 @@ func (f *FederationSenderQueryAPI) QueryJoinedHostServerNamesInRoom(
|
|||
|
||||
return
|
||||
}
|
||||
|
||||
// SetupHTTP adds the FederationSenderQueryAPI handlers to the http.ServeMux.
|
||||
func (f *FederationSenderQueryAPI) SetupHTTP(servMux *http.ServeMux) {
|
||||
servMux.Handle(
|
||||
api.FederationSenderQueryJoinedHostsInRoomPath,
|
||||
common.MakeInternalAPI("QueryJoinedHostsInRoom", func(req *http.Request) util.JSONResponse {
|
||||
var request api.QueryJoinedHostsInRoomRequest
|
||||
var response api.QueryJoinedHostsInRoomResponse
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
if err := f.QueryJoinedHostsInRoom(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
servMux.Handle(
|
||||
api.FederationSenderQueryJoinedHostServerNamesInRoomPath,
|
||||
common.MakeInternalAPI("QueryJoinedHostServerNamesInRoom", func(req *http.Request) util.JSONResponse {
|
||||
var request api.QueryJoinedHostServerNamesInRoomRequest
|
||||
var response api.QueryJoinedHostServerNamesInRoomResponse
|
||||
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
if err := f.QueryJoinedHostServerNamesInRoom(req.Context(), &request, &response); err != nil {
|
||||
return util.ErrorResponse(err)
|
||||
}
|
||||
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
|
|
@ -21,6 +21,7 @@ import (
|
|||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
opentracing "github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
@ -106,6 +107,9 @@ type InputRoomEventsResponse struct {
|
|||
|
||||
// RoomserverInputAPI is used to write events to the room server.
|
||||
type RoomserverInputAPI interface {
|
||||
// needed to avoid chicken and egg scenario when setting up the
|
||||
// interdependencies between the roomserver and the FS input API
|
||||
SetFederationSenderAPI(fsInputAPI fsAPI.FederationSenderInternalAPI)
|
||||
InputRoomEvents(
|
||||
ctx context.Context,
|
||||
request *InputRoomEventsRequest,
|
||||
|
@ -122,12 +126,22 @@ func NewRoomserverInputAPIHTTP(roomserverURL string, httpClient *http.Client) (R
|
|||
if httpClient == nil {
|
||||
return nil, errors.New("NewRoomserverInputAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpRoomserverInputAPI{roomserverURL, httpClient}, nil
|
||||
return &httpRoomserverInputAPI{roomserverURL, httpClient, nil}, nil
|
||||
}
|
||||
|
||||
type httpRoomserverInputAPI struct {
|
||||
roomserverURL string
|
||||
httpClient *http.Client
|
||||
// The federation sender API allows us to send federation
|
||||
// requests from the new perform input requests, still TODO.
|
||||
fsInputAPI fsAPI.FederationSenderInternalAPI
|
||||
}
|
||||
|
||||
// SetFederationSenderInputAPI passes in a federation sender input API reference
|
||||
// so that we can avoid the chicken-and-egg problem of both the roomserver input API
|
||||
// and the federation sender input API being interdependent.
|
||||
func (h *httpRoomserverInputAPI) SetFederationSenderAPI(fsInputAPI fsAPI.FederationSenderInternalAPI) {
|
||||
h.fsInputAPI = fsInputAPI
|
||||
}
|
||||
|
||||
// InputRoomEvents implements RoomserverInputAPI
|
||||
|
|
|
@ -26,6 +26,8 @@ import (
|
|||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage"
|
||||
"github.com/matrix-org/util"
|
||||
|
||||
fsAPI "github.com/matrix-org/dendrite/federationsender/api"
|
||||
)
|
||||
|
||||
// RoomserverInputAPI implements api.RoomserverInputAPI
|
||||
|
@ -37,6 +39,16 @@ type RoomserverInputAPI struct {
|
|||
OutputRoomEventTopic string
|
||||
// Protects calls to processRoomEvent
|
||||
mutex sync.Mutex
|
||||
// The federation sender API allows us to send federation
|
||||
// requests from the new perform input requests, still TODO.
|
||||
fsAPI fsAPI.FederationSenderInternalAPI
|
||||
}
|
||||
|
||||
// SetFederationSenderInputAPI passes in a federation sender input API reference
|
||||
// so that we can avoid the chicken-and-egg problem of both the roomserver input API
|
||||
// and the federation sender input API being interdependent.
|
||||
func (r *RoomserverInputAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSenderInternalAPI) {
|
||||
r.fsAPI = fsAPI
|
||||
}
|
||||
|
||||
// WriteOutputEvents implements OutputRoomEventWriter
|
||||
|
|
|
@ -21,6 +21,7 @@ import (
|
|||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
asQuery "github.com/matrix-org/dendrite/appservice/query"
|
||||
|
||||
"github.com/matrix-org/dendrite/common/basecomponent"
|
||||
"github.com/matrix-org/dendrite/roomserver/alias"
|
||||
"github.com/matrix-org/dendrite/roomserver/input"
|
||||
|
|
Loading…
Reference in a new issue