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

More comments

Browse source
This commit is contained in:
Mark Haines 2017-02-07 14:13:25 +00:00
parent 15e35062b9
commit 1d5b573a3f
1 changed files with 7 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/github.com/matrix-org/dendrite/roomserver/input/consumer.go
View file

@ -92,12 +92,19 @@ func (c *Consumer) consumePartition(pc sarama.PartitionConsumer) {
for message := range pc.Messages() { for message := range pc.Messages() {
var input api.InputRoomEvent var input api.InputRoomEvent
if err := json.Unmarshal(message.Value, &input); err != nil { if err := json.Unmarshal(message.Value, &input); err != nil {
// If the message is invalid then log it and move onto the next message in the stream.
c.logError(message, err) c.logError(message, err)
} else { } else {
if err := processRoomEvent(c.DB, input); err != nil { if err := processRoomEvent(c.DB, input); err != nil {
// If there was an error processing the message then log it and
// move onto the next message in the stream.
// TODO: If the error was due to a problem talking to the database
// then we shouldn't move onto the next message and we should either
// retry processing the message, or panic and kill ourselves.
c.logError(message, err) c.logError(message, err)
} }
} }
// Advance our position in the stream so that we will start at the right position after a restart.
if err := c.DB.SetPartitionOffset(c.RoomEventTopic, message.Partition, message.Offset); err != nil { if err := c.DB.SetPartitionOffset(c.RoomEventTopic, message.Partition, message.Offset); err != nil {
c.logError(message, err) c.logError(message, err)
} }