Change options to allow inbound/outbound presence

This commit is contained in:
Till Faelligen 2022-04-05 15:10:19 +02:00
parent 02d417c5c6
commit 37ad7297cc
8 changed files with 65 additions and 39 deletions

View file

@ -45,7 +45,7 @@ func SetPresence(
producer *producers.SyncAPIProducer,
userID string,
) util.JSONResponse {
if cfg.Matrix.DisablePresence {
if !cfg.Matrix.Presence.EnableInbound {
return util.JSONResponse{
Code: http.StatusOK,
JSON: struct{}{},

View file

@ -91,6 +91,10 @@ func main() {
cfg.UserAPI.BCryptCost = bcrypt.MinCost
cfg.Global.JetStream.InMemory = true
cfg.ClientAPI.RegistrationSharedSecret = "complement"
cfg.Global.Presence = config.PresenceOptions{
EnableInbound: true,
EnableOutbound: true,
}
}
j, err := yaml.Marshal(cfg)

View file

@ -33,16 +33,17 @@ import (
// OutputReceiptConsumer consumes events that originate in the clientapi.
type OutputPresenceConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable string
db storage.Database
queues *queue.OutgoingQueues
ServerName gomatrixserverlib.ServerName
topic string
ctx context.Context
jetstream nats.JetStreamContext
durable string
db storage.Database
queues *queue.OutgoingQueues
ServerName gomatrixserverlib.ServerName
topic string
outboundPresenceEnabled bool
}
// NewOutputReceiptConsumer creates a new OutputReceiptConsumer. Call Start() to begin consuming typing events.
// NewOutputPresenceConsumer creates a new OutputPresenceConsumer. Call Start() to begin consuming events.
func NewOutputPresenceConsumer(
process *process.ProcessContext,
cfg *config.FederationAPI,
@ -51,25 +52,29 @@ func NewOutputPresenceConsumer(
store storage.Database,
) *OutputPresenceConsumer {
return &OutputPresenceConsumer{
ctx: process.Context(),
jetstream: js,
queues: queues,
db: store,
ServerName: cfg.Matrix.ServerName,
durable: cfg.Matrix.JetStream.Durable("FederationAPIPresenceConsumer"),
topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
ctx: process.Context(),
jetstream: js,
queues: queues,
db: store,
ServerName: cfg.Matrix.ServerName,
durable: cfg.Matrix.JetStream.Durable("FederationAPIPresenceConsumer"),
topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
outboundPresenceEnabled: cfg.Matrix.Presence.EnableOutbound,
}
}
// Start consuming from the clientapi
func (t *OutputPresenceConsumer) Start() error {
if !t.outboundPresenceEnabled {
return nil
}
return jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.topic, t.durable, t.onMessage,
nats.DeliverAll(), nats.ManualAck(), nats.HeadersOnly(),
)
}
// onMessage is called in response to a message received on the receipt
// onMessage is called in response to a message received on the presence
// events topic from the client api.
func (t *OutputPresenceConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
// only send presence events which originated from us

View file

@ -128,13 +128,14 @@ func Send(
defer inFlightTxnsPerOrigin.Delete(index)
t := txnReq{
rsAPI: rsAPI,
keys: keys,
federation: federation,
servers: servers,
keyAPI: keyAPI,
roomsMu: mu,
producer: producer,
rsAPI: rsAPI,
keys: keys,
federation: federation,
servers: servers,
keyAPI: keyAPI,
roomsMu: mu,
producer: producer,
inboundPresenceEnabled: cfg.Matrix.Presence.EnableInbound,
}
var txnEvents struct {
@ -186,13 +187,14 @@ func Send(
type txnReq struct {
gomatrixserverlib.Transaction
rsAPI api.RoomserverInternalAPI
keyAPI keyapi.KeyInternalAPI
keys gomatrixserverlib.JSONVerifier
federation txnFederationClient
roomsMu *internal.MutexByRoom
servers federationAPI.ServersInRoomProvider
producer *producers.SyncAPIProducer
rsAPI api.RoomserverInternalAPI
keyAPI keyapi.KeyInternalAPI
keys gomatrixserverlib.JSONVerifier
federation txnFederationClient
roomsMu *internal.MutexByRoom
servers federationAPI.ServersInRoomProvider
producer *producers.SyncAPIProducer
inboundPresenceEnabled bool
}
// A subset of FederationClient functionality that txn requires. Useful for testing.
@ -391,8 +393,10 @@ func (t *txnReq) processEDUs(ctx context.Context) {
logrus.WithError(err).Errorf("Failed to process signing key update")
}
case gomatrixserverlib.MPresence:
if err := t.processPresence(ctx, e); err != nil {
logrus.WithError(err).Errorf("Failed to process presence update")
if t.inboundPresenceEnabled {
if err := t.processPresence(ctx, e); err != nil {
logrus.WithError(err).Errorf("Failed to process presence update")
}
}
default:
util.GetLogger(ctx).WithField("type", e.Type).Debug("Unhandled EDU")

View file

@ -42,7 +42,7 @@ type Global struct {
DisableFederation bool `yaml:"disable_federation"`
// Disable presence. Dendrite will not handle presence events.
DisablePresence bool `yaml:"disable_presence"`
Presence PresenceOptions `yaml:"presence"`
// List of domains that the server will trust as identity servers to
// verify third-party identifiers.
@ -71,7 +71,6 @@ func (c *Global) Defaults(generate bool) {
c.PrivateKeyPath = "matrix_key.pem"
_, c.PrivateKey, _ = ed25519.GenerateKey(rand.New(rand.NewSource(0)))
c.KeyID = "ed25519:auto"
c.DisablePresence = false
}
c.KeyValidityPeriod = time.Hour * 24 * 7
@ -229,3 +228,8 @@ func (c *DNSCacheOptions) Verify(configErrs *ConfigErrors, isMonolith bool) {
checkPositive(configErrs, "cache_size", int64(c.CacheSize))
checkPositive(configErrs, "cache_lifetime", int64(c.CacheLifetime))
}
type PresenceOptions struct {
EnableInbound bool `yaml:"enable_inbound"`
EnableOutbound bool `yaml:"enable_outbound"`
}

View file

@ -42,10 +42,11 @@ type PresenceConsumer struct {
stream types.StreamProvider
notifier *notifier.Notifier
deviceAPI api.UserDeviceAPI
cfg *config.SyncAPI
}
// NewOutputTypingEventConsumer creates a new OutputTypingEventConsumer.
// Call Start() to begin consuming from the EDU server.
// NewPresenceConsumer creates a new PresenceConsumer.
// Call Start() to begin consuming events.
func NewPresenceConsumer(
process *process.ProcessContext,
cfg *config.SyncAPI,
@ -67,6 +68,7 @@ func NewPresenceConsumer(
notifier: notifier,
stream: stream,
deviceAPI: deviceAPI,
cfg: cfg,
}
}
@ -115,6 +117,9 @@ func (s *PresenceConsumer) Start() error {
if err != nil {
return err
}
if !s.cfg.Matrix.Presence.EnableInbound && !s.cfg.Matrix.Presence.EnableOutbound {
return nil
}
return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.presenceTopic, s.durable, s.onMessage,
nats.DeliverAll(), nats.ManualAck(), nats.HeadersOnly(),

View file

@ -80,6 +80,7 @@ func NewRequestPool(
producer: producer,
}
go rp.cleanLastSeen()
go rp.cleanPresence(db, time.Minute*5)
return rp
}
@ -110,7 +111,7 @@ func (rp *RequestPool) cleanPresence(db storage.Presence, cleanupTime time.Durat
// updatePresence sends presence updates to the SyncAPI and FederationAPI
func (rp *RequestPool) updatePresence(db storage.Presence, presence string, userID string) {
if rp.cfg.Matrix.DisablePresence {
if !rp.cfg.Matrix.Presence.EnableInbound {
return
}
if presence == "" {

View file

@ -106,7 +106,10 @@ func TestRequestPool_updatePresence(t *testing.T) {
JetStream: config.JetStream{
TopicPrefix: "Dendrite",
},
DisablePresence: false,
Presence: config.PresenceOptions{
EnableInbound: true,
EnableOutbound: true,
},
},
},
}