transaction ID -> chan util.JSONResponse Send implements /_matrix/federation/v1/send/{txnID}
( httpReq *http.Request, request *gomatrixserverlib.FederationRequest, txnID gomatrixserverlib.TransactionID, cfg *config.FederationAPI, rsAPI api.FederationRoomserverAPI, keyAPI keyapi.FederationKeyAPI, keys gomatrixserverlib.JSONVerifier, federation federationAPI.FederationClient, mu *internal.MutexByRoom, servers federationAPI.ServersInRoomProvider, producer *producers.SyncAPIProducer, )
| 78 | |
| 79 | // Send implements /_matrix/federation/v1/send/{txnID} |
| 80 | func Send( |
| 81 | httpReq *http.Request, |
| 82 | request *gomatrixserverlib.FederationRequest, |
| 83 | txnID gomatrixserverlib.TransactionID, |
| 84 | cfg *config.FederationAPI, |
| 85 | rsAPI api.FederationRoomserverAPI, |
| 86 | keyAPI keyapi.FederationKeyAPI, |
| 87 | keys gomatrixserverlib.JSONVerifier, |
| 88 | federation federationAPI.FederationClient, |
| 89 | mu *internal.MutexByRoom, |
| 90 | servers federationAPI.ServersInRoomProvider, |
| 91 | producer *producers.SyncAPIProducer, |
| 92 | ) util.JSONResponse { |
| 93 | // First we should check if this origin has already submitted this |
| 94 | // txn ID to us. If they have and the txnIDs map contains an entry, |
| 95 | // the transaction is still being worked on. The new client can wait |
| 96 | // for it to complete rather than creating more work. |
| 97 | index := string(request.Origin()) + "\000" + string(txnID) |
| 98 | v, ok := inFlightTxnsPerOrigin.LoadOrStore(index, make(chan util.JSONResponse, 1)) |
| 99 | ch := v.(chan util.JSONResponse) |
| 100 | if ok { |
| 101 | // This origin already submitted this txn ID to us, and the work |
| 102 | // is still taking place, so we'll just wait for it to finish. |
| 103 | ctx, cancel := context.WithTimeout(httpReq.Context(), time.Minute*5) |
| 104 | defer cancel() |
| 105 | select { |
| 106 | case <-ctx.Done(): |
| 107 | // If the caller gives up then return straight away. We don't |
| 108 | // want to attempt to process what they sent us any further. |
| 109 | return util.JSONResponse{Code: http.StatusRequestTimeout} |
| 110 | case res := <-ch: |
| 111 | // The original task just finished processing so let's return |
| 112 | // the result of it. |
| 113 | if res.Code == 0 { |
| 114 | return util.JSONResponse{Code: http.StatusAccepted} |
| 115 | } |
| 116 | return res |
| 117 | } |
| 118 | } |
| 119 | // Otherwise, store that we're currently working on this txn from |
| 120 | // this origin. When we're done processing, close the channel. |
| 121 | defer close(ch) |
| 122 | defer inFlightTxnsPerOrigin.Delete(index) |
| 123 | |
| 124 | t := txnReq{ |
| 125 | rsAPI: rsAPI, |
| 126 | keys: keys, |
| 127 | ourServerName: cfg.Matrix.ServerName, |
| 128 | federation: federation, |
| 129 | servers: servers, |
| 130 | keyAPI: keyAPI, |
| 131 | roomsMu: mu, |
| 132 | producer: producer, |
| 133 | inboundPresenceEnabled: cfg.Matrix.Presence.EnableInbound, |
| 134 | } |
| 135 | |
| 136 | var txnEvents struct { |
| 137 | PDUs []json.RawMessage `json:"pdus"` |
no test coverage detected