processMessage updates the list of currently joined hosts in the room and then sends the event to the hosts that were joined before the event.
(ore api.OutputNewRoomEvent, rewritesState bool)
| 138 | // processMessage updates the list of currently joined hosts in the room |
| 139 | // and then sends the event to the hosts that were joined before the event. |
| 140 | func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent, rewritesState bool) error { |
| 141 | addsStateEvents, missingEventIDs := ore.NeededStateEventIDs() |
| 142 | |
| 143 | // Ask the roomserver and add in the rest of the results into the set. |
| 144 | // Finally, work out if there are any more events missing. |
| 145 | if len(missingEventIDs) > 0 { |
| 146 | eventsReq := &api.QueryEventsByIDRequest{ |
| 147 | EventIDs: missingEventIDs, |
| 148 | } |
| 149 | eventsRes := &api.QueryEventsByIDResponse{} |
| 150 | if err := s.rsAPI.QueryEventsByID(s.ctx, eventsReq, eventsRes); err != nil { |
| 151 | return fmt.Errorf("s.rsAPI.QueryEventsByID: %w", err) |
| 152 | } |
| 153 | if len(eventsRes.Events) != len(missingEventIDs) { |
| 154 | return fmt.Errorf("missing state events") |
| 155 | } |
| 156 | addsStateEvents = append(addsStateEvents, eventsRes.Events...) |
| 157 | } |
| 158 | |
| 159 | addsJoinedHosts, err := JoinedHostsFromEvents(gomatrixserverlib.UnwrapEventHeaders(addsStateEvents)) |
| 160 | if err != nil { |
| 161 | return err |
| 162 | } |
| 163 | // Update our copy of the current state. |
| 164 | // We keep a copy of the current state because the state at each event is |
| 165 | // expressed as a delta against the current state. |
| 166 | // TODO(#290): handle EventIDMismatchError and recover the current state by |
| 167 | // talking to the roomserver |
| 168 | oldJoinedHosts, err := s.db.UpdateRoom( |
| 169 | s.ctx, |
| 170 | ore.Event.RoomID(), |
| 171 | addsJoinedHosts, |
| 172 | ore.RemovesStateEventIDs, |
| 173 | rewritesState, // if we're re-writing state, nuke all joined hosts before adding |
| 174 | ) |
| 175 | if err != nil { |
| 176 | return err |
| 177 | } |
| 178 | |
| 179 | if oldJoinedHosts == nil { |
| 180 | // This means that there is nothing to update as this is a duplicate |
| 181 | // message. |
| 182 | // This can happen if dendrite crashed between reading the message and |
| 183 | // persisting the stream position. |
| 184 | return nil |
| 185 | } |
| 186 | |
| 187 | if ore.SendAsServer == api.DoNotSendToOtherServers { |
| 188 | // Ignore event that we don't need to send anywhere. |
| 189 | return nil |
| 190 | } |
| 191 | |
| 192 | // Work out which hosts were joined at the event itself. |
| 193 | joinedHostsAtEvent, err := s.joinedHostsAtEvent(ore, oldJoinedHosts) |
| 194 | if err != nil { |
| 195 | return err |
| 196 | } |
| 197 |
no test coverage detected