mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-16 18:43:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Another incremental renaming refactor of relay server

Browse source
This commit is contained in:
Devon Hudson 2023-01-11 16:28:16 -07:00
parent 7b2cbc7133
commit ed71fe63ae
No known key found for this signature in database
GPG key ID: CD06B18E77F6A628
13 changed files with 75 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
federationapi/api/api.go
View file

@ -92,7 +92,7 @@ type FederationClient interface {
SendTransaction(ctx context.Context, t gomatrixserverlib.Transaction) (res gomatrixserverlib.RespSend, err error)
P2PSendTransactionToRelay(ctx context.Context, u gomatrixserverlib.UserID, t gomatrixserverlib.Transaction, forwardingServer gomatrixserverlib.ServerName) (res gomatrixserverlib.EmptyResp, err error)
P2PGetTransactionFromRelay(ctx context.Context, u gomatrixserverlib.UserID, prev gomatrixserverlib.RelayEntry, relayServer gomatrixserverlib.ServerName) (res gomatrixserverlib.RespGetRelayTxn, err error)
P2PGetTransactionFromRelay(ctx context.Context, u gomatrixserverlib.UserID, prev gomatrixserverlib.RelayEntry, relayServer gomatrixserverlib.ServerName) (res gomatrixserverlib.RespGetRelayTransaction, err error)
// Perform operations
LookupRoomAlias(ctx context.Context, origin, s gomatrixserverlib.ServerName, roomAlias string) (res gomatrixserverlib.RespDirectory, err error)

24
relayapi/api/api.go
View file

@ -32,18 +32,18 @@ type RelayInternalAPI interface {
}
type RelayServerAPI interface {
// Store async transactions for forwarding to the destination at a later time.
PerformStoreAsync(
// Store transactions for forwarding to the destination at a later time.
PerformStoreTransaction(
ctx context.Context,
request *PerformStoreAsyncRequest,
response *PerformStoreAsyncResponse,
request *PerformStoreTransactionRequest,
response *PerformStoreTransactionResponse,
) error
// Obtain the oldest stored transaction for the specified userID.
QueryAsyncTransactions(
QueryTransactions(
ctx context.Context,
request *QueryAsyncTransactionsRequest,
response *QueryAsyncTransactionsResponse,
request *QueryRelayTransactionsRequest,
response *QueryRelayTransactionsResponse,
) error
}
@ -63,21 +63,21 @@ type QueryRelayServersResponse struct {
RelayServers []gomatrixserverlib.ServerName
}
type PerformStoreAsyncRequest struct {
type PerformStoreTransactionRequest struct {
Txn gomatrixserverlib.Transaction `json:"transaction"`
UserID gomatrixserverlib.UserID `json:"user_id"`
}
type PerformStoreAsyncResponse struct {
type PerformStoreTransactionResponse struct {
}
type QueryAsyncTransactionsRequest struct {
type QueryRelayTransactionsRequest struct {
UserID gomatrixserverlib.UserID `json:"user_id"`
PreviousEntry gomatrixserverlib.RelayEntry `json:"prev_entry,omitempty"`
}
type QueryAsyncTransactionsResponse struct {
Txn gomatrixserverlib.Transaction `json:"transaction"`
type QueryRelayTransactionsResponse struct {
Transaction gomatrixserverlib.Transaction `json:"transaction"`
EntryID int64 `json:"entry_id"`
EntriesQueued bool `json:"entries_queued"`
}

16
relayapi/internal/perform.go
View file

@ -53,10 +53,10 @@ func (r *RelayInternalAPI) PerformRelayServerSync(
}
// PerformStoreAsync implements api.RelayInternalAPI
func (r *RelayInternalAPI) PerformStoreAsync(
func (r *RelayInternalAPI) PerformStoreTransaction(
ctx context.Context,
request *api.PerformStoreAsyncRequest,
response *api.PerformStoreAsyncResponse,
request *api.PerformStoreTransactionRequest,
response *api.PerformStoreTransactionResponse,
) error {
logrus.Warnf("Storing transaction for %v", request.UserID)
receipt, err := r.db.StoreAsyncTransaction(ctx, request.Txn)
@ -76,12 +76,12 @@ func (r *RelayInternalAPI) PerformStoreAsync(
}
// QueryAsyncTransactions implements api.RelayInternalAPI
func (r *RelayInternalAPI) QueryAsyncTransactions(
func (r *RelayInternalAPI) QueryTransactions(
ctx context.Context,
request *api.QueryAsyncTransactionsRequest,
response *api.QueryAsyncTransactionsResponse,
request *api.QueryRelayTransactionsRequest,
response *api.QueryRelayTransactionsResponse,
) error {
logrus.Infof("QueryAsyncTransactions for %s", request.UserID.Raw())
logrus.Infof("QueryTransactions for %s", request.UserID.Raw())
if request.PreviousEntry.EntryID >= 0 {
logrus.Infof("Cleaning previous entry (%v) from db for %s",
request.PreviousEntry.EntryID,
@ -103,7 +103,7 @@ func (r *RelayInternalAPI) QueryAsyncTransactions(
if transaction != nil && receipt != nil {
logrus.Infof("Obtained transaction (%v) for %s", transaction.TransactionID, request.UserID.Raw())
response.Txn = *transaction
response.Transaction = *transaction
response.EntryID = receipt.GetNID()
response.EntriesQueued = true
} else {

4
relayapi/internal/perform_test.go
View file

@ -35,10 +35,10 @@ type testFedClient struct {
queueDepth uint
}
func (f *testFedClient) P2PGetTransactionFromRelay(ctx context.Context, u gomatrixserverlib.UserID, prev gomatrixserverlib.RelayEntry, relayServer gomatrixserverlib.ServerName) (res gomatrixserverlib.RespGetRelayTxn, err error) {
func (f *testFedClient) P2PGetTransactionFromRelay(ctx context.Context, u gomatrixserverlib.UserID, prev gomatrixserverlib.RelayEntry, relayServer gomatrixserverlib.ServerName) (res gomatrixserverlib.RespGetRelayTransaction, err error) {
f.queryCount++
if !f.shouldFail {
res = gomatrixserverlib.RespGetRelayTxn{
res = gomatrixserverlib.RespGetRelayTransaction{
Txn: gomatrixserverlib.Transaction{},
EntryID: 0,
}

22
relayapi/inthttp/client.go
View file

@ -26,9 +26,9 @@ import (
// HTTP paths for the internal HTTP API
const (
RelayAPIPerformRelayServerSyncPath = "/relayapi/performRelayServerSync"
RelayAPIPerformStoreAsyncPath = "/relayapi/performStoreAsync"
RelayAPIQueryAsyncTransactionsPath = "/relayapi/queryAsyncTransactions"
RelayAPIPerformRelayServerSyncPath = "/relayapi/performRelayServerSync"
RelayAPIPerformStoreTransactionPath = "/relayapi/performStoreTransaction"
RelayAPIQueryTransactionsPath = "/relayapi/queryTransactions"
)
// NewRelayAPIClient creates a RelayInternalAPI implemented by talking to a HTTP POST API.
@ -61,24 +61,24 @@ func (h *httpRelayInternalAPI) PerformRelayServerSync(
)
}
func (h *httpRelayInternalAPI) PerformStoreAsync(
func (h *httpRelayInternalAPI) PerformStoreTransaction(
ctx context.Context,
request *api.PerformStoreAsyncRequest,
response *api.PerformStoreAsyncResponse,
request *api.PerformStoreTransactionRequest,
response *api.PerformStoreTransactionResponse,
) error {
return httputil.CallInternalRPCAPI(
"PerformStoreAsync", h.relayAPIURL+RelayAPIPerformStoreAsyncPath,
"PerformStoreTransaction", h.relayAPIURL+RelayAPIPerformStoreTransactionPath,
h.httpClient, ctx, request, response,
)
}
func (h *httpRelayInternalAPI) QueryAsyncTransactions(
func (h *httpRelayInternalAPI) QueryTransactions(
ctx context.Context,
request *api.QueryAsyncTransactionsRequest,
response *api.QueryAsyncTransactionsResponse,
request *api.QueryRelayTransactionsRequest,
response *api.QueryRelayTransactionsResponse,
) error {
return httputil.CallInternalRPCAPI(
"QueryAsyncTransactions", h.relayAPIURL+RelayAPIQueryAsyncTransactionsPath,
"QueryTransactions", h.relayAPIURL+RelayAPIQueryTransactionsPath,
h.httpClient, ctx, request, response,
)
}

8
relayapi/inthttp/client_test.go
View file

@ -35,7 +35,7 @@ func TestRelayAPIClientPerormSync(t *testing.T) {
func TestRelayAPIClientStore(t *testing.T) {
// Start a local HTTP server
server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
assert.Equal(t, "/api"+RelayAPIPerformStoreAsyncPath, req.URL.String())
assert.Equal(t, "/api"+RelayAPIPerformStoreTransactionPath, req.URL.String())
}))
defer server.Close()
@ -45,13 +45,13 @@ func TestRelayAPIClientStore(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
cl.PerformStoreAsync(ctx, &api.PerformStoreAsyncRequest{}, &api.PerformStoreAsyncResponse{})
cl.PerformStoreTransaction(ctx, &api.PerformStoreTransactionRequest{}, &api.PerformStoreTransactionResponse{})
}
func TestRelayAPIClientQuery(t *testing.T) {
// Start a local HTTP server
server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
assert.Equal(t, "/api"+RelayAPIQueryAsyncTransactionsPath, req.URL.String())
assert.Equal(t, "/api"+RelayAPIQueryTransactionsPath, req.URL.String())
}))
defer server.Close()
@ -61,5 +61,5 @@ func TestRelayAPIClientQuery(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
cl.QueryAsyncTransactions(ctx, &api.QueryAsyncTransactionsRequest{}, &api.QueryAsyncTransactionsResponse{})
cl.QueryTransactions(ctx, &api.QueryRelayTransactionsRequest{}, &api.QueryRelayTransactionsResponse{})
}

8
relayapi/inthttp/server.go
View file

@ -30,12 +30,12 @@ func AddRoutes(intAPI api.RelayInternalAPI, internalAPIMux *mux.Router, enableMe
)
internalAPIMux.Handle(
RelayAPIPerformStoreAsyncPath,
httputil.MakeInternalRPCAPI("RelayAPIPerformStoreAsync", enableMetrics, intAPI.PerformStoreAsync),
RelayAPIPerformStoreTransactionPath,
httputil.MakeInternalRPCAPI("RelayAPIPerformStoreTransaction", enableMetrics, intAPI.PerformStoreTransaction),
)
internalAPIMux.Handle(
RelayAPIQueryAsyncTransactionsPath,
httputil.MakeInternalRPCAPI("RelayAPIQueryAsyncTransactions", enableMetrics, intAPI.QueryAsyncTransactions),
RelayAPIQueryTransactionsPath,
httputil.MakeInternalRPCAPI("RelayAPIQueryTransactions", enableMetrics, intAPI.QueryTransactions),
)
}

14
relayapi/routing/relaytxn.go
View file

@ -24,8 +24,8 @@ import (
"github.com/sirupsen/logrus"
)
type RelayTxnResponse struct {
Txn gomatrixserverlib.Transaction `json:"transaction"`
type RelayTransactionResponse struct {
Transaction gomatrixserverlib.Transaction `json:"transaction"`
EntryID int64 `json:"entry_id,omitempty"`
EntriesQueued bool `json:"entries_queued"`
}
@ -47,15 +47,15 @@ func GetTxnFromRelay(
entryProvided = true
}
request := api.QueryAsyncTransactionsRequest{
request := api.QueryRelayTransactionsRequest{
UserID: userID,
PreviousEntry: gomatrixserverlib.RelayEntry{EntryID: -1},
}
if entryProvided {
request.PreviousEntry = previousEntry
}
var response api.QueryAsyncTransactionsResponse
err := relayAPI.QueryAsyncTransactions(
var response api.QueryRelayTransactionsResponse
err := relayAPI.QueryTransactions(
httpReq.Context(),
&request,
&response)
@ -67,8 +67,8 @@ func GetTxnFromRelay(
return util.JSONResponse{
Code: http.StatusOK,
JSON: RelayTxnResponse{
Txn: response.Txn,
JSON: RelayTransactionResponse{
Transaction: response.Transaction,
EntryID: response.EntryID,
EntriesQueued: response.EntriesQueued,
},

24
relayapi/routing/relaytxn_test.go
View file

@ -64,9 +64,9 @@ func TestGetAsyncEmptyDatabaseReturnsNothing(t *testing.T) {
response := routing.GetTxnFromRelay(httpReq, &request, relayAPI, *userID)
assert.Equal(t, http.StatusOK, response.Code)
jsonResponse := response.JSON.(routing.RelayTxnResponse)
jsonResponse := response.JSON.(routing.RelayTransactionResponse)
assert.Equal(t, false, jsonResponse.EntriesQueued)
assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Txn)
assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Transaction)
count, err := db.GetAsyncTransactionCount(context.Background(), *userID)
assert.NoError(t, err)
@ -105,18 +105,18 @@ func TestGetAsyncReturnsSavedTransaction(t *testing.T) {
response := routing.GetTxnFromRelay(httpReq, &request, relayAPI, *userID)
assert.Equal(t, http.StatusOK, response.Code)
jsonResponse := response.JSON.(routing.RelayTxnResponse)
jsonResponse := response.JSON.(routing.RelayTransactionResponse)
assert.True(t, jsonResponse.EntriesQueued)
assert.Equal(t, transaction, jsonResponse.Txn)
assert.Equal(t, transaction, jsonResponse.Transaction)
// And once more to clear the queue
request = createAsyncQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: jsonResponse.EntryID})
response = routing.GetTxnFromRelay(httpReq, &request, relayAPI, *userID)
assert.Equal(t, http.StatusOK, response.Code)
jsonResponse = response.JSON.(routing.RelayTxnResponse)
jsonResponse = response.JSON.(routing.RelayTransactionResponse)
assert.False(t, jsonResponse.EntriesQueued)
assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Txn)
assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Transaction)
count, err := db.GetAsyncTransactionCount(context.Background(), *userID)
assert.NoError(t, err)
@ -168,26 +168,26 @@ func TestGetAsyncReturnsMultipleSavedTransactions(t *testing.T) {
response := routing.GetTxnFromRelay(httpReq, &request, relayAPI, *userID)
assert.Equal(t, http.StatusOK, response.Code)
jsonResponse := response.JSON.(routing.RelayTxnResponse)
jsonResponse := response.JSON.(routing.RelayTransactionResponse)
assert.True(t, jsonResponse.EntriesQueued)
assert.Equal(t, transaction, jsonResponse.Txn)
assert.Equal(t, transaction, jsonResponse.Transaction)
request = createAsyncQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: jsonResponse.EntryID})
response = routing.GetTxnFromRelay(httpReq, &request, relayAPI, *userID)
assert.Equal(t, http.StatusOK, response.Code)
jsonResponse = response.JSON.(routing.RelayTxnResponse)
jsonResponse = response.JSON.(routing.RelayTransactionResponse)
assert.True(t, jsonResponse.EntriesQueued)
assert.Equal(t, transaction2, jsonResponse.Txn)
assert.Equal(t, transaction2, jsonResponse.Transaction)
// And once more to clear the queue
request = createAsyncQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: jsonResponse.EntryID})
response = routing.GetTxnFromRelay(httpReq, &request, relayAPI, *userID)
assert.Equal(t, http.StatusOK, response.Code)
jsonResponse = response.JSON.(routing.RelayTxnResponse)
jsonResponse = response.JSON.(routing.RelayTransactionResponse)
assert.False(t, jsonResponse.EntriesQueued)
assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Txn)
assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Transaction)
count, err := db.GetAsyncTransactionCount(context.Background(), *userID)
assert.NoError(t, err)

6
relayapi/routing/routing.go
View file

@ -30,8 +30,8 @@ import (
)
const (
SendRelayTransactionRouteName = "SendRelayTxn"
GetRelayTransactionRouteName = "GetRelayTxn"
SendRelayTransactionRouteName = "SendRelayTransaction"
GetRelayTransactionRouteName = "GetRelayTransaction"
)
// Setup registers HTTP handlers with the given ServeMux.
@ -60,7 +60,7 @@ func Setup(
JSON: jsonerror.InvalidUsername("Username was invalid"),
}
}
return SendTxnToRelay(
return SendTransactionToRelay(
httpReq, request, relayAPI, gomatrixserverlib.TransactionID(vars["txnID"]),
*userID,
)

2
relayapi/routing/routing_test.go
View file

@ -131,7 +131,7 @@ func TestHandleSendRelayBadUserID(t *testing.T) {
assert.NotEqual(t, 200, res.StatusCode)
}
func TestHandleRelayTxn(t *testing.T) {
func TestHandleRelayTransaction(t *testing.T) {
base, close := testrig.CreateBaseDendrite(t, test.DBTypeSQLite)
defer close()

10
relayapi/routing/sendrelay.go
View file

@ -25,9 +25,9 @@ import (
"github.com/sirupsen/logrus"
)
// SendTxnToRelay implements PUT /_matrix/federation/v1/relay_txn/{txnID}/{userID}
// SendTransactionToRelay implements PUT /_matrix/federation/v1/relay_txn/{txnID}/{userID}
// This endpoint can be extracted into a separate relay server service.
func SendTxnToRelay(
func SendTransactionToRelay(
httpReq *http.Request,
fedReq *gomatrixserverlib.FederationRequest,
relayAPI api.RelayInternalAPI,
@ -65,12 +65,12 @@ func SendTxnToRelay(
util.GetLogger(httpReq.Context()).Warnf("Received transaction %q from %q containing %d PDUs, %d EDUs", txnID, fedReq.Origin(), len(t.PDUs), len(t.EDUs))
req := api.PerformStoreAsyncRequest{
req := api.PerformStoreTransactionRequest{
Txn: t,
UserID: userID,
}
res := api.PerformStoreAsyncResponse{}
err := relayAPI.PerformStoreAsync(httpReq.Context(), &req, &res)
res := api.PerformStoreTransactionResponse{}
err := relayAPI.PerformStoreTransaction(httpReq.Context(), &req, &res)
if err != nil {
return util.JSONResponse{
Code: http.StatusInternalServerError,

10
relayapi/routing/sendrelay_test.go
View file

@ -75,7 +75,7 @@ func TestForwardEmptyReturnsOk(t *testing.T) {
&db, nil, nil, nil, nil, false, "",
)
response := routing.SendTxnToRelay(httpReq, &request, relayAPI, "1", *userID)
response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
assert.Equal(t, 200, response.Code)
}
@ -104,7 +104,7 @@ func TestForwardBadJSONReturnsError(t *testing.T) {
&db, nil, nil, nil, nil, false, "",
)
response := routing.SendTxnToRelay(httpReq, &request, relayAPI, "1", *userID)
response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
assert.NotEqual(t, 200, response.Code)
}
@ -138,7 +138,7 @@ func TestForwardTooManyPDUsReturnsError(t *testing.T) {
&db, nil, nil, nil, nil, false, "",
)
response := routing.SendTxnToRelay(httpReq, &request, relayAPI, "1", *userID)
response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
assert.NotEqual(t, 200, response.Code)
}
@ -172,7 +172,7 @@ func TestForwardTooManyEDUsReturnsError(t *testing.T) {
&db, nil, nil, nil, nil, false, "",
)
response := routing.SendTxnToRelay(httpReq, &request, relayAPI, "1", *userID)
response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
assert.NotEqual(t, 200, response.Code)
}
@ -195,7 +195,7 @@ func TestUniqueTransactionStoredInDatabase(t *testing.T) {
&db, nil, nil, nil, nil, false, "",
)
response := routing.SendTxnToRelay(
response := routing.SendTransactionToRelay(
httpReq, &request, relayAPI, txn.TransactionID, *userID)
transaction, _, err := db.GetAsyncTransaction(context.TODO(), *userID)
assert.NoError(t, err, "Failed retrieving transaction")