mirror of
https://github.com/1f349/dendrite.git
synced 2024-11-12 23:01:40 +00:00
Use default transport for AS traffic (#1789)
* Use default transport for AS traffic * Update gmsl and use default client * Remove replace * Fix go.sum * Update gomatrixserverlib * Go back to appservices managing their own HTTP clients because argh * Add missing context
This commit is contained in:
parent
fe021d3742
commit
6aa262ead8
@ -16,7 +16,10 @@ package appservice
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"crypto/tls"
|
||||||
|
"net/http"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/gorilla/mux"
|
"github.com/gorilla/mux"
|
||||||
appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
|
appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
|
||||||
@ -46,7 +49,15 @@ func NewInternalAPI(
|
|||||||
userAPI userapi.UserInternalAPI,
|
userAPI userapi.UserInternalAPI,
|
||||||
rsAPI roomserverAPI.RoomserverInternalAPI,
|
rsAPI roomserverAPI.RoomserverInternalAPI,
|
||||||
) appserviceAPI.AppServiceQueryAPI {
|
) appserviceAPI.AppServiceQueryAPI {
|
||||||
client := base.CreateAppserviceClient()
|
client := &http.Client{
|
||||||
|
Timeout: time.Second * 30,
|
||||||
|
Transport: &http.Transport{
|
||||||
|
DisableKeepAlives: true,
|
||||||
|
TLSClientConfig: &tls.Config{
|
||||||
|
InsecureSkipVerify: base.Cfg.AppServiceAPI.DisableTLSValidation,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
consumer, _ := kafka.SetupConsumerProducer(&base.Cfg.Global.Kafka)
|
consumer, _ := kafka.SetupConsumerProducer(&base.Cfg.Global.Kafka)
|
||||||
|
|
||||||
// Create a connection to the appservice postgres DB
|
// Create a connection to the appservice postgres DB
|
||||||
|
@ -23,7 +23,6 @@ import (
|
|||||||
|
|
||||||
"github.com/matrix-org/dendrite/appservice/api"
|
"github.com/matrix-org/dendrite/appservice/api"
|
||||||
"github.com/matrix-org/dendrite/setup/config"
|
"github.com/matrix-org/dendrite/setup/config"
|
||||||
"github.com/matrix-org/gomatrixserverlib"
|
|
||||||
opentracing "github.com/opentracing/opentracing-go"
|
opentracing "github.com/opentracing/opentracing-go"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
)
|
)
|
||||||
@ -33,7 +32,7 @@ const userIDExistsPath = "/users/"
|
|||||||
|
|
||||||
// AppServiceQueryAPI is an implementation of api.AppServiceQueryAPI
|
// AppServiceQueryAPI is an implementation of api.AppServiceQueryAPI
|
||||||
type AppServiceQueryAPI struct {
|
type AppServiceQueryAPI struct {
|
||||||
HTTPClient *gomatrixserverlib.Client
|
HTTPClient *http.Client
|
||||||
Cfg *config.Dendrite
|
Cfg *config.Dendrite
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -63,7 +62,7 @@ func (a *AppServiceQueryAPI) RoomAliasExists(
|
|||||||
}
|
}
|
||||||
req = req.WithContext(ctx)
|
req = req.WithContext(ctx)
|
||||||
|
|
||||||
resp, err := a.HTTPClient.DoHTTPRequest(ctx, req)
|
resp, err := a.HTTPClient.Do(req)
|
||||||
if resp != nil {
|
if resp != nil {
|
||||||
defer func() {
|
defer func() {
|
||||||
err = resp.Body.Close()
|
err = resp.Body.Close()
|
||||||
@ -124,7 +123,7 @@ func (a *AppServiceQueryAPI) UserIDExists(
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
resp, err := a.HTTPClient.DoHTTPRequest(ctx, req)
|
resp, err := a.HTTPClient.Do(req.WithContext(ctx))
|
||||||
if resp != nil {
|
if resp != nil {
|
||||||
defer func() {
|
defer func() {
|
||||||
err = resp.Body.Close()
|
err = resp.Body.Close()
|
||||||
|
@ -42,7 +42,7 @@ var (
|
|||||||
// size), then send that off to the AS's /transactions/{txnID} endpoint. It also
|
// size), then send that off to the AS's /transactions/{txnID} endpoint. It also
|
||||||
// handles exponentially backing off in case the AS isn't currently available.
|
// handles exponentially backing off in case the AS isn't currently available.
|
||||||
func SetupTransactionWorkers(
|
func SetupTransactionWorkers(
|
||||||
client *gomatrixserverlib.Client,
|
client *http.Client,
|
||||||
appserviceDB storage.Database,
|
appserviceDB storage.Database,
|
||||||
workerStates []types.ApplicationServiceWorkerState,
|
workerStates []types.ApplicationServiceWorkerState,
|
||||||
) error {
|
) error {
|
||||||
@ -58,7 +58,7 @@ func SetupTransactionWorkers(
|
|||||||
|
|
||||||
// worker is a goroutine that sends any queued events to the application service
|
// worker is a goroutine that sends any queued events to the application service
|
||||||
// it is given.
|
// it is given.
|
||||||
func worker(client *gomatrixserverlib.Client, db storage.Database, ws types.ApplicationServiceWorkerState) {
|
func worker(client *http.Client, db storage.Database, ws types.ApplicationServiceWorkerState) {
|
||||||
log.WithFields(log.Fields{
|
log.WithFields(log.Fields{
|
||||||
"appservice": ws.AppService.ID,
|
"appservice": ws.AppService.ID,
|
||||||
}).Info("Starting application service")
|
}).Info("Starting application service")
|
||||||
@ -200,7 +200,7 @@ func createTransaction(
|
|||||||
// send sends events to an application service. Returns an error if an OK was not
|
// send sends events to an application service. Returns an error if an OK was not
|
||||||
// received back from the application service or the request timed out.
|
// received back from the application service or the request timed out.
|
||||||
func send(
|
func send(
|
||||||
client *gomatrixserverlib.Client,
|
client *http.Client,
|
||||||
appservice config.ApplicationService,
|
appservice config.ApplicationService,
|
||||||
txnID int,
|
txnID int,
|
||||||
transaction []byte,
|
transaction []byte,
|
||||||
@ -213,7 +213,7 @@ func send(
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
resp, err := client.DoHTTPRequest(context.TODO(), req)
|
resp, err := client.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
2
go.sum
2
go.sum
@ -866,8 +866,6 @@ github.com/shurcooL/webdavfs v0.0.0-20170829043945-18c3829fa133/go.mod h1:hKmq5k
|
|||||||
github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
|
github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
|
||||||
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
|
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
|
||||||
github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
|
github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
|
||||||
github.com/sirupsen/logrus v1.7.0 h1:ShrD1U9pZB12TX0cVy0DtePoCH97K8EtX+mg7ZARUtM=
|
|
||||||
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
|
|
||||||
github.com/sirupsen/logrus v1.8.0 h1:nfhvjKcUMhBMVqbKHJlk5RPrrfYr/NMo3692g0dwfWU=
|
github.com/sirupsen/logrus v1.8.0 h1:nfhvjKcUMhBMVqbKHJlk5RPrrfYr/NMo3692g0dwfWU=
|
||||||
github.com/sirupsen/logrus v1.8.0/go.mod h1:4GuYW9TZmE769R5STWrRakJc4UqQ3+QQ95fyz7ENv1A=
|
github.com/sirupsen/logrus v1.8.0/go.mod h1:4GuYW9TZmE769R5STWrRakJc4UqQ3+QQ95fyz7ENv1A=
|
||||||
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
|
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
|
||||||
|
@ -290,22 +290,6 @@ func (b *BaseDendrite) CreateClient() *gomatrixserverlib.Client {
|
|||||||
return client
|
return client
|
||||||
}
|
}
|
||||||
|
|
||||||
// CreateAppserviceClient creates a new client for application services.
|
|
||||||
// It has a specific timeout and obeys TLS validation from the appservice
|
|
||||||
// config rather than the federation config.
|
|
||||||
func (b *BaseDendrite) CreateAppserviceClient() *gomatrixserverlib.Client {
|
|
||||||
opts := []gomatrixserverlib.ClientOption{
|
|
||||||
gomatrixserverlib.WithSkipVerify(b.Cfg.AppServiceAPI.DisableTLSValidation),
|
|
||||||
gomatrixserverlib.WithTimeout(time.Second * 60),
|
|
||||||
}
|
|
||||||
if b.Cfg.Global.DNSCache.Enabled {
|
|
||||||
opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
|
|
||||||
}
|
|
||||||
client := gomatrixserverlib.NewClient(opts...)
|
|
||||||
client.SetUserAgent(fmt.Sprintf("Dendrite/%s", internal.VersionString()))
|
|
||||||
return client
|
|
||||||
}
|
|
||||||
|
|
||||||
// CreateFederationClient creates a new federation client. Should only be called
|
// CreateFederationClient creates a new federation client. Should only be called
|
||||||
// once per component.
|
// once per component.
|
||||||
func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
|
func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
|
||||||
|
Loading…
Reference in New Issue
Block a user