mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-20 13:23:22 -06:00
Code Issues Packages Projects Releases Wiki Activity

Implement wildcard

Browse source
This commit is contained in:
Neil Alexander 2020-05-29 16:59:32 +01:00
parent 43a83d375e
commit 4664f02b4d
3 changed files with 21 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
syncapi/consumers/eduserver_sendtodevice.go
View file

@ -19,6 +19,7 @@ import (
"encoding/json" "encoding/json"
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/clientapi/auth/storage/devices"
"github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/internal/config"
@ -34,6 +35,7 @@ import (
type OutputSendToDeviceEventConsumer struct { type OutputSendToDeviceEventConsumer struct {
sendToDeviceConsumer *internal.ContinualConsumer sendToDeviceConsumer *internal.ContinualConsumer
db storage.Database db storage.Database
deviceDB devices.Database
serverName gomatrixserverlib.ServerName // our server name serverName gomatrixserverlib.ServerName // our server name
notifier *sync.Notifier notifier *sync.Notifier
} }
@ -45,6 +47,7 @@ func NewOutputSendToDeviceEventConsumer(
kafkaConsumer sarama.Consumer, kafkaConsumer sarama.Consumer,
n *sync.Notifier, n *sync.Notifier,
store storage.Database, store storage.Database,
deviceDB devices.Database,
) *OutputSendToDeviceEventConsumer { ) *OutputSendToDeviceEventConsumer {
consumer := internal.ContinualConsumer{ consumer := internal.ContinualConsumer{
@ -56,6 +59,7 @@ func NewOutputSendToDeviceEventConsumer(
s := &OutputSendToDeviceEventConsumer{ s := &OutputSendToDeviceEventConsumer{
sendToDeviceConsumer: &consumer, sendToDeviceConsumer: &consumer,
db: store, db: store,
deviceDB: deviceDB,
serverName: cfg.Matrix.ServerName, serverName: cfg.Matrix.ServerName,
notifier: n, notifier: n,
} }
@ -78,7 +82,7 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
return err return err
} }
_, domain, err := gomatrixserverlib.SplitID('@', output.UserID) localpart, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
if err != nil { if err != nil {
return err return err
} }
@ -103,9 +107,22 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
return err return err
} }
devices := []string{}
if output.DeviceID == "*" {
devs, err := s.deviceDB.GetDevicesByLocalpart(context.TODO(), localpart)
if err != nil {
return err
}
for _, dev := range devs {
devices = append(devices, dev.ID)
}
} else {
devices = append(devices, output.DeviceID)
}
s.notifier.OnNewSendToDevice( s.notifier.OnNewSendToDevice(
output.UserID, output.UserID,
[]string{output.DeviceID}, // TODO: support wildcard here as per spec devices,
types.NewStreamToken(0, streamPos), types.NewStreamToken(0, streamPos),
) )

2
syncapi/syncapi.go
View file

@ -82,7 +82,7 @@ func SetupSyncAPIComponent(
} }
sendToDeviceConsumer := consumers.NewOutputSendToDeviceEventConsumer( sendToDeviceConsumer := consumers.NewOutputSendToDeviceEventConsumer(
base.Cfg, base.KafkaConsumer, notifier, syncDB, base.Cfg, base.KafkaConsumer, notifier, syncDB, deviceDB,
) )
if err = sendToDeviceConsumer.Start(); err != nil { if err = sendToDeviceConsumer.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start send-to-device consumer") logrus.WithError(err).Panicf("failed to start send-to-device consumer")

1
sytest-whitelist
View file

@ -295,3 +295,4 @@ Can send a to-device message to two users which both receive it using /sync
Can recv device messages until they are acknowledged Can recv device messages until they are acknowledged
Device messages wake up /sync Device messages wake up /sync
Device messages over federation wake up /sync Device messages over federation wake up /sync
Wildcard device messages wake up /sync