mirror of
https://github.com/matrix-org/dendrite.git
synced 2026-01-09 07:03:10 -06:00
Clean up producer
This commit is contained in:
parent
dc3a9b826d
commit
254e76efdf
|
|
@ -21,7 +21,6 @@ import (
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/matrix-org/dendrite/internal/eventutil"
|
|
||||||
"github.com/matrix-org/dendrite/setup/jetstream"
|
"github.com/matrix-org/dendrite/setup/jetstream"
|
||||||
"github.com/matrix-org/dendrite/syncapi/types"
|
"github.com/matrix-org/dendrite/syncapi/types"
|
||||||
userapi "github.com/matrix-org/dendrite/userapi/api"
|
userapi "github.com/matrix-org/dendrite/userapi/api"
|
||||||
|
|
@ -42,36 +41,6 @@ type SyncAPIProducer struct {
|
||||||
UserAPI userapi.ClientUserAPI
|
UserAPI userapi.ClientUserAPI
|
||||||
}
|
}
|
||||||
|
|
||||||
// SendData sends account data to the sync API server
|
|
||||||
func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string, readMarker *eventutil.ReadMarkerJSON, ignoredUsers *types.IgnoredUsers) error {
|
|
||||||
m := &nats.Msg{
|
|
||||||
Subject: p.TopicClientData,
|
|
||||||
Header: nats.Header{},
|
|
||||||
}
|
|
||||||
m.Header.Set(jetstream.UserID, userID)
|
|
||||||
|
|
||||||
data := eventutil.AccountData{
|
|
||||||
RoomID: roomID,
|
|
||||||
Type: dataType,
|
|
||||||
ReadMarker: readMarker,
|
|
||||||
IgnoredUsers: ignoredUsers,
|
|
||||||
}
|
|
||||||
var err error
|
|
||||||
m.Data, err = json.Marshal(data)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
log.WithFields(log.Fields{
|
|
||||||
"user_id": userID,
|
|
||||||
"room_id": roomID,
|
|
||||||
"data_type": dataType,
|
|
||||||
}).Tracef("Producing to topic '%s'", p.TopicClientData)
|
|
||||||
|
|
||||||
_, err = p.JetStream.PublishMsg(m)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
func (p *SyncAPIProducer) SendReceipt(
|
func (p *SyncAPIProducer) SendReceipt(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
userID, roomID, eventID, receiptType string, timestamp gomatrixserverlib.Timestamp,
|
userID, roomID, eventID, receiptType string, timestamp gomatrixserverlib.Timestamp,
|
||||||
|
|
|
||||||
|
|
@ -178,11 +178,6 @@ func SaveReadMarker(
|
||||||
return util.ErrorResponse(err)
|
return util.ErrorResponse(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := syncProducer.SendData(device.UserID, roomID, "m.fully_read", &r, nil); err != nil {
|
|
||||||
util.GetLogger(req.Context()).WithError(err).Error("syncProducer.SendData failed")
|
|
||||||
return jsonerror.InternalServerError()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Handle the read receipt that may be included in the read marker
|
// Handle the read receipt that may be included in the read marker
|
||||||
if r.Read != "" {
|
if r.Read != "" {
|
||||||
return SetReceipt(req, syncProducer, device, roomID, "m.read", r.Read)
|
return SetReceipt(req, syncProducer, device, roomID, "m.read", r.Read)
|
||||||
|
|
|
||||||
|
|
@ -18,8 +18,6 @@ import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
"github.com/sirupsen/logrus"
|
|
||||||
|
|
||||||
"github.com/matrix-org/dendrite/clientapi/httputil"
|
"github.com/matrix-org/dendrite/clientapi/httputil"
|
||||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||||
|
|
@ -98,10 +96,6 @@ func PutTag(
|
||||||
return jsonerror.InternalServerError()
|
return jsonerror.InternalServerError()
|
||||||
}
|
}
|
||||||
|
|
||||||
if err = syncProducer.SendData(userID, roomID, "m.tag", nil, nil); err != nil {
|
|
||||||
logrus.WithError(err).Error("Failed to send m.tag account data update to syncapi")
|
|
||||||
}
|
|
||||||
|
|
||||||
return util.JSONResponse{
|
return util.JSONResponse{
|
||||||
Code: http.StatusOK,
|
Code: http.StatusOK,
|
||||||
JSON: struct{}{},
|
JSON: struct{}{},
|
||||||
|
|
@ -150,11 +144,6 @@ func DeleteTag(
|
||||||
return jsonerror.InternalServerError()
|
return jsonerror.InternalServerError()
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: user API should do this since it's account data
|
|
||||||
if err := syncProducer.SendData(userID, roomID, "m.tag", nil, nil); err != nil {
|
|
||||||
logrus.WithError(err).Error("Failed to send m.tag account data update to syncapi")
|
|
||||||
}
|
|
||||||
|
|
||||||
return util.JSONResponse{
|
return util.JSONResponse{
|
||||||
Code: http.StatusOK,
|
Code: http.StatusOK,
|
||||||
JSON: struct{}{},
|
JSON: struct{}{},
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue