Add typing notifications to /sync responses - fixes #635 (#718)

This PR adds a new consumer for typing notifications in syncapi. It also brings changes to syncserver.go and some related files so EDUs can better fit in /sync responses.

Fixes #635.
Fixes #574.

1 files changed, 34 insertions, 9 deletions

diff --git a/syncapi/types/types.go b/syncapi/types/types.go
index d0b1c38a..af7ec865 100644
--- a/syncapi/types/types.go
+++ b/syncapi/types/types.go
@@ -21,12 +21,38 @@ import (
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
-// StreamPosition represents the offset in the sync stream a client is at.
-type StreamPosition int64
+// SyncPosition contains the PDU and EDU stream sync positions for a client.
+type SyncPosition struct {
+	// PDUPosition is the stream position for PDUs the client is at.
+	PDUPosition int64
+	// TypingPosition is the client's position for typing notifications.
+	TypingPosition int64
+}
 
 // String implements the Stringer interface.
-func (sp StreamPosition) String() string {
-	return strconv.FormatInt(int64(sp), 10)
+func (sp SyncPosition) String() string {
+	return strconv.FormatInt(sp.PDUPosition, 10) + "_" +
+		strconv.FormatInt(sp.TypingPosition, 10)
+}
+
+// IsAfter returns whether one SyncPosition refers to states newer than another SyncPosition.
+func (sp SyncPosition) IsAfter(other SyncPosition) bool {
+	return sp.PDUPosition > other.PDUPosition ||
+		sp.TypingPosition > other.TypingPosition
+}
+
+// WithUpdates returns a copy of the SyncPosition with updates applied from another SyncPosition.
+// If the latter SyncPosition contains a field that is not 0, it is considered an update,
+// and its value will replace the corresponding value in the SyncPosition on which WithUpdates is called.
+func (sp SyncPosition) WithUpdates(other SyncPosition) SyncPosition {
+	ret := sp
+	if other.PDUPosition != 0 {
+		ret.PDUPosition = other.PDUPosition
+	}
+	if other.TypingPosition != 0 {
+		ret.TypingPosition = other.TypingPosition
+	}
+	return ret
 }
 
 // PrevEventRef represents a reference to a previous event in a state event upgrade
@@ -53,11 +79,10 @@ type Response struct {
 }
 
 // NewResponse creates an empty response with initialised maps.
-func NewResponse(pos StreamPosition) *Response {
-	res := Response{}
-	// Make sure we send the next_batch as a string. We don't want to confuse clients by sending this
-	// as an integer even though (at the moment) it is.
-	res.NextBatch = pos.String()
+func NewResponse(pos SyncPosition) *Response {
+	res := Response{
+		NextBatch: pos.String(),
+	}
 	// Pre-initialise the maps. Synapse will return {} even if there are no rooms under a specific section,
 	// so let's do the same thing. Bonus: this means we can't get dreaded 'assignment to entry in nil map' errors.
 	res.Rooms.Join = make(map[string]JoinResponse)