onMessage is called in response to a message received on the send-to-device events topic from the client api.
(ctx context.Context, msg *nats.Msg)
| 71 | // onMessage is called in response to a message received on the |
| 72 | // send-to-device events topic from the client api. |
| 73 | func (t *OutputSendToDeviceConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool { |
| 74 | // only send send-to-device events which originated from us |
| 75 | sender := msg.Header.Get("sender") |
| 76 | _, originServerName, err := gomatrixserverlib.SplitID('@', sender) |
| 77 | if err != nil { |
| 78 | log.WithError(err).WithField("user_id", sender).Error("Failed to extract domain from send-to-device sender") |
| 79 | return true |
| 80 | } |
| 81 | if originServerName != t.ServerName { |
| 82 | log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere") |
| 83 | return true |
| 84 | } |
| 85 | // Extract the send-to-device event from msg. |
| 86 | var ote syncTypes.OutputSendToDeviceEvent |
| 87 | if err = json.Unmarshal(msg.Data, &ote); err != nil { |
| 88 | log.WithError(err).Errorf("output log: message parse failed (expected send-to-device)") |
| 89 | return true |
| 90 | } |
| 91 | |
| 92 | _, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID) |
| 93 | if err != nil { |
| 94 | log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination") |
| 95 | return true |
| 96 | } |
| 97 | |
| 98 | // The SyncAPI is already handling sendToDevice for the local server |
| 99 | if destServerName == t.ServerName { |
| 100 | return true |
| 101 | } |
| 102 | |
| 103 | // Pack the EDU and marshal it |
| 104 | edu := &gomatrixserverlib.EDU{ |
| 105 | Type: gomatrixserverlib.MDirectToDevice, |
| 106 | Origin: string(t.ServerName), |
| 107 | } |
| 108 | tdm := gomatrixserverlib.ToDeviceMessage{ |
| 109 | Sender: ote.Sender, |
| 110 | Type: ote.Type, |
| 111 | MessageID: util.RandomString(32), |
| 112 | Messages: map[string]map[string]json.RawMessage{ |
| 113 | ote.UserID: { |
| 114 | ote.DeviceID: ote.Content, |
| 115 | }, |
| 116 | }, |
| 117 | } |
| 118 | if edu.Content, err = json.Marshal(tdm); err != nil { |
| 119 | log.WithError(err).Error("failed to marshal EDU JSON") |
| 120 | return true |
| 121 | } |
| 122 | |
| 123 | log.Debugf("Sending send-to-device message into %q destination queue", destServerName) |
| 124 | if err := t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}); err != nil { |
| 125 | log.WithError(err).Error("failed to send EDU") |
| 126 | return false |
| 127 | } |
| 128 | |
| 129 | return true |
| 130 | } |