Add some more wakeups

syncapi/consumers/eduserver_sendtodevice.go

@@ -97,13 +97,7 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
 		return err
 	}
 
-	//s.notifier.OnNewSendToDevice(
+	s.db.SendToDeviceStream().Advance(streamPos)
-	//	output.UserID,
-	//	[]string{output.DeviceID},
-	//	types.StreamingToken{SendToDevicePosition: streamPos},
-	//)
-
-	_ = streamPos
 
 	return nil
 }

syncapi/consumers/keychange.go

@@ -110,14 +110,12 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er
 	}
 	// make sure we get our own key updates too!
 	queryRes.UserIDsToCount[output.UserID] = 1
-	posUpdate := types.StreamingToken{
+	posUpdate := types.LogPosition{
-		DeviceListPosition: types.LogPosition{
+		Offset:    msg.Offset,
-			Offset:    msg.Offset,
+		Partition: msg.Partition,
-			Partition: msg.Partition,
-		},
 	}
 
-	_ = posUpdate
+	s.db.DeviceListStream().Advance(posUpdate)
 
 	//for userID := range queryRes.UserIDsToCount {
 	//	s.notifier.OnNewKeyChange(posUpdate, userID, output.UserID)

syncapi/consumers/roomserver.go

@@ -271,8 +271,8 @@ func (s *OutputRoomEventConsumer) onNewInviteEvent(
 		return nil
 	}
 
-	_ = pduPos
+	s.db.InviteStream().Advance(pduPos)
-	//s.notifier.OnNewInvite(types.StreamingToken{InvitePosition: pduPos}, *msg.Event.StateKey())
+
 	return nil
 }
 
@@ -291,8 +291,8 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent(
 	// Notify any active sync requests that the invite has been retired.
 	// Invites share the same stream counter as PDUs
 
-	_ = pduPos
+	s.db.InviteStream().Advance(pduPos)
-	//s.notifier.OnNewInvite(types.StreamingToken{InvitePosition: pduPos}, msg.TargetUserID)
+
 	return nil
 }