a

distributor/go.mod

@@ -10,8 +10,8 @@ require (
 	github.com/google/uuid v1.3.0
 	mellium.im/sasl v0.2.1
 	mellium.im/xmlstream v0.15.3-0.20210221202126-7cc1407dad4c
-	mellium.im/xmpp v0.19.0
+	mellium.im/xmpp v0.19.1-0.20210901124536-6846e6241769
-	unifiedpush.org/go/np2p_dbus v0.0.0-20210916020553-6eec6f305585
+	unifiedpush.org/go/np2p_dbus v0.0.0-20210916024230-0bcac24079b3
 )
 
 require (
@@ -35,5 +35,3 @@ require (
 )
 
 replace dev.sum7.eu/genofire/unified-push-xmpp/messages => ../messages
-
-replace mellium.im/xmpp => ../../../../mellium.im/xmpp

distributor/go.sum

@@ -741,9 +741,12 @@ mellium.im/sasl v0.2.1 h1:nspKSRg7/SyO0cRGY71OkfHab8tf9kCts6a6oTDut0w=
 mellium.im/sasl v0.2.1/go.mod h1:ROaEDLQNuf9vjKqE1SrAfnsobm2YKXT1gnN1uDp1PjQ=
 mellium.im/xmlstream v0.15.3-0.20210221202126-7cc1407dad4c h1:1RCzOXu94kvNjuCC89G+5XTP6GOdoDrLsYdGIryyc2Y=
 mellium.im/xmlstream v0.15.3-0.20210221202126-7cc1407dad4c/go.mod h1:7SUlP7f2qnMczK+Cu/OFgqaIhldMolVjo8np7xG41D0=
+mellium.im/xmpp v0.19.0/go.mod h1:zpU69WRb0YuYcVTM/GVweZQP8r48nj1Tlq8RLXUOSNE=
+mellium.im/xmpp v0.19.1-0.20210901124536-6846e6241769 h1:V//oZVgzyQF427rqMAKFHfKLA7gbeTuPbENK3S8DPTU=
+mellium.im/xmpp v0.19.1-0.20210901124536-6846e6241769/go.mod h1:zpU69WRb0YuYcVTM/GVweZQP8r48nj1Tlq8RLXUOSNE=
 nhooyr.io/websocket v1.8.7/go.mod h1:B70DZP8IakI65RVQ51MsWP/8jndNma26DVA/nFSCgW0=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
 rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
-unifiedpush.org/go/np2p_dbus v0.0.0-20210916020553-6eec6f305585 h1:KVZ5crwEMgSPdxum+abczj/5IWGhE603sPPItqoQDDY=
+unifiedpush.org/go/np2p_dbus v0.0.0-20210916024230-0bcac24079b3 h1:zGMxFWjJEnVw11fTHO32WpR2JaqZbDVRT7/IA6IWEc4=
-unifiedpush.org/go/np2p_dbus v0.0.0-20210916020553-6eec6f305585/go.mod h1:4ug2cMRBjAeOiQVmArby6gg9GH0ZYvb8WXZ+yhrgzlw=
+unifiedpush.org/go/np2p_dbus v0.0.0-20210916024230-0bcac24079b3/go.mod h1:4ug2cMRBjAeOiQVmArby6gg9GH0ZYvb8WXZ+yhrgzlw=

distributor/xmpp.go

@@ -14,6 +14,7 @@ import (
 	"mellium.im/xmlstream"
 	"mellium.im/xmpp"
 	"mellium.im/xmpp/disco"
+	"mellium.im/xmpp/history"
 	"mellium.im/xmpp/jid"
 	"mellium.im/xmpp/mux"
 	"mellium.im/xmpp/stanza"
@@ -93,15 +94,24 @@ func (s *XMPPService) Run(dbus *distributor.DBus, store *storage.Storage) error
 		}
 	}()
 	log.Debug("xmpp client is running")
+
+	go func() {
+		log.Debug("starting fetch")
+		q := history.Query{Start: time.Now().Add(-2 * time.Hour), With: s.gateway.Bare(), Limit: 100}
+		_, _ = history.Fetch(context.TODO(), q, s.session.LocalAddr().Bare(), s.session)
+		log.Debug("ending fetch")
+	}()
+
 	s.session.Serve(mux.New(
 		// disco.Handle(),
-		mux.MessageFunc("", xml.Name{Local: "subject"}, s.message),
+		mux.MessageFunc("", xml.Name{Local: "subject"}, s.HandleMessage),
+		history.Handle(history.NewHandler(s)),
 	))
 	return nil
 }
 
-// handler of incoming message - forward to DBUS
+// handler of incoming HandleMessage - forward to DBUS
-func (s *XMPPService) message(msgHead stanza.Message, t xmlstream.TokenReadEncoder) error {
+func (s *XMPPService) HandleMessage(msgHead stanza.Message, t xmlstream.TokenReadEncoder) error {
 	logger := log.WithFields(map[string]interface{}{
 		"to":   msgHead.To.String(),
 		"from": msgHead.From.String(),
@@ -266,7 +276,7 @@ func (s *XMPPService) Register(appID, appToken string) (string, string, error) {
 			return "", errStr, err
 		}
 		logger.Info("success")
-		return conn.Endpoint, "", nil
+		return conn.Settings, "", nil
 	}
 	errStr := "Unknown Error"
 	if errr := iqRegister.Register.Error; errr != nil {
@@ -290,7 +300,7 @@ func (s *XMPPService) Unregister(appToken string) {
 	logger = logger.WithFields(map[string]interface{}{
 		"appID":       conn.AppID,
 		"publicToken": conn.PublicToken,
-		"endpoint":    conn.Endpoint,
+		"endpoint":    conn.Settings,
 	})
 	if err = s.dbus.NewConnector(conn.AppID).Unregistered(conn.AppToken); err != nil {
 		logger.WithField("error", err).Error("send unregister per dbus ")