forked from MirrorHub/mautrix-whatsapp
Compress portal event channels into one
Use a single channel of pointers to pre-allocate less memory per portal.
This commit is contained in:
parent
4e6e486e19
commit
2589049f14
2 changed files with 59 additions and 35 deletions
42
portal.go
42
portal.go
|
@ -110,7 +110,13 @@ func (portal *Portal) MarkEncrypted() {
|
|||
|
||||
func (portal *Portal) ReceiveMatrixEvent(user bridge.User, evt *event.Event) {
|
||||
if user.GetPermissionLevel() >= bridgeconfig.PermissionLevelUser || portal.HasRelaybot() {
|
||||
portal.matrixMessages <- PortalMatrixMessage{user: user.(*User), evt: evt, receivedAt: time.Now()}
|
||||
portal.events <- &PortalEvent{
|
||||
MatrixMessage: &PortalMatrixMessage{
|
||||
user: user.(*User),
|
||||
evt: evt,
|
||||
receivedAt: time.Now(),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -199,9 +205,7 @@ func (br *WABridge) newBlankPortal(key database.PortalKey) *Portal {
|
|||
log: br.Log.Sub(fmt.Sprintf("Portal/%s", key)),
|
||||
zlog: br.ZLog.With().Str("portal_key", key.String()).Logger(),
|
||||
|
||||
messages: make(chan PortalMessage, br.Config.Bridge.PortalMessageBuffer),
|
||||
matrixMessages: make(chan PortalMatrixMessage, br.Config.Bridge.PortalMessageBuffer),
|
||||
mediaRetries: make(chan PortalMediaRetry, br.Config.Bridge.PortalMessageBuffer),
|
||||
events: make(chan *PortalEvent, br.Config.Bridge.PortalMessageBuffer),
|
||||
|
||||
mediaErrorCache: make(map[types.MessageID]*FailedMediaMeta),
|
||||
}
|
||||
|
@ -232,6 +236,12 @@ type fakeMessage struct {
|
|||
Important bool
|
||||
}
|
||||
|
||||
type PortalEvent struct {
|
||||
Message *PortalMessage
|
||||
MatrixMessage *PortalMatrixMessage
|
||||
MediaRetry *PortalMediaRetry
|
||||
}
|
||||
|
||||
type PortalMessage struct {
|
||||
evt *events.Message
|
||||
undecryptable *events.UndecryptableMessage
|
||||
|
@ -279,9 +289,7 @@ type Portal struct {
|
|||
currentlyTyping []id.UserID
|
||||
currentlyTypingLock sync.Mutex
|
||||
|
||||
messages chan PortalMessage
|
||||
matrixMessages chan PortalMatrixMessage
|
||||
mediaRetries chan PortalMediaRetry
|
||||
events chan *PortalEvent
|
||||
|
||||
mediaErrorCache map[types.MessageID]*FailedMediaMeta
|
||||
|
||||
|
@ -337,7 +345,7 @@ var (
|
|||
_ bridge.TypingPortal = (*Portal)(nil)
|
||||
)
|
||||
|
||||
func (portal *Portal) handleWhatsAppMessageLoopItem(msg PortalMessage) {
|
||||
func (portal *Portal) handleWhatsAppMessageLoopItem(msg *PortalMessage) {
|
||||
if len(portal.MXID) == 0 {
|
||||
if msg.fake == nil && msg.undecryptable == nil && (msg.evt == nil || !containsSupportedMessage(msg.evt.Message)) {
|
||||
portal.log.Debugln("Not creating portal room for incoming message: message is not a chat message")
|
||||
|
@ -369,7 +377,7 @@ func (portal *Portal) handleWhatsAppMessageLoopItem(msg PortalMessage) {
|
|||
}
|
||||
}
|
||||
|
||||
func (portal *Portal) handleMatrixMessageLoopItem(msg PortalMatrixMessage) {
|
||||
func (portal *Portal) handleMatrixMessageLoopItem(msg *PortalMatrixMessage) {
|
||||
portal.latestEventBackfillLock.Lock()
|
||||
defer portal.latestEventBackfillLock.Unlock()
|
||||
evtTS := time.UnixMilli(msg.evt.Timestamp)
|
||||
|
@ -483,12 +491,16 @@ func (portal *Portal) handleOneMessageLoopItem() {
|
|||
}
|
||||
}()
|
||||
select {
|
||||
case msg := <-portal.messages:
|
||||
portal.handleWhatsAppMessageLoopItem(msg)
|
||||
case msg := <-portal.matrixMessages:
|
||||
portal.handleMatrixMessageLoopItem(msg)
|
||||
case retry := <-portal.mediaRetries:
|
||||
portal.handleMediaRetry(retry.evt, retry.source)
|
||||
case msg := <-portal.events:
|
||||
if msg.Message != nil {
|
||||
portal.handleWhatsAppMessageLoopItem(msg.Message)
|
||||
} else if msg.MatrixMessage != nil {
|
||||
portal.handleMatrixMessageLoopItem(msg.MatrixMessage)
|
||||
} else if msg.MediaRetry != nil {
|
||||
portal.handleMediaRetry(msg.MediaRetry.evt, msg.MediaRetry.source)
|
||||
} else {
|
||||
portal.log.Warn("Portal event loop returned an event without any data")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
52
user.go
52
user.go
|
@ -635,15 +635,17 @@ func (user *User) handleCallStart(sender types.JID, id, callType string, ts time
|
|||
if callType != "" {
|
||||
text = fmt.Sprintf("Incoming %s call. Use the WhatsApp app to answer.", callType)
|
||||
}
|
||||
portal.messages <- PortalMessage{
|
||||
fake: &fakeMessage{
|
||||
Sender: sender,
|
||||
Text: text,
|
||||
ID: id,
|
||||
Time: ts,
|
||||
Important: true,
|
||||
portal.events <- &PortalEvent{
|
||||
Message: &PortalMessage{
|
||||
fake: &fakeMessage{
|
||||
Sender: sender,
|
||||
Text: text,
|
||||
ID: id,
|
||||
Time: ts,
|
||||
Important: true,
|
||||
},
|
||||
source: user,
|
||||
},
|
||||
source: user,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -865,11 +867,15 @@ func (user *User) HandleEvent(event interface{}) {
|
|||
go user.handleChatPresence(v)
|
||||
case *events.Message:
|
||||
portal := user.GetPortalByMessageSource(v.Info.MessageSource)
|
||||
portal.messages <- PortalMessage{evt: v, source: user}
|
||||
portal.events <- &PortalEvent{
|
||||
Message: &PortalMessage{evt: v, source: user},
|
||||
}
|
||||
case *events.MediaRetry:
|
||||
user.phoneSeen(v.Timestamp)
|
||||
portal := user.GetPortalByJID(v.ChatID)
|
||||
portal.mediaRetries <- PortalMediaRetry{evt: v, source: user}
|
||||
portal.events <- &PortalEvent{
|
||||
MediaRetry: &PortalMediaRetry{evt: v, source: user},
|
||||
}
|
||||
case *events.CallOffer:
|
||||
user.handleCallStart(v.CallCreator, v.CallID, "", v.Timestamp)
|
||||
case *events.CallOfferNotice:
|
||||
|
@ -885,22 +891,26 @@ func (user *User) HandleEvent(event interface{}) {
|
|||
if v.Implicit {
|
||||
text = fmt.Sprintf("Your security code with %s (device #%d) changed.", puppet.Displayname, v.JID.Device)
|
||||
}
|
||||
portal.messages <- PortalMessage{
|
||||
fake: &fakeMessage{
|
||||
Sender: v.JID,
|
||||
Text: text,
|
||||
ID: strconv.FormatInt(v.Timestamp.Unix(), 10),
|
||||
Time: v.Timestamp,
|
||||
Important: false,
|
||||
portal.events <- &PortalEvent{
|
||||
Message: &PortalMessage{
|
||||
fake: &fakeMessage{
|
||||
Sender: v.JID,
|
||||
Text: text,
|
||||
ID: strconv.FormatInt(v.Timestamp.Unix(), 10),
|
||||
Time: v.Timestamp,
|
||||
Important: false,
|
||||
},
|
||||
source: user,
|
||||
},
|
||||
source: user,
|
||||
}
|
||||
}
|
||||
case *events.CallTerminate, *events.CallRelayLatency, *events.CallAccept, *events.UnknownCallEvent:
|
||||
// ignore
|
||||
case *events.UndecryptableMessage:
|
||||
portal := user.GetPortalByMessageSource(v.Info.MessageSource)
|
||||
portal.messages <- PortalMessage{undecryptable: v, source: user}
|
||||
portal.events <- &PortalEvent{
|
||||
Message: &PortalMessage{undecryptable: v, source: user},
|
||||
}
|
||||
case *events.HistorySync:
|
||||
if user.bridge.Config.Bridge.HistorySync.Backfill {
|
||||
user.historySyncs <- v
|
||||
|
@ -1235,7 +1245,9 @@ func (user *User) handleReceipt(receipt *events.Receipt) {
|
|||
if portal == nil || len(portal.MXID) == 0 {
|
||||
return
|
||||
}
|
||||
portal.messages <- PortalMessage{receipt: receipt, source: user}
|
||||
portal.events <- &PortalEvent{
|
||||
Message: &PortalMessage{receipt: receipt, source: user},
|
||||
}
|
||||
}
|
||||
|
||||
func (user *User) makeReadMarkerContent(eventID id.EventID, doublePuppet bool) CustomReadMarkers {
|
||||
|
|
Loading…
Reference in a new issue