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

Export Logs in StreamingToken

Browse source
This commit is contained in:
Neil Alexander 2020-12-10 18:39:28 +00:00
parent e524b10fb9
commit a390479909
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
3 changed files with 28 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
syncapi/consumers/keychange.go
View file

@ -114,12 +114,14 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er
return err return err
} }
// TODO: f.e queryRes.UserIDsToCount : notify users by waking up streams // TODO: f.e queryRes.UserIDsToCount : notify users by waking up streams
posUpdate := types.NewStreamToken(0, 0, 0, 0, map[string]*types.LogPosition{ posUpdate := types.StreamingToken{
Logs: map[string]*types.LogPosition{
syncinternal.DeviceListLogName: { syncinternal.DeviceListLogName: {
Offset: msg.Offset, Offset: msg.Offset,
Partition: msg.Partition, Partition: msg.Partition,
}, },
}) },
}
for userID := range queryRes.UserIDsToCount { for userID := range queryRes.UserIDsToCount {
s.notifier.OnNewKeyChange(posUpdate, userID, output.UserID) s.notifier.OnNewKeyChange(posUpdate, userID, output.UserID)
} }

8
syncapi/internal/keychange_test.go
View file

@ -16,13 +16,15 @@ import (
var ( var (
syncingUser = "@alice:localhost" syncingUser = "@alice:localhost"
emptyToken = types.NewStreamToken(0, 0, 0, 0, nil) emptyToken = types.StreamingToken{}
newestToken = types.NewStreamToken(0, 0, 0, 0, map[string]*types.LogPosition{ newestToken = types.StreamingToken{
Logs: map[string]*types.LogPosition{
DeviceListLogName: { DeviceListLogName: {
Offset: sarama.OffsetNewest, Offset: sarama.OffsetNewest,
Partition: 0, Partition: 0,
}, },
}) },
}
) )
type mockKeyAPI struct{} type mockKeyAPI struct{}

26
syncapi/types/types.go
View file

@ -110,18 +110,18 @@ type StreamingToken struct {
TypingPosition StreamPosition TypingPosition StreamPosition
ReceiptPosition StreamPosition ReceiptPosition StreamPosition
SendToDevicePosition StreamPosition SendToDevicePosition StreamPosition
logs map[string]*LogPosition Logs map[string]*LogPosition
} }
func (t *StreamingToken) SetLog(name string, lp *LogPosition) { func (t *StreamingToken) SetLog(name string, lp *LogPosition) {
if t.logs == nil { if t.Logs == nil {
t.logs = make(map[string]*LogPosition) t.Logs = make(map[string]*LogPosition)
} }
t.logs[name] = lp t.Logs[name] = lp
} }
func (t *StreamingToken) Log(name string) *LogPosition { func (t *StreamingToken) Log(name string) *LogPosition {
l, ok := t.logs[name] l, ok := t.Logs[name]
if !ok { if !ok {
return nil return nil
} }
@ -135,7 +135,7 @@ func (t StreamingToken) String() string {
t.ReceiptPosition, t.SendToDevicePosition, t.ReceiptPosition, t.SendToDevicePosition,
) )
var logStrings []string var logStrings []string
for name, lp := range t.logs { for name, lp := range t.Logs {
logStr := fmt.Sprintf("%s-%d-%d", name, lp.Partition, lp.Offset) logStr := fmt.Sprintf("%s-%d-%d", name, lp.Partition, lp.Offset)
logStrings = append(logStrings, logStr) logStrings = append(logStrings, logStr)
} }
@ -156,12 +156,12 @@ func (t *StreamingToken) IsAfter(other StreamingToken) bool {
case t.SendToDevicePosition > other.SendToDevicePosition: case t.SendToDevicePosition > other.SendToDevicePosition:
return true return true
} }
for name := range t.logs { for name := range t.Logs {
otherLog := other.Log(name) otherLog := other.Log(name)
if otherLog == nil { if otherLog == nil {
continue continue
} }
if t.logs[name].IsAfter(otherLog) { if t.Logs[name].IsAfter(otherLog) {
return true return true
} }
} }
@ -188,14 +188,14 @@ func (t *StreamingToken) WithUpdates(other StreamingToken) (ret StreamingToken)
case other.SendToDevicePosition > 0: case other.SendToDevicePosition > 0:
ret.SendToDevicePosition = other.SendToDevicePosition ret.SendToDevicePosition = other.SendToDevicePosition
} }
ret.logs = make(map[string]*LogPosition) ret.Logs = make(map[string]*LogPosition)
for name := range t.logs { for name := range t.Logs {
otherLog := other.Log(name) otherLog := other.Log(name)
if otherLog == nil { if otherLog == nil {
continue continue
} }
copy := *otherLog copy := *otherLog
ret.logs[name] = &copy ret.Logs[name] = &copy
} }
return ret return ret
} }
@ -294,7 +294,7 @@ func NewStreamTokenFromString(tok string) (token StreamingToken, err error) {
TypingPosition: positions[1], TypingPosition: positions[1],
ReceiptPosition: positions[2], ReceiptPosition: positions[2],
SendToDevicePosition: positions[3], SendToDevicePosition: positions[3],
logs: make(map[string]*LogPosition), Logs: make(map[string]*LogPosition),
} }
// dl-0-1234 // dl-0-1234
// $log_name-$partition-$offset // $log_name-$partition-$offset
@ -314,7 +314,7 @@ func NewStreamTokenFromString(tok string) (token StreamingToken, err error) {
if err != nil { if err != nil {
return return
} }
token.logs[segments[0]] = &LogPosition{ token.Logs[segments[0]] = &LogPosition{
Partition: int32(partition), Partition: int32(partition),
Offset: offset, Offset: offset,
} }