Allow sending messages offline

VERSION

@@ -1 +1 @@
-0.89.19
+0.89.20

protocol/common/message_sender.go

@@ -59,11 +59,14 @@ type MessageEvent struct {
 type MessageSender struct {
 	identity    *ecdsa.PrivateKey
 	datasync    *datasync.DataSync
+	database    *sql.DB
 	protocol    *encryption.Protocol
 	transport   *transport.Transport
 	logger      *zap.Logger
 	persistence *RawMessagesPersistence
 
+	datasyncEnabled bool
+
 	// ephemeralKeys is a map that contains the ephemeral keys of the client, used
 	// to decrypt messages
 	ephemeralKeys      map[string]*ecdsa.PrivateKey
@@ -102,8 +105,10 @@ func NewMessageSender(
 
 	p := &MessageSender{
 		identity:        identity,
+		datasyncEnabled: features.Datasync,
 		datasync:        ds,
 		protocol:        enc,
+		database:        database,
 		persistence:     NewRawMessagesPersistence(database),
 		transport:       transport,
 		logger:          logger,
@@ -838,3 +843,30 @@ func calculatePoW(payload []byte) float64 {
 	}
 	return whisperDefaultPoW
 }
+
+func (s *MessageSender) StopDatasync() {
+	s.datasync.Stop()
+}
+
+func (s *MessageSender) StartDatasync() {
+	dataSyncTransport := datasync.NewNodeTransport()
+	dataSyncNode, err := datasyncnode.NewPersistentNode(
+		s.database,
+		dataSyncTransport,
+		datasyncpeer.PublicKeyToPeerID(s.identity.PublicKey),
+		datasyncnode.BATCH,
+		datasync.CalculateSendTime,
+		s.logger,
+	)
+	if err != nil {
+		return
+	}
+	ds := datasync.New(dataSyncNode, dataSyncTransport, true, s.logger)
+
+	if s.datasyncEnabled {
+		ds.Init(s.sendDataSync, s.transport.MaxMessageSize()/4*3, s.logger)
+		ds.Start(datasync.DatasyncTicker)
+	}
+
+	s.datasync = ds
+}

protocol/messenger_mailserver.go

@@ -547,5 +547,13 @@ func (m *Messenger) RemoveFilters(filters []*transport.Filter) error {
 }
 
 func (m *Messenger) ConnectionChanged(state connection.State) {
+	if !m.connectionState.Offline && state.Offline {
+		m.sender.StopDatasync()
+	}
+
+	if m.connectionState.Offline && !state.Offline {
+		m.sender.StartDatasync()
+	}
+
 	m.connectionState = state
 }