nolint:gocyclo
(ctx context.Context)
| 307 | |
| 308 | // nolint:gocyclo |
| 309 | func (t *txnReq) processEDUs(ctx context.Context) { |
| 310 | for _, e := range t.EDUs { |
| 311 | eduCountTotal.Inc() |
| 312 | switch e.Type { |
| 313 | case gomatrixserverlib.MTyping: |
| 314 | // https://matrix.org/docs/spec/server_server/latest#typing-notifications |
| 315 | var typingPayload struct { |
| 316 | RoomID string `json:"room_id"` |
| 317 | UserID string `json:"user_id"` |
| 318 | Typing bool `json:"typing"` |
| 319 | } |
| 320 | if err := json.Unmarshal(e.Content, &typingPayload); err != nil { |
| 321 | util.GetLogger(ctx).WithError(err).Debug("Failed to unmarshal typing event") |
| 322 | continue |
| 323 | } |
| 324 | if _, serverName, err := gomatrixserverlib.SplitID('@', typingPayload.UserID); err != nil { |
| 325 | continue |
| 326 | } else if serverName == t.ourServerName { |
| 327 | continue |
| 328 | } else if serverName != t.Origin { |
| 329 | continue |
| 330 | } |
| 331 | if err := t.producer.SendTyping(ctx, typingPayload.UserID, typingPayload.RoomID, typingPayload.Typing, 30*1000); err != nil { |
| 332 | util.GetLogger(ctx).WithError(err).Error("Failed to send typing event to JetStream") |
| 333 | } |
| 334 | case gomatrixserverlib.MDirectToDevice: |
| 335 | // https://matrix.org/docs/spec/server_server/r0.1.3#m-direct-to-device-schema |
| 336 | var directPayload gomatrixserverlib.ToDeviceMessage |
| 337 | if err := json.Unmarshal(e.Content, &directPayload); err != nil { |
| 338 | util.GetLogger(ctx).WithError(err).Debug("Failed to unmarshal send-to-device events") |
| 339 | continue |
| 340 | } |
| 341 | if _, serverName, err := gomatrixserverlib.SplitID('@', directPayload.Sender); err != nil { |
| 342 | continue |
| 343 | } else if serverName == t.ourServerName { |
| 344 | continue |
| 345 | } else if serverName != t.Origin { |
| 346 | continue |
| 347 | } |
| 348 | for userID, byUser := range directPayload.Messages { |
| 349 | for deviceID, message := range byUser { |
| 350 | // TODO: check that the user and the device actually exist here |
| 351 | if err := t.producer.SendToDevice(ctx, directPayload.Sender, userID, deviceID, directPayload.Type, message); err != nil { |
| 352 | util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{ |
| 353 | "sender": directPayload.Sender, |
| 354 | "user_id": userID, |
| 355 | "device_id": deviceID, |
| 356 | }).Error("Failed to send send-to-device event to JetStream") |
| 357 | } |
| 358 | } |
| 359 | } |
| 360 | case gomatrixserverlib.MDeviceListUpdate: |
| 361 | t.processDeviceListUpdate(ctx, e) |
| 362 | case gomatrixserverlib.MReceipt: |
| 363 | // https://matrix.org/docs/spec/server_server/r0.1.4#receipts |
| 364 | payload := map[string]types.FederationReceiptMRead{} |
| 365 | |
| 366 | if err := json.Unmarshal(e.Content, &payload); err != nil { |
no test coverage detected